From ad27e5b89d7ce49cdd672565be5dedb9e237d435 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Fri, 23 Oct 2026 14:45:26 +0200 Subject: [PATCH] refactor(sync): move the delta feed's domain pages out of its route --- .../baselines/source-function-lines.json | 5 +- .../baselines/version-stamp-comments.json | 5 +- .../medication-intake-tracking-guard.test.ts | 2 +- src/app/api/sync/changes/route.ts | 570 +----------------- src/lib/sync/changes/domain-pages.ts | 229 +++++++ src/lib/sync/changes/pull.ts | 218 +++++++ src/lib/sync/changes/read-rows.ts | 147 +++++ src/lib/sync/changes/wire.ts | 114 ++++ 8 files changed, 722 insertions(+), 568 deletions(-) create mode 100644 src/lib/sync/changes/domain-pages.ts create mode 100644 src/lib/sync/changes/pull.ts create mode 100644 src/lib/sync/changes/read-rows.ts create mode 100644 src/lib/sync/changes/wire.ts diff --git a/src/__tests__/baselines/source-function-lines.json b/src/__tests__/baselines/source-function-lines.json index ed5d5955f..a853ae209 100644 --- a/src/__tests__/baselines/source-function-lines.json +++ b/src/__tests__/baselines/source-function-lines.json @@ -1,7 +1,7 @@ { "$comment": "Production functions over 200 lines, keyed \"file :: name\", and their length. Maintained by src/__tests__/source-size-ratchet.test.ts.", - "count": 380, - "total": 133935, + "count": 379, + "total": 133508, "entries": { "app/achievements/page.tsx :: AchievementsPage": 206, "app/api/admin/backups/upload/route.ts :: apiHandler(arg)": 314, @@ -52,7 +52,6 @@ "app/api/nutrients/batch/route.ts :: postBatch": 304, "app/api/onboarding/complete/route.ts :: completeNeedsFlow": 218, "app/api/record-settings/[family]/route.ts :: apiHandler(arg) #2": 262, - "app/api/sync/changes/route.ts :: apiHandler(arg)": 427, "app/api/user/ai-provider/route.ts :: apiHandler(arg) #2": 248, "app/api/withings/callback/route.ts :: apiHandler(arg)": 247, "app/auth/login/page.tsx :: LoginPage": 576, diff --git a/src/__tests__/baselines/version-stamp-comments.json b/src/__tests__/baselines/version-stamp-comments.json index dfeee8420..b60f0262b 100644 --- a/src/__tests__/baselines/version-stamp-comments.json +++ b/src/__tests__/baselines/version-stamp-comments.json @@ -1,7 +1,7 @@ { "$comment": "Comment lines that open with a release version, per file under src/. Maintained by src/__tests__/version-stamp-ratchet.test.ts.", - "count": 2915, - "total": 9853, + "count": 2914, + "total": 9848, "entries": { "__tests__/account-payload-consumer-guard.test.ts": 2, "__tests__/acting-account-boundary-guard.test.ts": 7, @@ -533,7 +533,6 @@ "app/api/strava/credentials/route.ts": 1, "app/api/strava/disconnect/route.ts": 1, "app/api/strava/status/route.ts": 1, - "app/api/sync/changes/route.ts": 5, "app/api/sync/state/route.ts": 2, "app/api/telegram/webhook/__tests__/route.test.ts": 3, "app/api/user/ai-provider/__tests__/legacy-private-host.test.ts": 1, diff --git a/src/__tests__/medication-intake-tracking-guard.test.ts b/src/__tests__/medication-intake-tracking-guard.test.ts index cb69019e5..8e8b531ff 100644 --- a/src/__tests__/medication-intake-tracking-guard.test.ts +++ b/src/__tests__/medication-intake-tracking-guard.test.ts @@ -63,7 +63,7 @@ const INTAKE_MAY_SEE_EVERY_ROW: Record = { "app/api/export/route.ts": "export of the recorded history", "lib/export/medications-backup.ts": "backup of the recorded history", "app/api/sync/state/route.ts": "sync watermarks, no rate", - "app/api/sync/changes/route.ts": + "lib/sync/changes/read-rows.ts": "delta feed mirroring the recorded rows to a paired client, no rate", "lib/jobs/measurement-tombstone-cleanup.ts": "tombstone purge", "lib/ai/coach/tools/availability.ts": diff --git a/src/app/api/sync/changes/route.ts b/src/app/api/sync/changes/route.ts index 28d268df6..d714e7fff 100644 --- a/src/app/api/sync/changes/route.ts +++ b/src/app/api/sync/changes/route.ts @@ -46,16 +46,11 @@ * `apiHandler` + `requireAuth` (cookie OR Bearer; iOS uses Bearer). * Read-only — no idempotency, no write side-effect (unlike the legacy * `/api/sync/state` checkpoint bump). The cursor is owned by the client. + * The pull itself lives in `src/lib/sync/changes/`. */ import type { NextRequest } from "next/server"; import { z } from "zod/v4"; -import { prisma } from "@/lib/db"; -import { - findCompactionTombstones, - isFoldedWindowCandidate, - loadFoldTimezone, -} from "@/lib/measurements/folded-window"; import { apiHandler, requireAuth } from "@/lib/api-handler"; import { apiError, @@ -63,153 +58,17 @@ import { apiValidationError, sanitiseZodIssues, } from "@/lib/api-response"; -import { annotate } from "@/lib/logging/context"; import { checkRateLimit } from "@/lib/rate-limit"; -import { readNote } from "@/lib/crypto/note-cipher"; -import { levelAForWire } from "@/lib/mood/level-a"; -import { contextForWire, type MoodContextWire } from "@/lib/mood/context"; -import { TOMBSTONE_RETENTION_DAYS } from "@/lib/auth/native-client"; -import { - decodeCursor, - decodeCursorIssuedAt, - encodeCursor, - SYNC_DOMAINS, - type DomainWatermark, - type SyncCursor, - type SyncDomain, -} from "@/lib/sync/cursor"; -import { - toCycleDayLogDTO, - toMenstrualCycleDTO, - dayLogSymptomInclude, - type CycleDayLogDTO, - type MenstrualCycleDTO, -} from "@/lib/cycle/dto"; -import { isModuleEnabled } from "@/lib/modules/gate"; +import { pullSyncChanges } from "@/lib/sync/changes/pull"; const DEFAULT_LIMIT = 200; const MAX_LIMIT = 500; -const DAY_MS = 86_400_000; const querySchema = z.object({ cursor: z.string().min(1).max(2048).optional(), limit: z.coerce.number().int().min(1).max(MAX_LIMIT).optional(), }); -interface MeasurementUpsert { - id: string; - externalId: string | null; - type: string; - value: number; - unit: string; - measuredAt: string; - source: string; - notes: string | null; - syncVersion: number; - updatedAt: string; -} - -interface MeasurementTombstone { - id: string; - externalId: string | null; - syncVersion: number; - deletedAt: string; - updatedAt: string; -} - -interface MoodUpsert { - id: string; - date: string; - mood: string; - score: number; - /** - * v1.38 — the five level-A self-state values, each 0-10 and each nullable. - * `a2` is inverse-oriented: a higher number means more stress, and a client - * that wants "up is better" flips the sign itself rather than the server - * storing a flipped value. - * - * Server-authoritative: `a1` is derived from the five-point label on every - * write, so a client that sends only `mood` still mirrors a complete row. - */ - a1: number | null; - a2: number | null; - a3: number | null; - a4: number | null; - a5: number | null; - /** - * v1.38 — the day's context, or null when the entry carries none. Null and - * an object of nulls are different answers and the feed keeps them apart. - */ - context: MoodContextWire | null; - tags: string | null; - note: string | null; - moodLoggedAt: string; - source: string; - syncVersion: number; - updatedAt: string; -} - -interface MoodTombstone { - id: string; - syncVersion: number; - deletedAt: string; - updatedAt: string; -} - -interface IntakeUpsert { - id: string; - medicationId: string; - scheduledFor: string; - takenAt: string | null; - skipped: boolean; - source: string; - /** v1.42.2 — recorded by the server (`autoMarkTaken`), not the person. */ - autoRecorded: boolean; - syncVersion: number; - updatedAt: string; -} - -interface IntakeTombstone { - id: string; - syncVersion: number; - deletedAt: string; - updatedAt: string; -} - -// v1.15.0 — cycle domains. cycleDays key tombstones on `externalId` when -// present (HealthKit-origin rows are externalId-keyed cross-device, like -// measurements) else the server `id`; cycles key on the server `id` -// (computed rows carry no externalId). -interface CycleDayTombstone { - id: string; - externalId: string | null; - syncVersion: number; - deletedAt: string; - updatedAt: string; -} - -interface CycleTombstone { - id: string; - syncVersion: number; - deletedAt: string; - updatedAt: string; -} - -/** - * Build the `(updatedAt, id)` keyset predicate for a domain's watermark. - * Empty object (full scan) when the client has no watermark for it yet. - */ -function keysetFilter(wm: DomainWatermark | undefined) { - if (!wm) return {}; - const at = new Date(wm.updatedAtMs); - return { - OR: [ - { updatedAt: { gt: at } }, - { AND: [{ updatedAt: at }, { id: { gt: wm.id } }] }, - ], - }; -} - export const GET = apiHandler(async (request: NextRequest) => { const { user } = await requireAuth(); @@ -232,424 +91,13 @@ export const GET = apiHandler(async (request: NextRequest) => { 422, ); } - const limit = parsed.data.limit ?? DEFAULT_LIMIT; - const serverNow = new Date(); - - // ── module gate: cycle ──────────────────────────────────────────── - // The `cycleDays` / `cycles` domains carry period days and per-day - // symptom rows — the same data every `/api/cycle/*` route refuses with a - // 403 when the module is off. Without this the delta feed re-served them - // ungated to the very same token. - // - // OMIT, NOT 403 — deliberately, and unlike the whole-record FHIR export. - // This feed multiplexes five domains over one cursor: refusing the whole - // request would wedge measurement / mood / intake catch-up for every - // account with cycle tracking off, which (the gate delegates to - // `isCycleEnabled`) is the default for most accounts. A partial payload - // is the only outcome that keeps the client syncing while still honouring - // the promise that a disabled module hands out nothing. - // - // The two cycle watermarks are deliberately NOT advanced while the module - // is off (the queries are skipped, so `advanceCursor` sees an empty page - // and leaves the prior watermark untouched). Re-enabling therefore resumes - // from where the client actually left off instead of silently skipping the - // rows that changed while the module was hidden. - // - // KNOWN GAP, out of scope here: a client that synced cycle rows BEFORE the - // module was turned off keeps its stale local mirror. Emitting tombstones - // for them would be wrong — a tombstone means deleted, and disabling a - // module must leave the rows intact so re-enabling finds a complete - // history. Purging a paired client's mirror needs an explicit - // "module went dark" signal in the feed contract plus client support for - // it; that is an additive wire change and an iOS-client change, not a - // server-side gate fix, so it is left for a coordinated contract cycle. - const cycleEnabled = await isModuleEnabled(user.id, "cycle"); - - // The retention horizon: tombstones older than this may have been - // pruned by the cleanup job, so a cursor that predates it can no longer - // be trusted to deliver every deletion incrementally. - const retentionHorizon = new Date( - serverNow.getTime() - TOMBSTONE_RETENTION_DAYS * DAY_MS, + return apiSuccess( + await pullSyncChanges({ + userId: user.id, + token: parsed.data.cursor, + limit: parsed.data.limit ?? DEFAULT_LIMIT, + serverNow: new Date(), + }), ); - - // A garbage / unparseable / old-format cursor decodes to null → fresh - // initial sync (every domain scans from the start). - const cursor: SyncCursor = parsed.data.cursor - ? (decodeCursor(parsed.data.cursor) ?? {}) - : {}; - - // A stale cursor (ANY domain watermark older than retention) forces a - // clean re-init: a deletion older than retention may already be pruned. - const pastRetention = SYNC_DOMAINS.some((domain) => { - const wm = cursor[domain]; - return wm !== undefined && wm.updatedAtMs < retentionHorizon.getTime(); - }); - // A cursor issued before the account's record was last replaced (a backup - // restore, or the person deleting all their data) cannot be caught up - // either: the restore wrote rows back with their original `updatedAt`, - // behind the cursor's watermarks, and removed rows without tombstones. The - // client re-initialises instead. Compared by the time the cursor was - // issued, not by its watermarks, which a restore leaves meaningless; a - // cursor from before the issue time was recorded counts as issued before - // any reset. - let beforeReset = false; - if (parsed.data.cursor) { - const account = await prisma.user.findUnique({ - where: { id: user.id }, - select: { syncResetAt: true }, - }); - if (account?.syncResetAt) { - const issuedAt = decodeCursorIssuedAt(parsed.data.cursor); - beforeReset = - issuedAt === null || issuedAt < account.syncResetAt.getTime(); - } - } - const expired = pastRetention || beforeReset; - if (expired) { - annotate({ - action: { name: "sync.changes.pull" }, - meta: { - cursor_expired: true, - cursor_expired_reason: beforeReset ? "sync_reset" : "retention", - returned: 0, - }, - }); - return apiSuccess({ - serverNow: serverNow.toISOString(), - cursor: parsed.data.cursor ?? null, - hasMore: false, - cursorExpired: true, - changes: { - measurements: { upserts: [], tombstones: [] }, - mood: { upserts: [], tombstones: [] }, - intakes: { upserts: [], tombstones: [] }, - cycleDays: { upserts: [], tombstones: [] }, - cycles: { upserts: [], tombstones: [] }, - }, - }); - } - - // Per-domain keyset walks. Both live and soft-deleted rows are in the - // same scan — a soft-delete bumps `updatedAt`, so a tombstone is just a - // row whose `deletedAt` is non-null. Fetch limit+1 to detect `hasMore` - // without a count query. - const [measurementRows, moodRows, intakeRows, cycleDayRows, cycleRows] = - await Promise.all([ - prisma.measurement.findMany({ - where: { userId: user.id, ...keysetFilter(cursor.measurements) }, - orderBy: [{ updatedAt: "asc" }, { id: "asc" }], - take: limit + 1, - select: { - id: true, - externalId: true, - type: true, - value: true, - unit: true, - measuredAt: true, - source: true, - notes: true, - notesEncrypted: true, - syncVersion: true, - deletedAt: true, - updatedAt: true, - }, - }), - prisma.moodEntry.findMany({ - where: { userId: user.id, ...keysetFilter(cursor.mood) }, - orderBy: [{ updatedAt: "asc" }, { id: "asc" }], - take: limit + 1, - select: { - id: true, - date: true, - mood: true, - score: true, - moodA1: true, - stressA2: true, - energyA3: true, - connectionA4: true, - stabilityA5: true, - context: true, - tags: true, - note: true, - noteEncrypted: true, - moodLoggedAt: true, - source: true, - syncVersion: true, - deletedAt: true, - updatedAt: true, - }, - }), - prisma.medicationIntakeEvent.findMany({ - where: { userId: user.id, ...keysetFilter(cursor.intakes) }, - orderBy: [{ updatedAt: "asc" }, { id: "asc" }], - take: limit + 1, - select: { - id: true, - medicationId: true, - scheduledFor: true, - takenAt: true, - skipped: true, - source: true, - autoRecorded: true, - syncVersion: true, - deletedAt: true, - updatedAt: true, - }, - }), - // Skipped outright when the module is off — not filtered after the - // fact — so the disabled rows are never read out of Postgres at all. - cycleEnabled - ? prisma.cycleDayLog.findMany({ - where: { userId: user.id, ...keysetFilter(cursor.cycleDays) }, - orderBy: [{ updatedAt: "asc" }, { id: "asc" }], - take: limit + 1, - include: dayLogSymptomInclude, - }) - : [], - cycleEnabled - ? prisma.menstrualCycle.findMany({ - where: { userId: user.id, ...keysetFilter(cursor.cycles) }, - orderBy: [{ updatedAt: "asc" }, { id: "asc" }], - take: limit + 1, - }) - : [], - ]); - - const nextCursor: SyncCursor = { ...cursor }; - let hasMore = false; - - // ── measurements ────────────────────────────────────────── - const mHasMore = measurementRows.length > limit; - const mPage = mHasMore ? measurementRows.slice(0, limit) : measurementRows; - const measurementUpserts: MeasurementUpsert[] = []; - const measurementTombstones: MeasurementTombstone[] = []; - // v1.42 — compaction tombstones are not deletions anyone made. The folds - // used to soft-delete the raw Apple Health samples they folded into an - // hourly or daily `stats:` row, and those rows rode this feed as - // tombstones for 75 days. They are left out now, by the same classifier the - // backlog purge uses (`folded-window.ts`); the cursor still advances past - // them, so the keyset walk is unchanged. A person's own deletions are - // reported exactly as before. - const compaction = new Set(); - const tombstonesOnPage = mPage.filter((row) => row.deletedAt !== null); - if (tombstonesOnPage.length > 0) { - const candidates = tombstonesOnPage.map((row) => ({ - type: row.type, - source: row.source, - externalId: row.externalId, - measuredAt: row.measuredAt, - deletedAt: row.deletedAt, - })); - if ( - candidates.some((row) => isFoldedWindowCandidate(row, row.deletedAt!)) - ) { - const classA = await findCompactionTombstones( - prisma, - user.id, - await loadFoldTimezone(prisma, user.id), - candidates, - ); - for (const index of classA) compaction.add(tombstonesOnPage[index].id); - } - } - for (const row of mPage) { - if (compaction.has(row.id)) continue; - if (row.deletedAt) { - measurementTombstones.push({ - id: row.id, - externalId: row.externalId, - syncVersion: row.syncVersion, - deletedAt: row.deletedAt.toISOString(), - updatedAt: row.updatedAt.toISOString(), - }); - } else { - // Every row carries its `source`. RECOVERY_SCORE is written by BOTH the - // WHOOP-native sync (`source = WHOOP`) and the COMPUTED proxy job - // (`source = COMPUTED`) for the same day; the client must resolve to ONE - // canonical recovery series — WHOOP wins when present, COMPUTED is the - // fallback — and never chart the proxy and the native value as two - // competing series. The feed faithfully mirrors both rows (a delta-sync - // mirror cannot drop a row mid-keyset without a tombstone); the - // canonical pick is the same WHOOP-over-COMPUTED rule the server's - // wellness tile + doctor PDF apply (`resolveCanonicalRecovery`). - measurementUpserts.push({ - id: row.id, - externalId: row.externalId, - type: row.type, - value: row.value, - unit: row.unit, - measuredAt: row.measuredAt.toISOString(), - source: row.source, - notes: readNote(row.notesEncrypted, row.notes), - syncVersion: row.syncVersion, - updatedAt: row.updatedAt.toISOString(), - }); - } - } - advanceCursor(nextCursor, "measurements", mPage); - hasMore = hasMore || mHasMore; - - // ── mood ────────────────────────────────────────────────── - const moodHasMore = moodRows.length > limit; - const moodPage = moodHasMore ? moodRows.slice(0, limit) : moodRows; - const moodUpserts: MoodUpsert[] = []; - const moodTombstones: MoodTombstone[] = []; - for (const row of moodPage) { - if (row.deletedAt) { - moodTombstones.push({ - id: row.id, - syncVersion: row.syncVersion, - deletedAt: row.deletedAt.toISOString(), - updatedAt: row.updatedAt.toISOString(), - }); - } else { - moodUpserts.push({ - id: row.id, - date: row.date, - mood: row.mood, - score: row.score, - // Under the same wire keys the write path accepts, so a client reads - // back exactly what it can post. - ...levelAForWire(row), - context: row.context ? contextForWire(row.context) : null, - tags: row.tags, - note: readNote(row.noteEncrypted, row.note), - moodLoggedAt: row.moodLoggedAt.toISOString(), - source: row.source, - syncVersion: row.syncVersion, - updatedAt: row.updatedAt.toISOString(), - }); - } - } - advanceCursor(nextCursor, "mood", moodPage); - hasMore = hasMore || moodHasMore; - - // ── intakes ─────────────────────────────────────────────── - const intakeHasMore = intakeRows.length > limit; - const intakePage = intakeHasMore ? intakeRows.slice(0, limit) : intakeRows; - const intakeUpserts: IntakeUpsert[] = []; - const intakeTombstones: IntakeTombstone[] = []; - for (const row of intakePage) { - if (row.deletedAt) { - intakeTombstones.push({ - id: row.id, - syncVersion: row.syncVersion, - deletedAt: row.deletedAt.toISOString(), - updatedAt: row.updatedAt.toISOString(), - }); - } else { - intakeUpserts.push({ - id: row.id, - medicationId: row.medicationId, - scheduledFor: row.scheduledFor.toISOString(), - takenAt: row.takenAt?.toISOString() ?? null, - skipped: row.skipped, - source: row.source, - autoRecorded: row.autoRecorded, - syncVersion: row.syncVersion, - updatedAt: row.updatedAt.toISOString(), - }); - } - } - advanceCursor(nextCursor, "intakes", intakePage); - hasMore = hasMore || intakeHasMore; - - // ── cycleDays ───────────────────────────────────────────── - const cycleDayHasMore = cycleDayRows.length > limit; - const cycleDayPage = cycleDayHasMore - ? cycleDayRows.slice(0, limit) - : cycleDayRows; - const cycleDayUpserts: CycleDayLogDTO[] = []; - const cycleDayTombstones: CycleDayTombstone[] = []; - for (const row of cycleDayPage) { - if (row.deletedAt) { - cycleDayTombstones.push({ - id: row.id, - externalId: row.externalId, - syncVersion: row.syncVersion, - deletedAt: row.deletedAt.toISOString(), - updatedAt: row.updatedAt.toISOString(), - }); - } else { - cycleDayUpserts.push(toCycleDayLogDTO(row)); - } - } - advanceCursor(nextCursor, "cycleDays", cycleDayPage); - hasMore = hasMore || cycleDayHasMore; - - // ── cycles ──────────────────────────────────────────────── - const cycleHasMore = cycleRows.length > limit; - const cyclePage = cycleHasMore ? cycleRows.slice(0, limit) : cycleRows; - const cycleUpserts: MenstrualCycleDTO[] = []; - const cycleTombstones: CycleTombstone[] = []; - for (const row of cyclePage) { - if (row.deletedAt) { - cycleTombstones.push({ - id: row.id, - syncVersion: row.syncVersion, - deletedAt: row.deletedAt.toISOString(), - updatedAt: row.updatedAt.toISOString(), - }); - } else { - cycleUpserts.push(toMenstrualCycleDTO(row)); - } - } - advanceCursor(nextCursor, "cycles", cyclePage); - hasMore = hasMore || cycleHasMore; - - annotate({ - action: { name: "sync.changes.pull" }, - meta: { - measurement_upserts: measurementUpserts.length, - measurement_tombstones: measurementTombstones.length, - mood_upserts: moodUpserts.length, - mood_tombstones: moodTombstones.length, - intake_upserts: intakeUpserts.length, - intake_tombstones: intakeTombstones.length, - cycle_day_upserts: cycleDayUpserts.length, - cycle_day_tombstones: cycleDayTombstones.length, - cycle_upserts: cycleUpserts.length, - cycle_tombstones: cycleTombstones.length, - cycle_module_enabled: cycleEnabled, - has_more: hasMore, - cursor_present: Boolean(parsed.data.cursor), - }, - }); - - return apiSuccess({ - serverNow: serverNow.toISOString(), - cursor: encodeCursor(nextCursor, serverNow.getTime()), - hasMore, - cursorExpired: false, - changes: { - measurements: { - upserts: measurementUpserts, - tombstones: measurementTombstones, - }, - mood: { upserts: moodUpserts, tombstones: moodTombstones }, - intakes: { upserts: intakeUpserts, tombstones: intakeTombstones }, - cycleDays: { - upserts: cycleDayUpserts, - tombstones: cycleDayTombstones, - }, - cycles: { upserts: cycleUpserts, tombstones: cycleTombstones }, - }, - }); }); - -/** - * Advance a domain's watermark to the last row of its page. A domain - * that returned no rows keeps its prior watermark untouched so a future - * pull does not re-scan from the start. - */ -function advanceCursor( - cursor: SyncCursor, - domain: SyncDomain, - page: Array<{ updatedAt: Date; id: string }>, -): void { - const last = page[page.length - 1]; - if (last) { - cursor[domain] = { updatedAtMs: last.updatedAt.getTime(), id: last.id }; - } -} diff --git a/src/lib/sync/changes/domain-pages.ts b/src/lib/sync/changes/domain-pages.ts new file mode 100644 index 000000000..a2100337c --- /dev/null +++ b/src/lib/sync/changes/domain-pages.ts @@ -0,0 +1,229 @@ +/** + * Turn each domain's rows into its wire block: tombstones for the + * soft-deleted rows, upserts for the live ones. + */ +import { prisma } from "@/lib/db"; +import { + findCompactionTombstones, + isFoldedWindowCandidate, + loadFoldTimezone, +} from "@/lib/measurements/folded-window"; +import { readNote } from "@/lib/crypto/note-cipher"; +import { levelAForWire } from "@/lib/mood/level-a"; +import { contextForWire } from "@/lib/mood/context"; +import { + toCycleDayLogDTO, + toMenstrualCycleDTO, + type CycleDayLogDTO, + type MenstrualCycleDTO, +} from "@/lib/cycle/dto"; + +import type { + CycleDayRow, + CycleRow, + IntakeRow, + MeasurementRow, + MoodRow, +} from "./read-rows"; +import type { + CycleDayTombstone, + DomainBlock, + IdTombstone, + IntakeUpsert, + MeasurementTombstone, + MeasurementUpsert, + MoodUpsert, +} from "./wire"; + +/** A domain's page: the rows inside the limit, and whether more follow. */ +export function pageOf( + rows: R[], + limit: number, +): { page: R[]; hasMore: boolean } { + const hasMore = rows.length > limit; + return { page: hasMore ? rows.slice(0, limit) : rows, hasMore }; +} + +function idTombstone(row: { + id: string; + syncVersion: number; + deletedAt: Date; + updatedAt: Date; +}): IdTombstone { + return { + id: row.id, + syncVersion: row.syncVersion, + deletedAt: row.deletedAt.toISOString(), + updatedAt: row.updatedAt.toISOString(), + }; +} + +/** + * Compaction tombstones are not deletions anyone made. The folds used to + * soft-delete the raw Apple Health samples they folded into an hourly or + * daily `stats:` row, and those rows rode this feed as tombstones for 75 + * days. They are left out, by the same classifier the backlog purge uses + * (`folded-window.ts`); the cursor still advances past them, so the keyset + * walk is unchanged. A person's own deletions are reported exactly as + * before. + */ +async function compactionIds( + userId: string, + page: MeasurementRow[], +): Promise> { + const compaction = new Set(); + const tombstonesOnPage = page.filter((row) => row.deletedAt !== null); + if (tombstonesOnPage.length === 0) return compaction; + const candidates = tombstonesOnPage.map((row) => ({ + type: row.type, + source: row.source, + externalId: row.externalId, + measuredAt: row.measuredAt, + deletedAt: row.deletedAt, + })); + if (candidates.some((row) => isFoldedWindowCandidate(row, row.deletedAt!))) { + const classA = await findCompactionTombstones( + prisma, + userId, + await loadFoldTimezone(prisma, userId), + candidates, + ); + for (const index of classA) compaction.add(tombstonesOnPage[index].id); + } + return compaction; +} + +export async function measurementBlock( + userId: string, + page: MeasurementRow[], +): Promise> { + const compaction = await compactionIds(userId, page); + const upserts: MeasurementUpsert[] = []; + const tombstones: MeasurementTombstone[] = []; + for (const row of page) { + if (compaction.has(row.id)) continue; + if (row.deletedAt) { + tombstones.push({ + id: row.id, + externalId: row.externalId, + syncVersion: row.syncVersion, + deletedAt: row.deletedAt.toISOString(), + updatedAt: row.updatedAt.toISOString(), + }); + } else { + // Every row carries its `source`. RECOVERY_SCORE is written by BOTH the + // WHOOP-native sync (`source = WHOOP`) and the COMPUTED proxy job + // (`source = COMPUTED`) for the same day; the client must resolve to ONE + // canonical recovery series (WHOOP wins when present, COMPUTED is the + // fallback) and never chart the proxy and the native value as two + // competing series. The feed faithfully mirrors both rows (a delta-sync + // mirror cannot drop a row mid-keyset without a tombstone); the + // canonical pick is the same WHOOP-over-COMPUTED rule the server's + // wellness tile + doctor PDF apply (`resolveCanonicalRecovery`). + upserts.push({ + id: row.id, + externalId: row.externalId, + type: row.type, + value: row.value, + unit: row.unit, + measuredAt: row.measuredAt.toISOString(), + source: row.source, + notes: readNote(row.notesEncrypted, row.notes), + syncVersion: row.syncVersion, + updatedAt: row.updatedAt.toISOString(), + }); + } + } + return { upserts, tombstones }; +} + +export function moodBlock( + page: MoodRow[], +): DomainBlock { + const upserts: MoodUpsert[] = []; + const tombstones: IdTombstone[] = []; + for (const row of page) { + if (row.deletedAt) { + tombstones.push(idTombstone({ ...row, deletedAt: row.deletedAt })); + } else { + upserts.push({ + id: row.id, + date: row.date, + mood: row.mood, + score: row.score, + // Under the same wire keys the write path accepts, so a client reads + // back exactly what it can post. + ...levelAForWire(row), + context: row.context ? contextForWire(row.context) : null, + tags: row.tags, + note: readNote(row.noteEncrypted, row.note), + moodLoggedAt: row.moodLoggedAt.toISOString(), + source: row.source, + syncVersion: row.syncVersion, + updatedAt: row.updatedAt.toISOString(), + }); + } + } + return { upserts, tombstones }; +} + +export function intakeBlock( + page: IntakeRow[], +): DomainBlock { + const upserts: IntakeUpsert[] = []; + const tombstones: IdTombstone[] = []; + for (const row of page) { + if (row.deletedAt) { + tombstones.push(idTombstone({ ...row, deletedAt: row.deletedAt })); + } else { + upserts.push({ + id: row.id, + medicationId: row.medicationId, + scheduledFor: row.scheduledFor.toISOString(), + takenAt: row.takenAt?.toISOString() ?? null, + skipped: row.skipped, + source: row.source, + autoRecorded: row.autoRecorded, + syncVersion: row.syncVersion, + updatedAt: row.updatedAt.toISOString(), + }); + } + } + return { upserts, tombstones }; +} + +export function cycleDayBlock( + page: CycleDayRow[], +): DomainBlock { + const upserts: CycleDayLogDTO[] = []; + const tombstones: CycleDayTombstone[] = []; + for (const row of page) { + if (row.deletedAt) { + tombstones.push({ + id: row.id, + externalId: row.externalId, + syncVersion: row.syncVersion, + deletedAt: row.deletedAt.toISOString(), + updatedAt: row.updatedAt.toISOString(), + }); + } else { + upserts.push(toCycleDayLogDTO(row)); + } + } + return { upserts, tombstones }; +} + +export function cycleBlock( + page: CycleRow[], +): DomainBlock { + const upserts: MenstrualCycleDTO[] = []; + const tombstones: IdTombstone[] = []; + for (const row of page) { + if (row.deletedAt) { + tombstones.push(idTombstone({ ...row, deletedAt: row.deletedAt })); + } else { + upserts.push(toMenstrualCycleDTO(row)); + } + } + return { upserts, tombstones }; +} diff --git a/src/lib/sync/changes/pull.ts b/src/lib/sync/changes/pull.ts new file mode 100644 index 000000000..553607f72 --- /dev/null +++ b/src/lib/sync/changes/pull.ts @@ -0,0 +1,218 @@ +/** + * One pull of the multi-domain delta feed behind `GET /api/sync/changes`. + * + * The route authenticates, meters and parses the query; this module decides + * whether the cursor can still be caught up, reads one page of every domain + * past its watermark, and answers the blocks and the next cursor. The feed + * contract is described on the route. + */ +import { prisma } from "@/lib/db"; +import { annotate } from "@/lib/logging/context"; +import { TOMBSTONE_RETENTION_DAYS } from "@/lib/auth/native-client"; +import { + decodeCursor, + decodeCursorIssuedAt, + encodeCursor, + SYNC_DOMAINS, + type SyncCursor, + type SyncDomain, +} from "@/lib/sync/cursor"; +import { isModuleEnabled } from "@/lib/modules/gate"; + +import { readDomainRows } from "./read-rows"; +import { + cycleBlock, + cycleDayBlock, + intakeBlock, + measurementBlock, + moodBlock, + pageOf, +} from "./domain-pages"; +import type { SyncChangesBody } from "./wire"; + +const DAY_MS = 86_400_000; + +/** + * Advance a domain's watermark to the last row of its page. A domain that + * returned no rows keeps its prior watermark untouched so a future pull does + * not re-scan from the start. + */ +function advanceCursor( + cursor: SyncCursor, + domain: SyncDomain, + page: Array<{ updatedAt: Date; id: string }>, +): void { + const last = page[page.length - 1]; + if (last) { + cursor[domain] = { updatedAtMs: last.updatedAt.getTime(), id: last.id }; + } +} + +/** + * Whether the cursor can no longer be caught up incrementally, and why. + * + * Past retention: ANY domain watermark older than the tombstone-retention + * horizon. Tombstones older than that may have been pruned by the cleanup + * job, so an incremental delta could silently miss a deletion. + * + * Before a reset: a cursor issued before the account's record was last + * replaced (a backup restore, or the person deleting all their data) cannot + * be caught up either: the restore wrote rows back with their original + * `updatedAt`, behind the cursor's watermarks, and removed rows without + * tombstones. Compared by the time the cursor was issued, not by its + * watermarks, which a restore leaves meaningless; a cursor from before the + * issue time was recorded counts as issued before any reset. + */ +async function expiry( + userId: string, + token: string | undefined, + cursor: SyncCursor, + serverNow: Date, +): Promise<"retention" | "sync_reset" | null> { + const retentionHorizon = new Date( + serverNow.getTime() - TOMBSTONE_RETENTION_DAYS * DAY_MS, + ); + const pastRetention = SYNC_DOMAINS.some((domain) => { + const wm = cursor[domain]; + return wm !== undefined && wm.updatedAtMs < retentionHorizon.getTime(); + }); + let beforeReset = false; + if (token) { + const account = await prisma.user.findUnique({ + where: { id: userId }, + select: { syncResetAt: true }, + }); + if (account?.syncResetAt) { + const issuedAt = decodeCursorIssuedAt(token); + beforeReset = + issuedAt === null || issuedAt < account.syncResetAt.getTime(); + } + } + if (!pastRetention && !beforeReset) return null; + return beforeReset ? "sync_reset" : "retention"; +} + +function emptyChanges(): SyncChangesBody["changes"] { + return { + measurements: { upserts: [], tombstones: [] }, + mood: { upserts: [], tombstones: [] }, + intakes: { upserts: [], tombstones: [] }, + cycleDays: { upserts: [], tombstones: [] }, + cycles: { upserts: [], tombstones: [] }, + }; +} + +export async function pullSyncChanges(input: { + userId: string; + /** The opaque cursor the client echoed, if any. */ + token: string | undefined; + limit: number; + serverNow: Date; +}): Promise { + const { userId, token, limit, serverNow } = input; + + // Module gate: cycle. The `cycleDays` / `cycles` domains carry period + // days and per-day symptom rows, the same data every `/api/cycle/*` route + // refuses with a 403 when the module is off. Without this the delta feed + // re-served them ungated to the very same token. + // + // OMIT, NOT 403, deliberately, and unlike the whole-record FHIR export. + // This feed multiplexes five domains over one cursor: refusing the whole + // request would wedge measurement / mood / intake catch-up for every + // account with cycle tracking off, which (the gate delegates to + // `isCycleEnabled`) is the default for most accounts. A partial payload is + // the only outcome that keeps the client syncing while still honouring the + // promise that a disabled module hands out nothing. + // + // The two cycle watermarks are deliberately NOT advanced while the module + // is off (the queries are skipped, so `advanceCursor` sees an empty page + // and leaves the prior watermark untouched). Re-enabling therefore resumes + // from where the client actually left off instead of silently skipping the + // rows that changed while the module was hidden. + // + // KNOWN GAP, out of scope here: a client that synced cycle rows BEFORE the + // module was turned off keeps its stale local mirror. Emitting tombstones + // for them would be wrong (a tombstone means deleted, and disabling a + // module must leave the rows intact so re-enabling finds a complete + // history). Purging a paired client's mirror needs an explicit "module went + // dark" signal in the feed contract plus client support for it; that is an + // additive wire change and an iOS-client change, not a server-side gate + // fix, so it is left for a coordinated contract cycle. + const cycleEnabled = await isModuleEnabled(userId, "cycle"); + + // A garbage / unparseable / old-format cursor decodes to null → fresh + // initial sync (every domain scans from the start). + const cursor: SyncCursor = token ? (decodeCursor(token) ?? {}) : {}; + + const expired = await expiry(userId, token, cursor, serverNow); + if (expired) { + annotate({ + action: { name: "sync.changes.pull" }, + meta: { + cursor_expired: true, + cursor_expired_reason: expired, + returned: 0, + }, + }); + return { + serverNow: serverNow.toISOString(), + cursor: token ?? null, + hasMore: false, + cursorExpired: true, + changes: emptyChanges(), + }; + } + + const rows = await readDomainRows(userId, cursor, limit, cycleEnabled); + const nextCursor: SyncCursor = { ...cursor }; + + const m = pageOf(rows.measurements, limit); + const measurements = await measurementBlock(userId, m.page); + advanceCursor(nextCursor, "measurements", m.page); + + const md = pageOf(rows.mood, limit); + const mood = moodBlock(md.page); + advanceCursor(nextCursor, "mood", md.page); + + const ik = pageOf(rows.intakes, limit); + const intakes = intakeBlock(ik.page); + advanceCursor(nextCursor, "intakes", ik.page); + + const cd = pageOf(rows.cycleDays, limit); + const cycleDays = cycleDayBlock(cd.page); + advanceCursor(nextCursor, "cycleDays", cd.page); + + const cy = pageOf(rows.cycles, limit); + const cycles = cycleBlock(cy.page); + advanceCursor(nextCursor, "cycles", cy.page); + + const hasMore = + m.hasMore || md.hasMore || ik.hasMore || cd.hasMore || cy.hasMore; + + annotate({ + action: { name: "sync.changes.pull" }, + meta: { + measurement_upserts: measurements.upserts.length, + measurement_tombstones: measurements.tombstones.length, + mood_upserts: mood.upserts.length, + mood_tombstones: mood.tombstones.length, + intake_upserts: intakes.upserts.length, + intake_tombstones: intakes.tombstones.length, + cycle_day_upserts: cycleDays.upserts.length, + cycle_day_tombstones: cycleDays.tombstones.length, + cycle_upserts: cycles.upserts.length, + cycle_tombstones: cycles.tombstones.length, + cycle_module_enabled: cycleEnabled, + has_more: hasMore, + cursor_present: Boolean(token), + }, + }); + + return { + serverNow: serverNow.toISOString(), + cursor: encodeCursor(nextCursor, serverNow.getTime()), + hasMore, + cursorExpired: false, + changes: { measurements, mood, intakes, cycleDays, cycles }, + }; +} diff --git a/src/lib/sync/changes/read-rows.ts b/src/lib/sync/changes/read-rows.ts new file mode 100644 index 000000000..9beae88f4 --- /dev/null +++ b/src/lib/sync/changes/read-rows.ts @@ -0,0 +1,147 @@ +/** + * The per-domain keyset walks of the delta feed. + * + * Live and soft-deleted rows come out of the same scan: a soft-delete bumps + * `updatedAt`, so a tombstone is just a row whose `deletedAt` is non-null. + * Each walk fetches `limit + 1` rows to detect `hasMore` without a count + * query. + */ +import { prisma } from "@/lib/db"; +import type { Prisma } from "@/generated/prisma/client"; +import type { DomainWatermark, SyncCursor } from "@/lib/sync/cursor"; +import { dayLogSymptomInclude } from "@/lib/cycle/dto"; + +const MEASUREMENT_SELECT = { + id: true, + externalId: true, + type: true, + value: true, + unit: true, + measuredAt: true, + source: true, + notes: true, + notesEncrypted: true, + syncVersion: true, + deletedAt: true, + updatedAt: true, +} satisfies Prisma.MeasurementSelect; + +const MOOD_SELECT = { + id: true, + date: true, + mood: true, + score: true, + moodA1: true, + stressA2: true, + energyA3: true, + connectionA4: true, + stabilityA5: true, + context: true, + tags: true, + note: true, + noteEncrypted: true, + moodLoggedAt: true, + source: true, + syncVersion: true, + deletedAt: true, + updatedAt: true, +} satisfies Prisma.MoodEntrySelect; + +const INTAKE_SELECT = { + id: true, + medicationId: true, + scheduledFor: true, + takenAt: true, + skipped: true, + source: true, + autoRecorded: true, + syncVersion: true, + deletedAt: true, + updatedAt: true, +} satisfies Prisma.MedicationIntakeEventSelect; + +export type MeasurementRow = Prisma.MeasurementGetPayload<{ + select: typeof MEASUREMENT_SELECT; +}>; +export type MoodRow = Prisma.MoodEntryGetPayload<{ + select: typeof MOOD_SELECT; +}>; +export type IntakeRow = Prisma.MedicationIntakeEventGetPayload<{ + select: typeof INTAKE_SELECT; +}>; +export type CycleDayRow = Prisma.CycleDayLogGetPayload<{ + include: typeof dayLogSymptomInclude; +}>; +export type CycleRow = Prisma.MenstrualCycleGetPayload; + +/** + * Build the `(updatedAt, id)` keyset predicate for a domain's watermark. + * Empty object (full scan) when the client has no watermark for it yet. + */ +function keysetFilter(wm: DomainWatermark | undefined) { + if (!wm) return {}; + const at = new Date(wm.updatedAtMs); + return { + OR: [ + { updatedAt: { gt: at } }, + { AND: [{ updatedAt: at }, { id: { gt: wm.id } }] }, + ], + }; +} + +export interface DomainRows { + measurements: MeasurementRow[]; + mood: MoodRow[]; + intakes: IntakeRow[]; + cycleDays: CycleDayRow[]; + cycles: CycleRow[]; +} + +/** + * Read one page plus one row of every domain past its watermark. The cycle + * walks are skipped outright when the module is off, not filtered after the + * fact, so the disabled rows are never read out of Postgres at all. + */ +export async function readDomainRows( + userId: string, + cursor: SyncCursor, + limit: number, + cycleEnabled: boolean, +): Promise { + const [measurements, mood, intakes, cycleDays, cycles] = await Promise.all([ + prisma.measurement.findMany({ + where: { userId, ...keysetFilter(cursor.measurements) }, + orderBy: [{ updatedAt: "asc" }, { id: "asc" }], + take: limit + 1, + select: MEASUREMENT_SELECT, + }), + prisma.moodEntry.findMany({ + where: { userId, ...keysetFilter(cursor.mood) }, + orderBy: [{ updatedAt: "asc" }, { id: "asc" }], + take: limit + 1, + select: MOOD_SELECT, + }), + prisma.medicationIntakeEvent.findMany({ + where: { userId, ...keysetFilter(cursor.intakes) }, + orderBy: [{ updatedAt: "asc" }, { id: "asc" }], + take: limit + 1, + select: INTAKE_SELECT, + }), + cycleEnabled + ? prisma.cycleDayLog.findMany({ + where: { userId, ...keysetFilter(cursor.cycleDays) }, + orderBy: [{ updatedAt: "asc" }, { id: "asc" }], + take: limit + 1, + include: dayLogSymptomInclude, + }) + : [], + cycleEnabled + ? prisma.menstrualCycle.findMany({ + where: { userId, ...keysetFilter(cursor.cycles) }, + orderBy: [{ updatedAt: "asc" }, { id: "asc" }], + take: limit + 1, + }) + : [], + ]); + return { measurements, mood, intakes, cycleDays, cycles }; +} diff --git a/src/lib/sync/changes/wire.ts b/src/lib/sync/changes/wire.ts new file mode 100644 index 000000000..d4a95ed6b --- /dev/null +++ b/src/lib/sync/changes/wire.ts @@ -0,0 +1,114 @@ +/** + * The wire rows of `GET /api/sync/changes`, one upsert and one tombstone + * shape per domain. Tombstone identity follows the iOS §7.3 table: + * measurements key on `externalId` (the cross-device key); mood and intakes + * key on the server `id` (the client mirrors those by server id); cycle days + * key on `externalId` when present (HealthKit-origin rows are + * externalId-keyed cross-device, like measurements) else the server `id`; + * cycles key on the server `id` (computed rows carry no externalId). + */ +import type { MoodContextWire } from "@/lib/mood/context"; +import type { CycleDayLogDTO, MenstrualCycleDTO } from "@/lib/cycle/dto"; + +export interface MeasurementUpsert { + id: string; + externalId: string | null; + type: string; + value: number; + unit: string; + measuredAt: string; + source: string; + notes: string | null; + syncVersion: number; + updatedAt: string; +} + +export interface MeasurementTombstone { + id: string; + externalId: string | null; + syncVersion: number; + deletedAt: string; + updatedAt: string; +} + +export interface MoodUpsert { + id: string; + date: string; + mood: string; + score: number; + /** + * The five level-A self-state values, each 0-10 and each nullable. `a2` is + * inverse-oriented: a higher number means more stress, and a client that + * wants "up is better" flips the sign itself rather than the server + * storing a flipped value. + * + * Server-authoritative: `a1` is derived from the five-point label on every + * write, so a client that sends only `mood` still mirrors a complete row. + */ + a1: number | null; + a2: number | null; + a3: number | null; + a4: number | null; + a5: number | null; + /** + * The day's context, or null when the entry carries none. Null and an + * object of nulls are different answers and the feed keeps them apart. + */ + context: MoodContextWire | null; + tags: string | null; + note: string | null; + moodLoggedAt: string; + source: string; + syncVersion: number; + updatedAt: string; +} + +/** The tombstone of a domain keyed on the server id alone. */ +export interface IdTombstone { + id: string; + syncVersion: number; + deletedAt: string; + updatedAt: string; +} + +export interface IntakeUpsert { + id: string; + medicationId: string; + scheduledFor: string; + takenAt: string | null; + skipped: boolean; + source: string; + /** Recorded by the server (`autoMarkTaken`), not the person. */ + autoRecorded: boolean; + syncVersion: number; + updatedAt: string; +} + +export interface CycleDayTombstone { + id: string; + externalId: string | null; + syncVersion: number; + deletedAt: string; + updatedAt: string; +} + +export interface DomainBlock { + upserts: U[]; + tombstones: T[]; +} + +export interface SyncChanges { + measurements: DomainBlock; + mood: DomainBlock; + intakes: DomainBlock; + cycleDays: DomainBlock; + cycles: DomainBlock; +} + +export interface SyncChangesBody { + serverNow: string; + cursor: string | null; + hasMore: boolean; + cursorExpired: boolean; + changes: SyncChanges; +}