feat: implement Haptika PPG and Watch V1
This commit is contained in:
Sacha
@@ -0,0 +1,224 @@
|
||||
import {readJSON, storageKeys, writeJSON} from '../ppg/storage/database';
|
||||
import type {
|
||||
WatchMeasurementFeelingV1,
|
||||
WatchMeasurementSummaryV1,
|
||||
WatchRegulationSessionV1,
|
||||
WatchSyncEnvelopeV1,
|
||||
WatchTombstoneV1,
|
||||
} from './watchTypes';
|
||||
import {
|
||||
parseWatchEnvelope,
|
||||
parseWatchMeasurement,
|
||||
parseWatchRegulationSession,
|
||||
parseWatchTombstone,
|
||||
} from './watchValidation';
|
||||
|
||||
export type WatchImportStatus = 'applied' | 'duplicate' | 'stale' | 'invalid' | 'ignored';
|
||||
|
||||
export interface WatchImportResult {
|
||||
eventId: string | null;
|
||||
status: WatchImportStatus;
|
||||
}
|
||||
|
||||
let mutationQueue: Promise<void> = Promise.resolve();
|
||||
|
||||
const enqueue = <T>(operation: () => Promise<T>): Promise<T> => {
|
||||
const pending = mutationQueue.then(operation);
|
||||
mutationQueue = pending.then(() => undefined, () => undefined);
|
||||
return pending;
|
||||
};
|
||||
|
||||
const updatedAt = (value: {updatedAt: string}): number => Date.parse(value.updatedAt);
|
||||
|
||||
async function readMeasurements(): Promise<WatchMeasurementSummaryV1[]> {
|
||||
const raw = await readJSON<unknown>(storageKeys.watchMeasurements, []);
|
||||
if (!Array.isArray(raw)) return [];
|
||||
const unique = new Map<string, WatchMeasurementSummaryV1>();
|
||||
raw.forEach(value => {
|
||||
const parsed = parseWatchMeasurement(value);
|
||||
const previous = parsed === null ? undefined : unique.get(parsed.id);
|
||||
if (parsed !== null && (previous === undefined || updatedAt(parsed) > updatedAt(previous))) {
|
||||
unique.set(parsed.id, parsed);
|
||||
}
|
||||
});
|
||||
return [...unique.values()].sort((a, b) => Date.parse(b.endedAt) - Date.parse(a.endedAt));
|
||||
}
|
||||
|
||||
async function readSessions(): Promise<WatchRegulationSessionV1[]> {
|
||||
const raw = await readJSON<unknown>(storageKeys.watchRegulationSessions, []);
|
||||
if (!Array.isArray(raw)) return [];
|
||||
const unique = new Map<string, WatchRegulationSessionV1>();
|
||||
raw.forEach(value => {
|
||||
const parsed = parseWatchRegulationSession(value);
|
||||
const previous = parsed === null ? undefined : unique.get(parsed.id);
|
||||
if (parsed !== null && (previous === undefined || updatedAt(parsed) > updatedAt(previous))) {
|
||||
unique.set(parsed.id, parsed);
|
||||
}
|
||||
});
|
||||
return [...unique.values()].sort((a, b) => Date.parse(b.startedAt) - Date.parse(a.startedAt));
|
||||
}
|
||||
|
||||
async function readTombstones(): Promise<WatchTombstoneV1[]> {
|
||||
const raw = await readJSON<unknown>(storageKeys.watchTombstones, []);
|
||||
if (!Array.isArray(raw)) return [];
|
||||
const unique = new Map<string, WatchTombstoneV1>();
|
||||
raw.forEach(value => {
|
||||
const parsed = parseWatchTombstone(value);
|
||||
const previous = parsed === null ? undefined : unique.get(`${parsed.entityKind}:${parsed.entityId}`);
|
||||
if (parsed !== null && (previous === undefined || updatedAt(parsed) >= updatedAt(previous))) {
|
||||
unique.set(`${parsed.entityKind}:${parsed.entityId}`, parsed);
|
||||
}
|
||||
});
|
||||
return [...unique.values()];
|
||||
}
|
||||
|
||||
async function processedEvents(): Promise<string[]> {
|
||||
const raw = await readJSON<unknown>(storageKeys.watchProcessedEvents, []);
|
||||
return Array.isArray(raw) ? raw.filter((item): item is string => typeof item === 'string') : [];
|
||||
}
|
||||
|
||||
async function markProcessed(eventId: string, current: readonly string[]): Promise<void> {
|
||||
await writeJSON(storageKeys.watchProcessedEvents, [eventId, ...current.filter(id => id !== eventId)].slice(0, 1_000));
|
||||
}
|
||||
|
||||
function tombstoneFor(
|
||||
tombstones: readonly WatchTombstoneV1[],
|
||||
entityKind: WatchTombstoneV1['entityKind'],
|
||||
entityId: string,
|
||||
): WatchTombstoneV1 | undefined {
|
||||
return tombstones.find(item => item.entityKind === entityKind && item.entityId === entityId);
|
||||
}
|
||||
|
||||
async function upsertMeasurement(
|
||||
candidate: WatchMeasurementSummaryV1,
|
||||
tombstones: readonly WatchTombstoneV1[],
|
||||
): Promise<WatchImportStatus> {
|
||||
const tombstone = tombstoneFor(tombstones, 'measurement', candidate.id);
|
||||
if (tombstone !== undefined && updatedAt(tombstone) >= updatedAt(candidate)) return 'stale';
|
||||
const current = await readMeasurements();
|
||||
const previous = current.find(item => item.id === candidate.id);
|
||||
if (previous !== undefined && updatedAt(previous) >= updatedAt(candidate)) return 'stale';
|
||||
await writeJSON(storageKeys.watchMeasurements, [candidate, ...current.filter(item => item.id !== candidate.id)]);
|
||||
return 'applied';
|
||||
}
|
||||
|
||||
async function upsertSession(
|
||||
candidate: WatchRegulationSessionV1,
|
||||
tombstones: readonly WatchTombstoneV1[],
|
||||
): Promise<WatchImportStatus> {
|
||||
const tombstone = tombstoneFor(tombstones, 'regulationSession', candidate.id);
|
||||
if (tombstone !== undefined && updatedAt(tombstone) >= updatedAt(candidate)) return 'stale';
|
||||
const current = await readSessions();
|
||||
const previous = current.find(item => item.id === candidate.id);
|
||||
if (previous !== undefined && updatedAt(previous) >= updatedAt(candidate)) return 'stale';
|
||||
await writeJSON(storageKeys.watchRegulationSessions, [candidate, ...current.filter(item => item.id !== candidate.id)]);
|
||||
return 'applied';
|
||||
}
|
||||
|
||||
async function applyFeeling(
|
||||
payload: unknown,
|
||||
tombstones: readonly WatchTombstoneV1[],
|
||||
): Promise<WatchImportStatus> {
|
||||
if (typeof payload !== 'object' || payload === null) return 'invalid';
|
||||
const value = payload as Record<string, unknown>;
|
||||
if (typeof value.measurementId !== 'string' || typeof value.updatedAt !== 'string' ||
|
||||
!Number.isFinite(Date.parse(value.updatedAt)) || typeof value.feeling !== 'object' || value.feeling === null) return 'invalid';
|
||||
const feeling = value.feeling as Record<string, unknown>;
|
||||
if (![1, 2, 3, 4, 5].includes(Number(feeling.score)) || typeof feeling.recordedAt !== 'string' ||
|
||||
!Number.isFinite(Date.parse(feeling.recordedAt))) return 'invalid';
|
||||
const current = await readMeasurements();
|
||||
const measurement = current.find(item => item.id === value.measurementId);
|
||||
if (measurement === undefined || tombstoneFor(tombstones, 'measurement', measurement.id) !== undefined) return 'ignored';
|
||||
if (updatedAt(measurement) > Date.parse(value.updatedAt)) return 'stale';
|
||||
const next: WatchMeasurementSummaryV1 = {
|
||||
...measurement,
|
||||
feeling: feeling as unknown as WatchMeasurementFeelingV1,
|
||||
updatedAt: value.updatedAt,
|
||||
};
|
||||
await writeJSON(storageKeys.watchMeasurements, [next, ...current.filter(item => item.id !== next.id)]);
|
||||
return 'applied';
|
||||
}
|
||||
|
||||
async function applyTombstone(candidate: WatchTombstoneV1): Promise<WatchImportStatus> {
|
||||
const tombstones = await readTombstones();
|
||||
const previous = tombstoneFor(tombstones, candidate.entityKind, candidate.entityId);
|
||||
if (previous !== undefined && updatedAt(previous) > updatedAt(candidate)) return 'stale';
|
||||
const next = [candidate, ...tombstones.filter(item => item.entityKind !== candidate.entityKind || item.entityId !== candidate.entityId)];
|
||||
await writeJSON(storageKeys.watchTombstones, next);
|
||||
if (candidate.entityKind === 'measurement') {
|
||||
const measurements = await readMeasurements();
|
||||
await writeJSON(storageKeys.watchMeasurements, measurements.filter(item => item.id !== candidate.entityId));
|
||||
} else {
|
||||
const sessions = await readSessions();
|
||||
await writeJSON(storageKeys.watchRegulationSessions, sessions.filter(item => item.id !== candidate.entityId));
|
||||
}
|
||||
return 'applied';
|
||||
}
|
||||
|
||||
export const watchRepository = {
|
||||
async getMeasurements(): Promise<WatchMeasurementSummaryV1[]> {
|
||||
await mutationQueue;
|
||||
const [items, tombstones] = await Promise.all([readMeasurements(), readTombstones()]);
|
||||
return items.filter(item => tombstoneFor(tombstones, 'measurement', item.id) === undefined);
|
||||
},
|
||||
|
||||
async getRegulationSessions(): Promise<WatchRegulationSessionV1[]> {
|
||||
await mutationQueue;
|
||||
const [items, tombstones] = await Promise.all([readSessions(), readTombstones()]);
|
||||
return items.filter(item => tombstoneFor(tombstones, 'regulationSession', item.id) === undefined);
|
||||
},
|
||||
|
||||
async deleteMeasurement(id: string, deletedAt = new Date().toISOString()): Promise<WatchTombstoneV1> {
|
||||
return enqueue(async () => {
|
||||
const tombstone: WatchTombstoneV1 = {schemaVersion: 1, entityId: id, entityKind: 'measurement', deletedAt, updatedAt: deletedAt};
|
||||
await applyTombstone(tombstone);
|
||||
return tombstone;
|
||||
});
|
||||
},
|
||||
|
||||
async clearAll(deletedAt = new Date().toISOString()): Promise<WatchTombstoneV1[]> {
|
||||
return enqueue(async () => {
|
||||
const [measurements, sessions] = await Promise.all([readMeasurements(), readSessions()]);
|
||||
const created = [
|
||||
...measurements.map(item => ({schemaVersion: 1 as const, entityId: item.id, entityKind: 'measurement' as const, deletedAt, updatedAt: deletedAt})),
|
||||
...sessions.map(item => ({schemaVersion: 1 as const, entityId: item.id, entityKind: 'regulationSession' as const, deletedAt, updatedAt: deletedAt})),
|
||||
];
|
||||
await writeJSON(storageKeys.watchTombstones, created);
|
||||
await Promise.all([writeJSON(storageKeys.watchMeasurements, []), writeJSON(storageKeys.watchRegulationSessions, [])]);
|
||||
return created;
|
||||
});
|
||||
},
|
||||
|
||||
async importEnvelope(input: WatchSyncEnvelopeV1 | unknown): Promise<WatchImportResult> {
|
||||
return enqueue(async () => {
|
||||
const envelope = parseWatchEnvelope(input);
|
||||
if (envelope === null) return {eventId: null, status: 'invalid'};
|
||||
const processed = await processedEvents();
|
||||
if (processed.includes(envelope.eventId)) return {eventId: envelope.eventId, status: 'duplicate'};
|
||||
const tombstones = await readTombstones();
|
||||
let status: WatchImportStatus;
|
||||
switch (envelope.kind) {
|
||||
case 'measurementUpsert': {
|
||||
const measurement = parseWatchMeasurement(envelope.payload);
|
||||
status = measurement === null ? 'invalid' : await upsertMeasurement(measurement, tombstones);
|
||||
break;
|
||||
}
|
||||
case 'regulationSessionUpsert': {
|
||||
const session = parseWatchRegulationSession(envelope.payload);
|
||||
status = session === null ? 'invalid' : await upsertSession(session, tombstones);
|
||||
break;
|
||||
}
|
||||
case 'feelingUpsert': status = await applyFeeling(envelope.payload, tombstones); break;
|
||||
case 'tombstone': {
|
||||
const tombstone = parseWatchTombstone(envelope.payload);
|
||||
status = tombstone === null ? 'invalid' : await applyTombstone(tombstone);
|
||||
break;
|
||||
}
|
||||
default: status = 'ignored';
|
||||
}
|
||||
// Invalid payloads are intentionally not acknowledged as processed so a fixed retry can win.
|
||||
if (status !== 'invalid') await markProcessed(envelope.eventId, processed);
|
||||
return {eventId: envelope.eventId, status};
|
||||
});
|
||||
},
|
||||
};
|
||||
Reference in New Issue
Block a user