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