From aab2e0ae7d8041f403ec4b352cfe6092a22ef32f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Fri, 23 Oct 2026 19:28:04 +0200 Subject: [PATCH] feat(sync): re-deliver Apple Health write-back changes Measurement upserts in /api/sync/changes carry appleHealthMirror. A new measurementMirror domain delivers { id, appleHealthMirror } for the live readings of every token whose setting changed past the cursor, keyed on (token.mirrorChangedAt, measurement.id). Switching the setting writes the token row only, so the readings keep their updatedAt and syncVersion and nothing keyed on them recomputes. The domain is additive inside the v2 cursor and is left out of the retention expiry, since its watermark is a setting's age, not a deletion's. --- .../changes/__tests__/module-gate.test.ts | 35 ++--- src/lib/sync/changes/domain-pages.ts | 16 +++ src/lib/sync/changes/pull.ts | 35 ++++- src/lib/sync/changes/read-rows.ts | 123 +++++++++++++----- src/lib/sync/changes/wire.ts | 13 ++ src/lib/sync/cursor.ts | 19 ++- .../sync-changes.01-first-page.json.snap | 7 + .../sync-changes.02-second-page.json.snap | 6 + .../sync-changes.03-the-rest.json.snap | 6 + .../sync-changes.04-caught-up.json.snap | 5 + .../sync-changes.05-past-retention.json.snap | 4 + .../sync-changes.06-before-reset.json.snap | 4 + .../sync-changes.07-cycle-off.json.snap | 9 ++ 13 files changed, 226 insertions(+), 56 deletions(-) diff --git a/src/app/api/sync/changes/__tests__/module-gate.test.ts b/src/app/api/sync/changes/__tests__/module-gate.test.ts index 5b5e004c9..e75cd7145 100644 --- a/src/app/api/sync/changes/__tests__/module-gate.test.ts +++ b/src/app/api/sync/changes/__tests__/module-gate.test.ts @@ -106,22 +106,25 @@ beforeEach(() => { } as never); // Non-cycle domains always return rows so a 403/empty-everything result // cannot be mistaken for a passing "cycle omitted" assertion. - vi.mocked(prisma.measurement.findMany).mockResolvedValue([ - { - id: "m-1", - externalId: "ext-m-1", - type: "WEIGHT", - value: 80, - unit: "kg", - measuredAt: NOW, - source: "MANUAL", - notes: null, - notesEncrypted: null, - syncVersion: 1, - deletedAt: null, - updatedAt: NOW, - }, - ] as never); + // The `measurementMirror` walk reads measurements too (the readings of + // tokens whose Apple Health setting changed); it finds none here. + const measurementRow = { + id: "m-1", + externalId: "ext-m-1", + type: "WEIGHT", + value: 80, + unit: "kg", + measuredAt: NOW, + source: "MANUAL", + notes: null, + notesEncrypted: null, + syncVersion: 1, + deletedAt: null, + updatedAt: NOW, + }; + vi.mocked(prisma.measurement.findMany).mockImplementation((async (args: { + where?: { ingestToken?: unknown }; + }) => (args.where?.ingestToken ? [] : [measurementRow])) as never); vi.mocked(prisma.moodEntry.findMany).mockResolvedValue([] as never); vi.mocked(prisma.medicationIntakeEvent.findMany).mockResolvedValue( [] as never, diff --git a/src/lib/sync/changes/domain-pages.ts b/src/lib/sync/changes/domain-pages.ts index a2100337c..84b3eaf64 100644 --- a/src/lib/sync/changes/domain-pages.ts +++ b/src/lib/sync/changes/domain-pages.ts @@ -9,6 +9,7 @@ import { loadFoldTimezone, } from "@/lib/measurements/folded-window"; import { readNote } from "@/lib/crypto/note-cipher"; +import { resolveAppleHealthMirror } from "@/lib/measurements/apple-health-mirror"; import { levelAForWire } from "@/lib/mood/level-a"; import { contextForWire } from "@/lib/mood/context"; import { @@ -22,6 +23,7 @@ import type { CycleDayRow, CycleRow, IntakeRow, + MeasurementMirrorRow, MeasurementRow, MoodRow, } from "./read-rows"; @@ -30,6 +32,7 @@ import type { DomainBlock, IdTombstone, IntakeUpsert, + MeasurementMirrorUpsert, MeasurementTombstone, MeasurementUpsert, MoodUpsert, @@ -129,6 +132,7 @@ export async function measurementBlock( measuredAt: row.measuredAt.toISOString(), source: row.source, notes: readNote(row.notesEncrypted, row.notes), + appleHealthMirror: resolveAppleHealthMirror(row), syncVersion: row.syncVersion, updatedAt: row.updatedAt.toISOString(), }); @@ -137,6 +141,18 @@ export async function measurementBlock( return { upserts, tombstones }; } +export function measurementMirrorBlock( + page: MeasurementMirrorRow[], +): DomainBlock { + return { + upserts: page.map((row) => ({ + id: row.id, + appleHealthMirror: resolveAppleHealthMirror(row), + })), + tombstones: [], + }; +} + export function moodBlock( page: MoodRow[], ): DomainBlock { diff --git a/src/lib/sync/changes/pull.ts b/src/lib/sync/changes/pull.ts index 594ec3a3d..cae7aed19 100644 --- a/src/lib/sync/changes/pull.ts +++ b/src/lib/sync/changes/pull.ts @@ -25,6 +25,7 @@ import { cycleDayBlock, intakeBlock, measurementBlock, + measurementMirrorBlock, moodBlock, pageOf, } from "./domain-pages"; @@ -72,7 +73,11 @@ async function expiry( const retentionHorizon = new Date( serverNow.getTime() - TOMBSTONE_RETENTION_DAYS * DAY_MS, ); + // Not `measurementMirror`: its watermark is when a token's setting last + // changed, which can be any age, and the domain re-delivers values, not + // deletions, so no pruned tombstone can hide behind it. const pastRetention = SYNC_DOMAINS.some((domain) => { + if (domain === "measurementMirror") return false; const wm = cursor[domain]; return wm !== undefined && wm.updatedAtMs < retentionHorizon.getTime(); }); @@ -99,6 +104,7 @@ function emptyChanges(): SyncChangesBody["changes"] { intakes: { upserts: [], tombstones: [] }, cycleDays: { upserts: [], tombstones: [] }, cycles: { upserts: [], tombstones: [] }, + measurementMirror: { upserts: [], tombstones: [] }, }; } @@ -186,8 +192,25 @@ export async function pullSyncChanges(input: { const cycles = cycleBlock(cy.page); advanceCursor(nextCursor, "cycles", cy.page); + const mm = pageOf(rows.measurementMirror, limit); + const measurementMirror = measurementMirrorBlock(mm.page); + advanceCursor( + nextCursor, + "measurementMirror", + mm.page.map((row) => ({ + // The walk only returns rows whose token carries a change stamp. + updatedAt: row.ingestToken!.mirrorChangedAt!, + id: row.id, + })), + ); + const hasMore = - m.hasMore || md.hasMore || ik.hasMore || cd.hasMore || cy.hasMore; + m.hasMore || + md.hasMore || + ik.hasMore || + cd.hasMore || + cy.hasMore || + mm.hasMore; annotate({ action: { name: "sync.changes.pull" }, @@ -202,6 +225,7 @@ export async function pullSyncChanges(input: { cycle_day_tombstones: cycleDays.tombstones.length, cycle_upserts: cycles.upserts.length, cycle_tombstones: cycles.tombstones.length, + measurement_mirror_upserts: measurementMirror.upserts.length, cycle_module_enabled: cycleEnabled, has_more: hasMore, cursor_present: Boolean(token), @@ -213,6 +237,13 @@ export async function pullSyncChanges(input: { cursor: encodeCursor(nextCursor, serverNow.getTime()), hasMore, cursorExpired: false, - changes: { measurements, mood, intakes, cycleDays, cycles }, + changes: { + measurements, + mood, + intakes, + cycleDays, + cycles, + measurementMirror, + }, }; } diff --git a/src/lib/sync/changes/read-rows.ts b/src/lib/sync/changes/read-rows.ts index 9beae88f4..b23595c79 100644 --- a/src/lib/sync/changes/read-rows.ts +++ b/src/lib/sync/changes/read-rows.ts @@ -10,6 +10,7 @@ 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"; +import { APPLE_HEALTH_MIRROR_SELECT } from "@/lib/measurements/apple-health-mirror"; const MEASUREMENT_SELECT = { id: true, @@ -24,6 +25,13 @@ const MEASUREMENT_SELECT = { syncVersion: true, deletedAt: true, updatedAt: true, + ingestToken: APPLE_HEALTH_MIRROR_SELECT.ingestToken, +} satisfies Prisma.MeasurementSelect; + +const MEASUREMENT_MIRROR_SELECT = { + id: true, + source: true, + ingestToken: { select: { mirrorToAppleHealth: true, mirrorChangedAt: true } }, } satisfies Prisma.MeasurementSelect; const MOOD_SELECT = { @@ -73,6 +81,9 @@ export type CycleDayRow = Prisma.CycleDayLogGetPayload<{ include: typeof dayLogSymptomInclude; }>; export type CycleRow = Prisma.MenstrualCycleGetPayload; +export type MeasurementMirrorRow = Prisma.MeasurementGetPayload<{ + select: typeof MEASUREMENT_MIRROR_SELECT; +}>; /** * Build the `(updatedAt, id)` keyset predicate for a domain's watermark. @@ -95,6 +106,44 @@ export interface DomainRows { intakes: IntakeRow[]; cycleDays: CycleDayRow[]; cycles: CycleRow[]; + measurementMirror: MeasurementMirrorRow[]; +} + +/** + * The live readings of every token whose Apple Health setting changed past + * the watermark, in `(token.mirrorChangedAt, measurement.id)` order. Switching + * the setting writes the token row only, so the readings keep their + * `updatedAt` (nothing keyed on it recomputes) and arrive through this walk + * instead of the measurement one. + */ +function readMeasurementMirrorRows( + userId: string, + wm: DomainWatermark | undefined, + limit: number, +): Promise { + const at = wm ? new Date(wm.updatedAtMs) : null; + return prisma.measurement.findMany({ + where: { + userId, + deletedAt: null, + ingestToken: { mirrorChangedAt: { not: null } }, + ...(at && + wm && { + OR: [ + { ingestToken: { mirrorChangedAt: { gt: at } } }, + { + AND: [ + { ingestToken: { mirrorChangedAt: at } }, + { id: { gt: wm.id } }, + ], + }, + ], + }), + }, + orderBy: [{ ingestToken: { mirrorChangedAt: "asc" } }, { id: "asc" }], + take: limit + 1, + select: MEASUREMENT_MIRROR_SELECT, + }); } /** @@ -108,40 +157,42 @@ export async function readDomainRows( 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 }; + const [measurements, mood, intakes, cycleDays, cycles, measurementMirror] = + 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, + }) + : [], + readMeasurementMirrorRows(userId, cursor.measurementMirror, limit), + ]); + return { measurements, mood, intakes, cycleDays, cycles, measurementMirror }; } diff --git a/src/lib/sync/changes/wire.ts b/src/lib/sync/changes/wire.ts index d4a95ed6b..cc01311df 100644 --- a/src/lib/sync/changes/wire.ts +++ b/src/lib/sync/changes/wire.ts @@ -19,6 +19,8 @@ export interface MeasurementUpsert { measuredAt: string; source: string; notes: string | null; + /** Whether the app writes the reading into Apple Health (#1272). */ + appleHealthMirror: boolean; syncVersion: number; updatedAt: string; } @@ -97,12 +99,23 @@ export interface DomainBlock { tombstones: T[]; } +/** + * A reading whose Apple Health write-back changed because its token's + * setting did (#1272). The reading itself did not change and is not re-sent. + */ +export interface MeasurementMirrorUpsert { + id: string; + appleHealthMirror: boolean; +} + export interface SyncChanges { measurements: DomainBlock; mood: DomainBlock; intakes: DomainBlock; cycleDays: DomainBlock; cycles: DomainBlock; + /** Never carries tombstones: a deleted reading is a measurement tombstone. */ + measurementMirror: DomainBlock; } export interface SyncChangesBody { diff --git a/src/lib/sync/cursor.ts b/src/lib/sync/cursor.ts index 3b207e9d5..4d8d1c161 100644 --- a/src/lib/sync/cursor.ts +++ b/src/lib/sync/cursor.ts @@ -28,19 +28,34 @@ /** The sync domains the feed serves. */ export type SyncDomain = - "measurements" | "mood" | "intakes" | "cycleDays" | "cycles"; + | "measurements" + | "mood" + | "intakes" + | "cycleDays" + | "cycles" + | "measurementMirror"; +/** + * `measurementMirror` (#1272) is additive inside the v2 envelope: a token + * without it decodes as before and that domain scans from the start. Its + * watermark is not a row's `updatedAt` but the `mirrorChangedAt` of the + * token whose readings were last re-delivered (see `read-rows.ts`). + */ export const SYNC_DOMAINS: readonly SyncDomain[] = [ "measurements", "mood", "intakes", "cycleDays", "cycles", + "measurementMirror", ]; /** A single domain's keyset position. */ export interface DomainWatermark { - /** `updatedAt` of the last drained row, as epoch milliseconds. */ + /** + * `updatedAt` of the last drained row, as epoch milliseconds; for + * `measurementMirror`, the `mirrorChangedAt` of that row's token. + */ updatedAtMs: number; /** `id` of the last drained row (cuid) — the keyset tie-breaker. */ id: string; diff --git a/tests/integration/routes/parity/__snapshots__/sync-changes.01-first-page.json.snap b/tests/integration/routes/parity/__snapshots__/sync-changes.01-first-page.json.snap index cac388d58..6e7ef10b0 100644 --- a/tests/integration/routes/parity/__snapshots__/sync-changes.01-first-page.json.snap +++ b/tests/integration/routes/parity/__snapshots__/sync-changes.01-first-page.json.snap @@ -18,6 +18,7 @@ "measuredAt": "2026-10-14T06:30:00.000Z", "source": "MANUAL", "notes": "after the run", + "appleHealthMirror": true, "syncVersion": "", "updatedAt": "" }, @@ -30,6 +31,7 @@ "measuredAt": "2026-10-14T07:00:00.000Z", "source": "WITHINGS", "notes": null, + "appleHealthMirror": true, "syncVersion": "", "updatedAt": "" } @@ -197,6 +199,10 @@ } ], "tombstones": [] + }, + "measurementMirror": { + "upserts": [], + "tombstones": [] } } }, @@ -219,6 +225,7 @@ "cycle_day_tombstones": 0, "cycle_upserts": 2, "cycle_tombstones": 0, + "measurement_mirror_upserts": 0, "cycle_module_enabled": true, "has_more": true, "cursor_present": false diff --git a/tests/integration/routes/parity/__snapshots__/sync-changes.02-second-page.json.snap b/tests/integration/routes/parity/__snapshots__/sync-changes.02-second-page.json.snap index 96ebe0b69..735d06770 100644 --- a/tests/integration/routes/parity/__snapshots__/sync-changes.02-second-page.json.snap +++ b/tests/integration/routes/parity/__snapshots__/sync-changes.02-second-page.json.snap @@ -18,6 +18,7 @@ "measuredAt": "2026-06-17T09:30:00.000Z", "source": "APPLE_HEALTH", "notes": null, + "appleHealthMirror": false, "syncVersion": "", "updatedAt": "" } @@ -69,6 +70,10 @@ "updatedAt": "" } ] + }, + "measurementMirror": { + "upserts": [], + "tombstones": [] } } }, @@ -91,6 +96,7 @@ "cycle_day_tombstones": 1, "cycle_upserts": 0, "cycle_tombstones": 1, + "measurement_mirror_upserts": 0, "cycle_module_enabled": true, "has_more": true, "cursor_present": true diff --git a/tests/integration/routes/parity/__snapshots__/sync-changes.03-the-rest.json.snap b/tests/integration/routes/parity/__snapshots__/sync-changes.03-the-rest.json.snap index 88e5797a6..97841d250 100644 --- a/tests/integration/routes/parity/__snapshots__/sync-changes.03-the-rest.json.snap +++ b/tests/integration/routes/parity/__snapshots__/sync-changes.03-the-rest.json.snap @@ -18,6 +18,7 @@ "measuredAt": "2026-10-14T08:00:00.000Z", "source": "MANUAL", "notes": null, + "appleHealthMirror": true, "syncVersion": "", "updatedAt": "" } @@ -39,6 +40,10 @@ "cycles": { "upserts": [], "tombstones": [] + }, + "measurementMirror": { + "upserts": [], + "tombstones": [] } } }, @@ -61,6 +66,7 @@ "cycle_day_tombstones": 0, "cycle_upserts": 0, "cycle_tombstones": 0, + "measurement_mirror_upserts": 0, "cycle_module_enabled": true, "has_more": false, "cursor_present": true diff --git a/tests/integration/routes/parity/__snapshots__/sync-changes.04-caught-up.json.snap b/tests/integration/routes/parity/__snapshots__/sync-changes.04-caught-up.json.snap index d50f7efae..e74d39368 100644 --- a/tests/integration/routes/parity/__snapshots__/sync-changes.04-caught-up.json.snap +++ b/tests/integration/routes/parity/__snapshots__/sync-changes.04-caught-up.json.snap @@ -26,6 +26,10 @@ "cycles": { "upserts": [], "tombstones": [] + }, + "measurementMirror": { + "upserts": [], + "tombstones": [] } } }, @@ -48,6 +52,7 @@ "cycle_day_tombstones": 0, "cycle_upserts": 0, "cycle_tombstones": 0, + "measurement_mirror_upserts": 0, "cycle_module_enabled": true, "has_more": false, "cursor_present": true diff --git a/tests/integration/routes/parity/__snapshots__/sync-changes.05-past-retention.json.snap b/tests/integration/routes/parity/__snapshots__/sync-changes.05-past-retention.json.snap index 06894579d..237975e23 100644 --- a/tests/integration/routes/parity/__snapshots__/sync-changes.05-past-retention.json.snap +++ b/tests/integration/routes/parity/__snapshots__/sync-changes.05-past-retention.json.snap @@ -26,6 +26,10 @@ "cycles": { "upserts": [], "tombstones": [] + }, + "measurementMirror": { + "upserts": [], + "tombstones": [] } } }, diff --git a/tests/integration/routes/parity/__snapshots__/sync-changes.06-before-reset.json.snap b/tests/integration/routes/parity/__snapshots__/sync-changes.06-before-reset.json.snap index 360c66c24..52d0edffe 100644 --- a/tests/integration/routes/parity/__snapshots__/sync-changes.06-before-reset.json.snap +++ b/tests/integration/routes/parity/__snapshots__/sync-changes.06-before-reset.json.snap @@ -26,6 +26,10 @@ "cycles": { "upserts": [], "tombstones": [] + }, + "measurementMirror": { + "upserts": [], + "tombstones": [] } } }, diff --git a/tests/integration/routes/parity/__snapshots__/sync-changes.07-cycle-off.json.snap b/tests/integration/routes/parity/__snapshots__/sync-changes.07-cycle-off.json.snap index e26c893fc..f4c382042 100644 --- a/tests/integration/routes/parity/__snapshots__/sync-changes.07-cycle-off.json.snap +++ b/tests/integration/routes/parity/__snapshots__/sync-changes.07-cycle-off.json.snap @@ -18,6 +18,7 @@ "measuredAt": "2026-10-14T06:30:00.000Z", "source": "MANUAL", "notes": "after the run", + "appleHealthMirror": true, "syncVersion": "", "updatedAt": "" }, @@ -30,6 +31,7 @@ "measuredAt": "2026-10-14T07:00:00.000Z", "source": "WITHINGS", "notes": null, + "appleHealthMirror": true, "syncVersion": "", "updatedAt": "" }, @@ -42,6 +44,7 @@ "measuredAt": "2026-06-17T09:30:00.000Z", "source": "APPLE_HEALTH", "notes": null, + "appleHealthMirror": false, "syncVersion": "", "updatedAt": "" }, @@ -54,6 +57,7 @@ "measuredAt": "2026-10-14T08:00:00.000Z", "source": "MANUAL", "notes": null, + "appleHealthMirror": true, "syncVersion": "", "updatedAt": "" } @@ -158,6 +162,10 @@ "cycles": { "upserts": [], "tombstones": [] + }, + "measurementMirror": { + "upserts": [], + "tombstones": [] } } }, @@ -180,6 +188,7 @@ "cycle_day_tombstones": 0, "cycle_upserts": 0, "cycle_tombstones": 0, + "measurement_mirror_upserts": 0, "cycle_module_enabled": false, "has_more": false, "cursor_present": false