From a8ac376cf7b11a50e297f6da1c9bbb285f0973a8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Mon, 20 Jul 2026 11:55:09 +0200 Subject: [PATCH] fix(measurements): reconcile dual identities atomically --- docs/api/openapi.yaml | 12 +- .../__tests__/aggregated-hr-wire.test.ts | 139 +++++--- .../insert-returning-arrivals.test.ts | 59 +++- .../batch/__tests__/pr-detection-hook.test.ts | 45 +++ src/app/api/measurements/batch/route.ts | 313 +++++++----------- src/lib/__tests__/idempotency.test.ts | 18 + src/lib/idempotency.ts | 10 +- .../reconcile-external-measurement.test.ts | 233 +++++++++++++ .../reconcile-external-measurement.ts | 296 +++++++++++++++++ src/lib/openapi/routes/measurements.ts | 5 +- src/lib/oura/__tests__/sync.test.ts | 77 +++-- src/lib/oura/sync.ts | 104 +++--- src/lib/polar/__tests__/sync.test.ts | 71 +++- src/lib/polar/sync.ts | 111 +++---- src/lib/whoop/__tests__/sync.test.ts | 66 +++- src/lib/whoop/sync.ts | 102 +++--- src/lib/withings/__tests__/sync-sleep.test.ts | 162 +++++---- src/lib/withings/sync-sleep.ts | 239 ++++--------- .../measurement-identity-reconcile.test.ts | 224 +++++++++++++ tests/integration/measurements-batch.test.ts | 60 ++-- 20 files changed, 1549 insertions(+), 797 deletions(-) create mode 100644 src/lib/measurements/__tests__/reconcile-external-measurement.test.ts create mode 100644 src/lib/measurements/reconcile-external-measurement.ts create mode 100644 tests/integration/measurement-identity-reconcile.test.ts diff --git a/docs/api/openapi.yaml b/docs/api/openapi.yaml index f05c28259..f91e1dd7b 100644 --- a/docs/api/openapi.yaml +++ b/docs/api/openapi.yaml @@ -17434,6 +17434,10 @@ components: type: integer minimum: 0 maximum: 9007199254740991 + failed: + type: integer + minimum: 0 + maximum: 9007199254740991 skipped: type: array items: @@ -17458,6 +17462,7 @@ components: - inserted - updated - duplicates + - failed - skipped - entries additionalProperties: false @@ -17475,9 +17480,10 @@ components: - updated - duplicate - skipped - description: "`inserted`/`duplicate` — the row landed (advance the cursor). `updated` — a `stats:` aggregate (per-day - cumulative total or hourly heart-rate bucket) overwrote an existing row. `skipped` — not stored (e.g. - unmappable identifier, out-of-range value, or `malformed_hr_bucket_id`); see `reason`." + - failed + description: "`inserted`/`duplicate` — the row landed (advance the cursor). `updated` — a `stats:` aggregate overwrote + an existing row. `skipped` — validation no-op; see `reason`. `failed` — retryable database failure that must + not advance the entry cursor; the response is marked `Cache-Control: no-store`." reason: type: string required: diff --git a/src/app/api/measurements/batch/__tests__/aggregated-hr-wire.test.ts b/src/app/api/measurements/batch/__tests__/aggregated-hr-wire.test.ts index 46fd0de7a..51de39107 100644 --- a/src/app/api/measurements/batch/__tests__/aggregated-hr-wire.test.ts +++ b/src/app/api/measurements/batch/__tests__/aggregated-hr-wire.test.ts @@ -31,6 +31,62 @@ vi.mock("@/lib/db", () => ({ }, })); +vi.mock("@/lib/measurements/reconcile-external-measurement", () => ({ + reconcileExternalMeasurement: async ( + tx: { + measurement: { + findMany: (args: unknown) => Promise< + Array<{ + id?: string; + type?: string; + source?: string; + externalId?: string | null; + }> + >; + createManyAndReturn: (args: { + data: Record; + }) => Promise>>; + updateMany: (args: { + where: Record; + data: Record; + }) => Promise; + }; + }, + input: Record & { + userId: string; + type: string; + source: string; + externalId: string; + }, + options: { exactExternalMatch?: "update" | "duplicate" } = {}, + ) => { + const existing = await tx.measurement.findMany({}); + const exact = existing.find( + (row: { type?: string; source?: string; externalId?: string | null }) => + row.type === input.type && + row.source === input.source && + row.externalId === input.externalId, + ); + if (exact && options.exactExternalMatch !== "update") { + return { status: "duplicate", row: exact }; + } + if (exact) { + await tx.measurement.updateMany({ + where: { id: exact.id }, + data: { ...input, deletedAt: null }, + }); + return { status: "updated", row: { ...exact, ...input } }; + } + const [inserted] = await tx.measurement.createManyAndReturn({ data: input }); + if (inserted) return { status: "inserted", row: inserted }; + if (options.exactExternalMatch === "update") { + await tx.measurement.updateMany({ where: {}, data: input }); + return { status: "updated", row: input }; + } + return { status: "duplicate" }; + }, +})); + vi.mock("@/lib/auth/session", () => ({ getSession: vi.fn() })); vi.mock("@/lib/auth/audit", () => ({ auditLog: vi.fn().mockResolvedValue(undefined), @@ -159,12 +215,10 @@ describe("POST /api/measurements/batch — aggregated HR wire contract (iOS #34) expect(data.entries[0].status).toBe("inserted"); // Stored as a PULSE row carrying the 10-min average value. - const createArg = vi.mocked(prisma.measurement.createManyAndReturn).mock - .calls[0][0]; - const rows = (createArg as { data: { type: string; value: number }[] }) - .data; - expect(rows[0].type).toBe("PULSE"); - expect(rows[0].value).toBe(72); + const row = vi.mocked(prisma.measurement.createManyAndReturn).mock + .calls[0]![0]!.data as { type: string; value: number }; + expect(row.type).toBe("PULSE"); + expect(row.value).toBe(72); }); it("persists per-bucket min/max on a fresh 10-min HR bucket (iOS #34 ext)", async () => { @@ -176,16 +230,15 @@ describe("POST /api/measurements/batch — aggregated HR wire contract (iOS #34) }), ); expect(res.status).toBe(200); - const createArg = vi.mocked(prisma.measurement.createManyAndReturn).mock - .calls[0][0]; - const rows = ( - createArg as { - data: { value: number; valueMin: number; valueMax: number }[]; - } - ).data; - expect(rows[0].value).toBe(72); - expect(rows[0].valueMin).toBe(58); - expect(rows[0].valueMax).toBe(96); + const row = vi.mocked(prisma.measurement.createManyAndReturn).mock + .calls[0]![0]!.data as { + value: number; + valueMin: number; + valueMax: number; + }; + expect(row.value).toBe(72); + expect(row.valueMin).toBe(58); + expect(row.valueMax).toBe(96); }); it.each([ @@ -202,21 +255,15 @@ describe("POST /api/measurements/batch — aggregated HR wire contract (iOS #34) makeRequest({ entries: [hrBucketEntry(HR_BUCKET_ID, 72, spread)] }), ); expect(res.status).toBe(200); - const createArg = vi.mocked(prisma.measurement.createManyAndReturn).mock - .calls[0][0]; - const rows = ( - createArg as { - data: { - value: number; - valueMin: number | null; - valueMax: number | null; - }[]; - } - ).data; - // The trustworthy average survives; the invalid spread is dropped. - expect(rows[0].value).toBe(72); - expect(rows[0].valueMin).toBeNull(); - expect(rows[0].valueMax).toBeNull(); + const row = vi.mocked(prisma.measurement.createManyAndReturn).mock + .calls[0]![0]!.data as { + value: number; + valueMin: number | null; + valueMax: number | null; + }; + expect(row.value).toBe(72); + expect(row.valueMin).toBeNull(); + expect(row.valueMax).toBeNull(); }, ); @@ -251,15 +298,13 @@ describe("POST /api/measurements/batch — aggregated HR wire contract (iOS #34) }), ); expect(res.status).toBe(200); - const createArg = vi.mocked(prisma.measurement.createManyAndReturn).mock - .calls[0][0]; - const rows = ( - createArg as { - data: { valueMin: number | null; valueMax: number | null }[]; - } - ).data; - expect(rows[0].valueMin).toBeNull(); - expect(rows[0].valueMax).toBeNull(); + const row = vi.mocked(prisma.measurement.createManyAndReturn).mock + .calls[0]![0]!.data as { + valueMin: number | null; + valueMax: number | null; + }; + expect(row.valueMin).toBeNull(); + expect(row.valueMax).toBeNull(); }); it("overwrites the same bucket on a re-post (updated, not duplicate)", async () => { @@ -445,12 +490,10 @@ describe("POST /api/measurements/batch — intra-batch `stats:` supersession", ( expect(res.status).toBe(200); const { data } = await readJson(res); - // Exactly ONE row reaches the insert, carrying the NEWER value. - const createArg = vi.mocked(prisma.measurement.createManyAndReturn).mock - .calls[0][0]; - const rows = (createArg as { data: { value: number }[] }).data; - expect(rows).toHaveLength(1); - expect(rows[0].value).toBe(78); + expect(prisma.measurement.createManyAndReturn).toHaveBeenCalledTimes(1); + const row = vi.mocked(prisma.measurement.createManyAndReturn).mock + .calls[0]![0]!.data as { value: number }; + expect(row.value).toBe(78); // The superseded entry is reported honestly, not as a phantom insert. expect(data.entries[0].status).toBe("duplicate"); @@ -500,9 +543,7 @@ describe("POST /api/measurements/batch — intra-batch `stats:` supersession", ( expect(res.status).toBe(200); const { data } = await readJson(res); - const createArg = vi.mocked(prisma.measurement.createManyAndReturn).mock - .calls[0][0]; - expect((createArg as { data: unknown[] }).data).toHaveLength(2); + expect(prisma.measurement.createManyAndReturn).toHaveBeenCalledTimes(2); expect(data.inserted).toBe(2); expect(data.duplicates).toBe(0); }); diff --git a/src/app/api/measurements/batch/__tests__/insert-returning-arrivals.test.ts b/src/app/api/measurements/batch/__tests__/insert-returning-arrivals.test.ts index 5f0bf6752..d376658b3 100644 --- a/src/app/api/measurements/batch/__tests__/insert-returning-arrivals.test.ts +++ b/src/app/api/measurements/batch/__tests__/insert-returning-arrivals.test.ts @@ -15,6 +15,58 @@ vi.mock("@/lib/db", () => ({ }), }, })); +vi.mock("@/lib/measurements/reconcile-external-measurement", () => ({ + reconcileExternalMeasurement: async ( + tx: { + measurement: { + findMany: (args: unknown) => Promise< + Array<{ + id?: string; + type?: string; + source?: string; + externalId?: string | null; + }> + >; + createManyAndReturn: (args: { + data: Record; + }) => Promise>>; + updateMany: (args: { + where: Record; + data: Record; + }) => Promise; + }; + }, + input: Record & { + userId: string; + type: string; + source: string; + externalId: string; + }, + options: { exactExternalMatch?: "update" | "duplicate" } = {}, + ) => { + const existing = await tx.measurement.findMany({}); + const exact = existing.find( + (row: { type?: string; source?: string; externalId?: string | null }) => + row.type === input.type && + row.source === input.source && + row.externalId === input.externalId, + ); + if (exact && options.exactExternalMatch !== "update") { + return { status: "duplicate", row: exact }; + } + if (exact) { + await tx.measurement.updateMany({ where: { id: exact.id }, data: input }); + return { status: "updated", row: { ...exact, ...input } }; + } + const [inserted] = await tx.measurement.createManyAndReturn({ data: input }); + if (inserted) return { status: "inserted", row: inserted }; + if (options.exactExternalMatch === "update") { + await tx.measurement.updateMany({ where: {}, data: input }); + return { status: "updated", row: input }; + } + return { status: "duplicate" }; + }, +})); vi.mock("@/lib/auth/session", () => ({ getSession: vi.fn() })); vi.mock("@/lib/auth/audit", () => ({ auditLog: vi.fn().mockResolvedValue(undefined), @@ -190,12 +242,7 @@ describe("POST /api/measurements/batch — exact INSERT RETURNING identity", () "duplicate", "duplicate", ]); - expect(prisma.measurement.createManyAndReturn).toHaveBeenCalledWith( - expect.objectContaining({ - skipDuplicates: true, - select: { type: true, source: true, externalId: true }, - }), - ); + expect(prisma.measurement.createManyAndReturn).toHaveBeenCalledTimes(4); expect(maybeEnqueueMorningRefresh).toHaveBeenCalledWith("user-1", [ new Date(insertedSleepAt), ]); diff --git a/src/app/api/measurements/batch/__tests__/pr-detection-hook.test.ts b/src/app/api/measurements/batch/__tests__/pr-detection-hook.test.ts index eeeb051ea..ab9ff6113 100644 --- a/src/app/api/measurements/batch/__tests__/pr-detection-hook.test.ts +++ b/src/app/api/measurements/batch/__tests__/pr-detection-hook.test.ts @@ -8,6 +8,8 @@ import { describe, expect, it, vi, beforeEach } from "vitest"; import { NextRequest } from "next/server"; +const reconcileOverrideMock = vi.hoisted(() => vi.fn()); + vi.mock("@/lib/db", () => ({ prisma: { measurement: { @@ -23,6 +25,26 @@ vi.mock("@/lib/db", () => ({ }, })); +vi.mock("@/lib/measurements/reconcile-external-measurement", () => ({ + reconcileExternalMeasurement: async ( + tx: { + measurement: { + createManyAndReturn: (args: { + data: Record; + }) => Promise>>; + }; + }, + input: Record, + ) => { + const override = await reconcileOverrideMock(tx, input); + if (override !== undefined) return override; + const [inserted] = await tx.measurement.createManyAndReturn({ data: input }); + return inserted + ? { status: "inserted", row: inserted } + : { status: "duplicate" }; + }, +})); + vi.mock("@/lib/auth/session", () => ({ getSession: vi.fn() })); vi.mock("@/lib/auth/audit", () => ({ @@ -118,6 +140,29 @@ describe("POST /api/measurements/batch — PR detection enqueue (v1.4.25 W16c)", }); }); + it("surfaces a hard reconciliation failure without checkpointing it", async () => { + reconcileOverrideMock.mockResolvedValueOnce({ + status: "failed", + error: { message: "write rejected", code: "P2002" }, + }); + + const res = await POST(makeRequest({ entries: [validStepEntry("ext-a")] })); + expect(res.status).toBe(200); + expect(res.headers.get("Cache-Control")).toContain("no-store"); + expect(await res.json()).toMatchObject({ + data: { + processed: 1, + inserted: 0, + updated: 0, + duplicates: 0, + failed: 1, + entries: [{ index: 0, status: "failed", reason: "P2002" }], + }, + error: null, + }); + expect(enqueuePrDetection).not.toHaveBeenCalled(); + }); + it("sets silent=true once the batch crosses the 50-entry threshold", async () => { const big = Array.from({ length: 51 }, (_, i) => validStepEntry(`ext-${i}`), diff --git a/src/app/api/measurements/batch/route.ts b/src/app/api/measurements/batch/route.ts index 13d259ceb..25013650d 100644 --- a/src/app/api/measurements/batch/route.ts +++ b/src/app/api/measurements/batch/route.ts @@ -60,6 +60,7 @@ import { oppositeMergeSource, type MergeCandidate, } from "@/lib/measurements/cross-source-merge"; +import { reconcileExternalMeasurement } from "@/lib/measurements/reconcile-external-measurement"; import { validateMeasurementRange } from "@/lib/validations/measurement"; import { deviceTypeEnum } from "@/lib/validations/source-priority"; import { checkRateLimit } from "@/lib/rate-limit"; @@ -173,11 +174,15 @@ type BatchEntry = z.infer; * we already have, the server overwrites the row's `value` rather than * dropping the new payload as a duplicate. Sample-class rows (every * other HK metric) keep the strict `duplicate` semantics because each - * sample is a canonical, immutable reading. `skipped` is the only case - * where the client may want to surface a diagnostic — typically because - * Apple introduced a new identifier the server doesn't know about yet. + * sample is a canonical, immutable reading. `skipped` is a validation no-op; + * `failed` is a hard database verdict and must not be checkpointed as landed. */ -type EntryStatus = "inserted" | "updated" | "duplicate" | "skipped"; +type EntryStatus = + | "inserted" + | "updated" + | "duplicate" + | "skipped" + | "failed"; // v1.5.0 issue #213 — `stats:*` externalIds carry per-day cumulative // totals that the iOS HealthKit observer re-posts as the day progresses. @@ -372,47 +377,20 @@ async function postBatch(request: NextRequest): Promise { let insertedCount = 0; let updatedCount = 0; let duplicateCount = 0; + let failedCount = 0; // v1.11.4 (iOS #2) — subset of `duplicateCount` collapsed by the // MANUAL↔APPLE_HEALTH same-reading merge (as opposed to a plain // composite-key duplicate). Surfaced as a dedicated wide-event count so // an operator can see how often the standalone-pair mirror collapses. let crossSourceMergedCount = 0; - // Exact prepared rows returned by INSERT ... RETURNING. This is the only - // source for post-write arrival and sleep side effects: preflight checks can - // lose a unique-key race after they run. + // Only rows the reconciler actually inserted drive arrival side effects. const insertedPrepared: Prepared[] = []; + const writtenIdentities: Array<{ + type: MeasurementType; + measuredAt: Date; + }> = []; if (prepared.length > 0) { - // Pre-flight duplicate detection classifies rows already present before - // this request. The later INSERT ... RETURNING remains authoritative when - // a competing request wins the unique-key race after this probe. - // v1.8.6 W6 — `source` is now part of the dedup key, so the lookup - // matches on (type, source, externalId) per incoming row rather than - // a single hardcoded `APPLE_HEALTH`. A `MANUAL` row and an - // `APPLE_HEALTH` row sharing an externalId are distinct, so each must - // carry its source through the existence probe and the composite key. - const incomingKeys = prepared.map((p) => ({ - type: p.row.type as Prisma.MeasurementCreateManyInput["type"], - source: p.row.source as Prisma.MeasurementCreateManyInput["source"], - externalId: p.row.externalId as string, - })); - - const existing = await prisma.measurement.findMany({ - where: { - userId: user.id, - OR: incomingKeys.map((k) => ({ - type: k.type, - source: k.source, - externalId: k.externalId, - })), - }, - select: { type: true, source: true, externalId: true }, - }); - - const existingSet = new Set( - existing.map((row) => `${row.type}::${row.source}::${row.externalId}`), - ); - // v1.11.4 (iOS #2) — same-reading merge for the MANUAL ↔ APPLE_HEALTH // mirror duplicate. When the iOS app logs a manual reading offline it // mirrors that reading into HealthKit; on pairing, adopt-on-pair @@ -485,20 +463,8 @@ async function postBatch(request: NextRequest): Promise { return false; } - // Intra-batch `stats:` supersession. Two entries carrying the same - // (type, source, externalId) `stats:` key inside ONE batch both miss - // `existingSet` — neither is in the DB yet — so the overwrite branch - // never runs, both land in `toInsert`, and - // `createMany({ skipDuplicates: true })` keeps whichever row the unique - // index saw first. The OLDER snapshot of a per-day cumulative total (or - // of a still-filling ten-minute heart-rate bucket) therefore won and the - // newer, larger figure was dropped silently: both entries reported - // `inserted`, the counters still summed, and the client checkpointed - // past both. `stats:` rows carry overwrite semantics by contract — a - // re-post replaces value / unit / measuredAt — so within one batch the - // LAST entry for a key is the authoritative one. Resolve it here and - // mark the superseded earlier entries `duplicate`, the same status the - // client already checkpoints past for a composite-key duplicate. + // `stats:` rows carry overwrite semantics and the last entry in one batch + // is authoritative. Earlier snapshots are safe duplicates. const lastStatsIndexByKey = new Map(); for (const p of prepared) { if (!isStatsExternalId(p.row.externalId as string)) continue; @@ -508,57 +474,96 @@ async function postBatch(request: NextRequest): Promise { ); } - const toInsert: Prepared[] = []; - const insertKeys = new Set(); - const toOverwrite: Prepared[] = []; for (const p of prepared) { - const key = `${p.row.type}::${p.row.source}::${p.row.externalId}`; - if ( - isStatsExternalId(p.row.externalId as string) && - lastStatsIndexByKey.get(key) !== p.index - ) { - results[p.index] = { - index: p.index, - status: "duplicate", - reason: "superseded_in_batch", - }; - duplicateCount += 1; - continue; - } - if (existingSet.has(key) || insertKeys.has(key)) { - // v1.5.0 issue #213 — per-day cumulative `stats:*` rows are - // intentionally overwritten on a re-post so today's tile reflects - // the latest HealthKit total instead of freezing at the - // first-sync value. Sample-class rows keep the immutable - // duplicate contract. - if (isStatsExternalId(p.row.externalId as string)) { - toOverwrite.push(p); - } else { - results[p.index] = { index: p.index, status: "duplicate" }; - duplicateCount += 1; + const externalId = p.row.externalId as string; + const statsRow = isStatsExternalId(externalId); + const key = `${p.row.type}::${p.row.source}::${externalId}`; + if (statsRow && lastStatsIndexByKey.get(key) !== p.index) { + results[p.index] = { + index: p.index, + status: "duplicate", + reason: "superseded_in_batch", + }; + duplicateCount++; + continue; } - } else if ( - !isStatsExternalId(p.row.externalId as string) && - hasSameReadingSibling(p) - ) { - // v1.11.4 (iOS #2) — cross-source same-reading collapse. The - // physical reading already exists under the opposite client - // source; drop this mirror copy. Status stays `duplicate` so the - // iOS sync cursor checkpoints past it exactly as it does for a - // composite-key duplicate; the `reason` distinguishes it for ops. - results[p.index] = { - index: p.index, - status: "duplicate", - reason: "cross_source_merge", - }; - duplicateCount += 1; - crossSourceMergedCount += 1; - } else { - results[p.index] = { index: p.index, status: "inserted" }; - toInsert.push(p); - insertKeys.add(key); - // Track this row so a same-reading sibling later in the SAME - // batch collapses against it. + + if (!statsRow && hasSameReadingSibling(p)) { + results[p.index] = { + index: p.index, + status: "duplicate", + reason: "cross_source_merge", + }; + duplicateCount++; + crossSourceMergedCount++; + continue; + } + + const verdict = await prisma.$transaction((tx) => + reconcileExternalMeasurement( + tx, + { + userId: user.id, + type: p.row.type as MeasurementType, + value: p.row.value as number, + valueMin: p.row.valueMin as number | null, + valueMax: p.row.valueMax as number | null, + unit: p.row.unit as string, + source: + p.row.source as Prisma.MeasurementUncheckedCreateInput["source"], + measuredAt: p.row.measuredAt as Date, + externalId, + externalSourceVersion: p.row + .externalSourceVersion as string | null, + glucoseContext: + p.row.glucoseContext as Prisma.MeasurementUncheckedCreateInput["glucoseContext"], + sleepStage: + p.row.sleepStage as Prisma.MeasurementUncheckedCreateInput["sleepStage"], + rhythmClassification: + p.row + .rhythmClassification as Prisma.MeasurementUncheckedCreateInput["rhythmClassification"], + deviceType: p.row.deviceType as string | null, + }, + { exactExternalMatch: statsRow ? "update" : "duplicate" }, + ), + ); + + switch (verdict.status) { + case "inserted": + results[p.index] = { index: p.index, status: "inserted" }; + insertedCount++; + insertedPrepared.push(p); + writtenIdentities.push({ + type: p.row.type as MeasurementType, + measuredAt: p.row.measuredAt as Date, + }); + break; + case "updated": + case "resurrected": + results[p.index] = { index: p.index, status: "updated" }; + updatedCount++; + writtenIdentities.push( + { + type: p.row.type as MeasurementType, + measuredAt: p.row.measuredAt as Date, + }, + ...(verdict.dirtyIdentities ?? []), + ); + break; + case "duplicate": + results[p.index] = { index: p.index, status: "duplicate" }; + duplicateCount++; + break; + case "failed": + results[p.index] = { + index: p.index, + status: "failed", + reason: verdict.error.code ?? "write_failed", + }; + failedCount++; + continue; + } + if (isMergeableSource(p.row.source as string)) { inBatchCandidates.push({ type: p.row.type as MeasurementType, @@ -567,95 +572,6 @@ async function postBatch(request: NextRequest): Promise { measuredAt: p.row.measuredAt as Date, }); } - } - } - - if (toOverwrite.length > 0) { - await prisma.$transaction(async (tx) => { - for (const p of toOverwrite) { - await tx.measurement.updateMany({ - where: { - userId: user.id, - // v1.8.6 W6 — scope the overwrite to the row's own source - // so a `stats:*` re-post only touches the matching - // (type, source, externalId) row. - source: p.row - .source as Prisma.MeasurementCreateManyInput["source"], - type: p.row.type as MeasurementType, - externalId: p.row.externalId as string, - }, - data: { - value: p.row.value as number, - // v1.19.2 (iOS #34 extension) — overwrite the per-bucket - // spread alongside `value` so a within-hour re-post (the - // running mean + range shift as more samples land) replaces - // both. Null for every non-HR-bucket `stats:` overwrite. - valueMin: p.row.valueMin as number | null, - valueMax: p.row.valueMax as number | null, - unit: p.row.unit as string, - measuredAt: p.row.measuredAt as Date, - externalSourceVersion: p.row.externalSourceVersion as - string | null, - deviceType: p.row.deviceType as Prepared["row"]["deviceType"], - sleepStage: p.row.sleepStage as Prepared["row"]["sleepStage"], - // No-op on a live row, a deliberate RESURRECTION on a - // tombstoned one — the observer re-posts the day's canonical - // total, so a `stats:` day-total row follows the - // source-owned resurrect rule (mirrors Google / Fitbit). - // SAMPLE-grain rows never reach this branch: their - // tombstones stick (the iOS LWW reconciler propagates - // HealthKit deletions and reports them as `duplicate`). - deletedAt: null, - }, - }); - results[p.index] = { index: p.index, status: "updated" }; - updatedCount += 1; - } - }); - } - - if (toInsert.length > 0) { - // PostgreSQL RETURNING identifies the exact winner when another request - // races the same composite unique key. An aggregate insert count can say - // how many rows landed but not which prepared rows won, which would make - // arrival and morning-refresh side effects attach to arbitrary rows. - const CHUNK = 200; - const insertedIndexes = new Set(); - await prisma.$transaction(async (tx) => { - for (let i = 0; i < toInsert.length; i += CHUNK) { - const chunk = toInsert.slice(i, i + CHUNK); - const pendingByKey = new Map(); - for (const p of chunk) { - pendingByKey.set( - `${p.row.type}::${p.row.source}::${p.row.externalId}`, - p, - ); - } - - const inserted = await tx.measurement.createManyAndReturn({ - data: chunk.map((p) => p.row), - skipDuplicates: true, - select: { type: true, source: true, externalId: true }, - }); - // Count only returned identities that map to an attempted row, so - // aggregate counts and per-entry statuses share one source of truth. - for (const row of inserted) { - const preparedRow = pendingByKey.get( - `${row.type}::${row.source}::${row.externalId}`, - ); - if (!preparedRow) continue; - insertedIndexes.add(preparedRow.index); - insertedCount += 1; - insertedPrepared.push(preparedRow); - } - } - }); - - for (const p of toInsert) { - if (insertedIndexes.has(p.index)) continue; - results[p.index] = { index: p.index, status: "duplicate" }; - duplicateCount += 1; - } } } @@ -680,6 +596,7 @@ async function postBatch(request: NextRequest): Promise { updated: updatedCount, duplicates: duplicateCount, skipped: skipped.length, + failed: failedCount, }, }); @@ -760,6 +677,7 @@ async function postBatch(request: NextRequest): Promise { updated: updatedCount, duplicates: duplicateCount, skipped: skipped.length, + failed: failedCount, }, }); @@ -784,15 +702,7 @@ async function postBatch(request: NextRequest): Promise { // by day so a 500-row Apple Health batch fires only one recompute per // type/day pair. Best-effort — a populator hiccup never fails ingest. try { - const insertedKeys = collapseToTypeDayKeys( - prepared.map((p) => ({ - type: p.row.type as MeasurementType, - measuredAt: - p.row.measuredAt instanceof Date - ? p.row.measuredAt - : new Date(p.row.measuredAt as string), - })), - ); + const insertedKeys = collapseToTypeDayKeys(writtenIdentities); for (const k of insertedKeys) { await recomputeBucketsForMeasurement(user.id, k.type, k.measuredAt); } @@ -851,12 +761,17 @@ async function postBatch(request: NextRequest): Promise { } } - return apiSuccess({ + const response = apiSuccess({ processed: entries.length, inserted: insertedCount, updated: updatedCount, duplicates: duplicateCount, + failed: failedCount, skipped, entries: results, }); + if (failedCount > 0) { + response.headers.set("Cache-Control", "private, no-store"); + } + return response; } diff --git a/src/lib/__tests__/idempotency.test.ts b/src/lib/__tests__/idempotency.test.ts index 15c10ca01..5d15ba7db 100644 --- a/src/lib/__tests__/idempotency.test.ts +++ b/src/lib/__tests__/idempotency.test.ts @@ -145,6 +145,24 @@ describe("withIdempotency", () => { expect(handler).toHaveBeenCalledTimes(1); // not called again }); + it("does not cache a successful response marked no-store", async () => { + const handler = vi.fn(async () => + NextResponse.json( + { data: { failed: 1 }, error: null }, + { headers: { "Cache-Control": "private, no-store" } }, + ), + ); + const wrapped = withIdempotency<[NextRequest]>(handler, async () => "u-1"); + + const response = await wrapped( + makeRequest("POST", { "idempotency-key": "abc-12345678" }), + ); + + expect(response.status).toBe(200); + expect(prisma.idempotencyKey.updateMany).not.toHaveBeenCalled(); + expect(prisma.idempotencyKey.deleteMany).toHaveBeenCalled(); + }); + it("ignores expired cache rows and re-runs the handler", async () => { const handler = vi.fn(async () => NextResponse.json({ data: "fresh", error: null }, { status: 200 }), diff --git a/src/lib/idempotency.ts b/src/lib/idempotency.ts index a10155bc2..7c5dbb220 100644 --- a/src/lib/idempotency.ts +++ b/src/lib/idempotency.ts @@ -411,7 +411,11 @@ export function withIdempotency< throw err; } - if (isCachableStatus(response.status)) { + const noStore = response.headers + .get("Cache-Control") + ?.split(",") + .some((directive) => directive.trim().toLowerCase() === "no-store"); + if (isCachableStatus(response.status) && !noStore) { // Defence-in-depth: never persist a body that carries a freshly-issued // bearer / refresh token or a third-party AI provider key. Auth and // settings routes shouldn't be wrapped in withIdempotency to begin @@ -437,8 +441,8 @@ export function withIdempotency< await releaseClaim(ctx); } } else { - // Non-cachable status (401/403/408/429/5xx) — release the claim so a - // retry gets a fresh attempt rather than a stuck pending row. + // Non-cachable status or an explicit `no-store` response releases the + // claim so a retry gets a fresh attempt. await releaseClaim(ctx); } diff --git a/src/lib/measurements/__tests__/reconcile-external-measurement.test.ts b/src/lib/measurements/__tests__/reconcile-external-measurement.test.ts new file mode 100644 index 000000000..41ca58bb2 --- /dev/null +++ b/src/lib/measurements/__tests__/reconcile-external-measurement.test.ts @@ -0,0 +1,233 @@ +import { describe, expect, it, vi } from "vitest"; + +import type { Prisma } from "@/generated/prisma/client"; +import { + reconcileExternalMeasurement, + type ExternalMeasurementWrite, +} from "../reconcile-external-measurement"; + +function desired( + overrides: Partial = {}, +): ExternalMeasurementWrite { + return { + userId: "user-1", + type: "SLEEP_DURATION", + value: 42, + unit: "minutes", + source: "OURA", + measuredAt: new Date("2026-07-20T06:30:00.000Z"), + sleepStage: "DEEP", + externalId: "sleep:new", + externalSourceVersion: "v2", + deviceType: "ring", + ...overrides, + }; +} + +function row( + overrides: Record = {}, +): Record { + return { + id: "row-1", + userId: "user-1", + type: "SLEEP_DURATION", + value: 30, + valueMin: null, + valueMax: null, + unit: "minutes", + source: "OURA", + measuredAt: new Date("2026-07-20T06:00:00.000Z"), + notes: null, + notesEncrypted: null, + externalId: "sleep:new", + externalSourceVersion: null, + glucoseContext: null, + sleepStage: "DEEP", + rhythmClassification: null, + deviceType: "ring", + syncVersion: 1, + deletedAt: null, + createdAt: new Date("2026-07-20T07:00:00.000Z"), + updatedAt: new Date("2026-07-20T07:00:00.000Z"), + ...overrides, + }; +} + +function transaction(rows: Record[]) { + const measurement = { + findMany: vi.fn().mockResolvedValue(rows), + create: vi.fn(), + update: vi.fn(), + deleteMany: vi.fn().mockResolvedValue({ count: 1 }), + findFirst: vi.fn().mockResolvedValue(null), + }; + const tx = { + measurement, + $queryRaw: vi.fn().mockResolvedValue([]), + $executeRawUnsafe: vi.fn().mockResolvedValue(0), + } as unknown as Prisma.TransactionClient; + return { tx, measurement }; +} + +describe("reconcileExternalMeasurement", () => { + it("keeps the external-id row canonical when its desired natural key is occupied", async () => { + const external = row({ id: "external-row" }); + const natural = row({ + id: "natural-row", + externalId: "sleep:legacy", + measuredAt: new Date("2026-07-20T06:30:00.000Z"), + }); + const { tx, measurement } = transaction([external, natural]); + measurement.update.mockResolvedValue( + row({ + id: "external-row", + value: 42, + measuredAt: new Date("2026-07-20T06:30:00.000Z"), + externalSourceVersion: "v2", + }), + ); + + const verdict = await reconcileExternalMeasurement(tx, desired()); + expect(verdict).toMatchObject({ + status: "updated", + row: { id: "external-row", externalId: "sleep:new" }, + retiredCollisionId: "natural-row", + }); + expect(measurement.deleteMany).not.toHaveBeenCalled(); + expect(measurement.update).toHaveBeenLastCalledWith( + expect.objectContaining({ + where: { id: "external-row" }, + data: expect.objectContaining({ + value: 42, + measuredAt: new Date("2026-07-20T06:30:00.000Z"), + externalId: "sleep:new", + externalSourceVersion: "v2", + deviceType: "ring", + deletedAt: null, + }), + }), + ); + }); + + it("retires the redundant collision as a durable tombstone", async () => { + const external = row({ id: "external-row" }); + const natural = row({ + id: "natural-row", + externalId: "sleep:legacy", + measuredAt: new Date("2026-07-20T06:30:00.000Z"), + }); + const { tx, measurement } = transaction([external, natural]); + measurement.update + .mockResolvedValueOnce( + row({ + id: "natural-row", + externalId: "sleep:legacy", + measuredAt: new Date(0), + deletedAt: expect.any(Date), + }), + ) + .mockResolvedValueOnce( + row({ + id: "external-row", + value: 42, + measuredAt: new Date("2026-07-20T06:30:00.000Z"), + }), + ); + + const verdict = await reconcileExternalMeasurement(tx, desired()); + + expect(verdict).toMatchObject({ + status: "updated", + retiredCollisionId: "natural-row", + }); + expect(measurement.deleteMany).not.toHaveBeenCalled(); + expect(measurement.update).toHaveBeenNthCalledWith( + 1, + expect.objectContaining({ + where: { id: "natural-row" }, + data: expect.objectContaining({ + measuredAt: new Date(0), + }), + }), + ); + }); + + it("resurrects a tombstoned natural-key row and adopts the desired external identity", async () => { + const tombstone = row({ + id: "tombstone-row", + externalId: "sleep:legacy", + measuredAt: new Date("2026-07-20T06:30:00.000Z"), + deletedAt: new Date("2026-07-20T08:00:00.000Z"), + }); + const { tx, measurement } = transaction([tombstone]); + measurement.update.mockResolvedValue( + row({ + id: "tombstone-row", + value: 42, + measuredAt: new Date("2026-07-20T06:30:00.000Z"), + deletedAt: null, + }), + ); + + const verdict = await reconcileExternalMeasurement(tx, desired()); + + expect(verdict).toMatchObject({ + status: "resurrected", + row: { id: "tombstone-row", externalId: "sleep:new" }, + }); + }); + + it("returns duplicate for an immutable exact external/natural match", async () => { + const existing = row({ + id: "exact-row", + measuredAt: new Date("2026-07-20T06:30:00.000Z"), + }); + const { tx, measurement } = transaction([existing]); + + const verdict = await reconcileExternalMeasurement(tx, desired(), { + exactExternalMatch: "duplicate", + }); + + expect(verdict).toMatchObject({ + status: "duplicate", + row: { id: "exact-row" }, + }); + expect(measurement.update).not.toHaveBeenCalled(); + expect(measurement.deleteMany).not.toHaveBeenCalled(); + }); + + it("keeps an immutable exact-match tombstone deleted", async () => { + const tombstone = row({ + id: "deleted-sample", + measuredAt: new Date("2026-07-20T06:30:00.000Z"), + deletedAt: new Date("2026-07-20T08:00:00.000Z"), + }); + const { tx, measurement } = transaction([tombstone]); + + const verdict = await reconcileExternalMeasurement(tx, desired(), { + exactExternalMatch: "duplicate", + }); + + expect(verdict.status).toBe("duplicate"); + expect(measurement.update).not.toHaveBeenCalled(); + }); + + it("rolls back its savepoint and returns failed for a non-benign write error", async () => { + const existing = row({ + id: "exact-row", + measuredAt: new Date("2026-07-20T06:30:00.000Z"), + }); + const { tx, measurement } = transaction([existing]); + measurement.update.mockRejectedValue(new Error("connection reset")); + + const verdict = await reconcileExternalMeasurement(tx, desired()); + + expect(verdict).toEqual({ + status: "failed", + error: { message: "connection reset" }, + }); + expect(tx.$executeRawUnsafe).toHaveBeenCalledWith( + "ROLLBACK TO SAVEPOINT measurement_identity_reconcile", + ); + }); +}); diff --git a/src/lib/measurements/reconcile-external-measurement.ts b/src/lib/measurements/reconcile-external-measurement.ts new file mode 100644 index 000000000..95ed27a51 --- /dev/null +++ b/src/lib/measurements/reconcile-external-measurement.ts @@ -0,0 +1,296 @@ +import { + Prisma, + type Measurement, + type Prisma as PrismaTypes, +} from "@/generated/prisma/client"; + +const SAVEPOINT = "measurement_identity_reconcile"; + +const rowSelect = { + id: true, + userId: true, + type: true, + source: true, + measuredAt: true, + sleepStage: true, + externalId: true, + deletedAt: true, +} satisfies PrismaTypes.MeasurementSelect; + +export type ExternalMeasurementWrite = Omit< + PrismaTypes.MeasurementUncheckedCreateInput, + | "id" + | "createdAt" + | "updatedAt" + | "syncVersion" + | "deletedAt" + | "externalId" + | "measuredAt" +> & { + externalId: string; + measuredAt: Date; +}; + +export type ReconciledMeasurementRow = Pick< + Measurement, + "id" | "type" | "measuredAt" | "externalId" +>; + +export type DirtyMeasurementIdentity = Pick< + Measurement, + "type" | "measuredAt" +>; + +export type MeasurementReconciliationVerdict = + | { + status: "inserted" | "updated" | "resurrected" | "duplicate"; + row: ReconciledMeasurementRow; + retiredCollisionId?: string; + dirtyIdentities?: DirtyMeasurementIdentity[]; + } + | { + status: "failed"; + error: { message: string; code?: string }; + }; + +export type FailedMeasurementReconciliation = Extract< + MeasurementReconciliationVerdict, + { status: "failed" } +>; + +export class MeasurementReconciliationError extends Error { + readonly verdict: FailedMeasurementReconciliation; + + constructor(verdict: FailedMeasurementReconciliation) { + super(verdict.error.message); + this.name = "MeasurementReconciliationError"; + this.verdict = verdict; + } +} + +export interface MeasurementReconciliationOptions { + /** + * Provider re-scores overwrite an exact identity match. Immutable sample + * uploads instead report an exact live match as a duplicate. Identity moves, + * collision merges, and tombstone resurrection always reconcile. + */ + exactExternalMatch?: "update" | "duplicate"; +} + +function errorVerdict(err: unknown): MeasurementReconciliationVerdict { + const message = err instanceof Error ? err.message : String(err); + if (err instanceof Prisma.PrismaClientKnownRequestError) { + return { status: "failed", error: { message, code: err.code } }; + } + return { status: "failed", error: { message } }; +} + +function sameNaturalIdentity( + row: Pick, + desired: ExternalMeasurementWrite, +): boolean { + return ( + row.measuredAt.getTime() === desired.measuredAt.getTime() && + row.sleepStage === (desired.sleepStage ?? null) + ); +} + +function lockKeys(desired: ExternalMeasurementWrite): string[] { + const prefix = [desired.userId, desired.type, desired.source]; + return [ + JSON.stringify([...prefix, "external", desired.externalId]), + JSON.stringify([ + ...prefix, + "natural", + desired.measuredAt.toISOString(), + desired.sleepStage ?? null, + ]), + ].sort(); +} + +async function lockUser( + tx: PrismaTypes.TransactionClient, + userId: string, +): Promise { + const key = JSON.stringify(["measurement-reconcile-user", userId]); + await tx.$queryRaw( + Prisma.sql` + SELECT 1::int AS "locked" + FROM ( + SELECT pg_advisory_xact_lock(hashtextextended(${key}, 0)) + ) AS user_lock + `, + ); +} + +async function lockAndReloadCandidates( + tx: PrismaTypes.TransactionClient, + desired: ExternalMeasurementWrite, +): Promise>> { + const where = { + userId: desired.userId, + OR: [ + { + type: desired.type, + source: desired.source, + externalId: desired.externalId, + }, + { + type: desired.type, + source: desired.source, + measuredAt: desired.measuredAt, + sleepStage: desired.sleepStage ?? null, + }, + ], + } satisfies PrismaTypes.MeasurementWhereInput; + const candidates = await tx.measurement.findMany({ + where, + select: rowSelect, + }); + if (candidates.length > 0) { + await tx.$queryRaw( + Prisma.sql` + SELECT "id" + FROM "measurements" + WHERE "id" IN (${Prisma.join(candidates.map((row) => row.id))}) + FOR UPDATE + `, + ); + } + return tx.measurement.findMany({ where, select: rowSelect }); +} + +async function retireCollision( + tx: PrismaTypes.TransactionClient, + row: PrismaTypes.MeasurementGetPayload<{ select: typeof rowSelect }>, +): Promise { + let measuredAt = new Date(0); + while ( + await tx.measurement.findFirst({ + where: { + id: { not: row.id }, + userId: row.userId, + type: row.type, + source: row.source, + measuredAt, + sleepStage: row.sleepStage, + }, + select: { id: true }, + }) + ) { + measuredAt = new Date(measuredAt.getTime() + 1); + } + await tx.measurement.update({ + where: { id: row.id }, + data: { + measuredAt, + deletedAt: new Date(), + syncVersion: { increment: 1 }, + }, + select: rowSelect, + }); +} + +/** + * Atomically reconcile the two full unique identities held by Measurement. + * + * The caller supplies an interactive Prisma transaction. A transaction-local + * advisory lock serializes cooperative writers sharing either identity, and a + * savepoint turns a non-benign database error into a structured `failed` + * verdict without poisoning the caller's surrounding transaction. + */ +export async function reconcileExternalMeasurement( + tx: PrismaTypes.TransactionClient, + desired: ExternalMeasurementWrite, + options: MeasurementReconciliationOptions = {}, +): Promise { + await tx.$executeRawUnsafe(`SAVEPOINT ${SAVEPOINT}`); + + try { + await lockUser(tx, desired.userId); + for (const key of lockKeys(desired)) { + await tx.$queryRaw( + Prisma.sql` + SELECT 1::int AS "locked" + FROM ( + SELECT pg_advisory_xact_lock(hashtextextended(${key}, 0)) + ) AS identity_lock + `, + ); + } + + const candidates = await lockAndReloadCandidates(tx, desired); + + const externalHit = candidates.find( + (row) => row.externalId === desired.externalId, + ); + const naturalHit = candidates.find((row) => + sameNaturalIdentity(row, desired), + ); + + if (!externalHit && !naturalHit) { + const created = await tx.measurement.create({ + data: { + ...desired, + sleepStage: desired.sleepStage ?? null, + deletedAt: null, + }, + select: rowSelect, + }); + await tx.$executeRawUnsafe(`RELEASE SAVEPOINT ${SAVEPOINT}`); + return { status: "inserted", row: created }; + } + + const canonical = externalHit ?? naturalHit!; + const redundant = + externalHit && naturalHit && externalHit.id !== naturalHit.id + ? naturalHit + : undefined; + const exactExternalMatch = externalHit?.id === naturalHit?.id; + + if ( + options.exactExternalMatch === "duplicate" && + exactExternalMatch && + !redundant + ) { + await tx.$executeRawUnsafe(`RELEASE SAVEPOINT ${SAVEPOINT}`); + return { status: "duplicate", row: canonical }; + } + + if (redundant) { + await retireCollision(tx, redundant); + } + + const { userId: _ownerId, ...mutable } = desired; + const updated = await tx.measurement.update({ + where: { id: canonical.id }, + data: { + ...mutable, + sleepStage: desired.sleepStage ?? null, + deletedAt: null, + syncVersion: { increment: 1 }, + }, + select: rowSelect, + }); + await tx.$executeRawUnsafe(`RELEASE SAVEPOINT ${SAVEPOINT}`); + return { + status: canonical.deletedAt === null ? "updated" : "resurrected", + row: updated, + dirtyIdentities: [ + { type: canonical.type, measuredAt: canonical.measuredAt }, + ...(redundant + ? [{ type: redundant.type, measuredAt: redundant.measuredAt }] + : []), + ], + ...(redundant ? { retiredCollisionId: redundant.id } : {}), + }; + } catch (err) { + try { + await tx.$executeRawUnsafe(`ROLLBACK TO SAVEPOINT ${SAVEPOINT}`); + await tx.$executeRawUnsafe(`RELEASE SAVEPOINT ${SAVEPOINT}`); + } catch { + // The original database error is the actionable failure verdict. + } + return errorVerdict(err); + } +} diff --git a/src/lib/openapi/routes/measurements.ts b/src/lib/openapi/routes/measurements.ts index 23ad8f028..b02c6c217 100644 --- a/src/lib/openapi/routes/measurements.ts +++ b/src/lib/openapi/routes/measurements.ts @@ -121,9 +121,9 @@ const batchEntryResult = z .object({ index: z.number().int().nonnegative(), status: z - .enum(["inserted", "updated", "duplicate", "skipped"]) + .enum(["inserted", "updated", "duplicate", "skipped", "failed"]) .describe( - "`inserted`/`duplicate` — the row landed (advance the cursor). `updated` — a `stats:` aggregate (per-day cumulative total or hourly heart-rate bucket) overwrote an existing row. `skipped` — not stored (e.g. unmappable identifier, out-of-range value, or `malformed_hr_bucket_id`); see `reason`.", + "`inserted`/`duplicate` — the row landed (advance the cursor). `updated` — a `stats:` aggregate overwrote an existing row. `skipped` — validation no-op; see `reason`. `failed` — retryable database failure that must not advance the entry cursor; the response is marked `Cache-Control: no-store`.", ), reason: z.string().optional(), }) @@ -135,6 +135,7 @@ const batchResponse = z inserted: z.number().int().nonnegative(), updated: z.number().int().nonnegative(), duplicates: z.number().int().nonnegative(), + failed: z.number().int().nonnegative(), skipped: z.array( z.object({ index: z.number().int().nonnegative(), diff --git a/src/lib/oura/__tests__/sync.test.ts b/src/lib/oura/__tests__/sync.test.ts index 8c3a77fce..af6578fee 100644 --- a/src/lib/oura/__tests__/sync.test.ts +++ b/src/lib/oura/__tests__/sync.test.ts @@ -23,6 +23,8 @@ const { recordFailureMock, recomputeMock, invalidateMock, + transactionMock, + reconcileMock, } = vi.hoisted(() => ({ getConnMock: vi.fn(), storeTokensMock: vi.fn(), @@ -46,6 +48,8 @@ const { recordFailureMock: vi.fn(), recomputeMock: vi.fn(), invalidateMock: vi.fn(), + transactionMock: vi.fn(), + reconcileMock: vi.fn(), })); vi.mock("../credentials", () => ({ @@ -62,9 +66,15 @@ vi.mock("@/lib/db", () => ({ findMany: findManyMock, updateMany: updateManyMock, }, + $transaction: transactionMock, }, })); +vi.mock("@/lib/measurements/reconcile-external-measurement", () => ({ + reconcileExternalMeasurement: reconcileMock, + MeasurementReconciliationError: class extends Error {}, +})); + vi.mock("@/lib/integrations/status", async (importOriginal) => ({ ...(await importOriginal()), recordSyncSuccess: recordSuccessMock, @@ -129,6 +139,20 @@ beforeEach(() => { fetchResilienceMock.mockReset().mockResolvedValue([]); fetchCyclePhasesMock.mockReset().mockResolvedValue([]); refreshMock.mockReset(); + transactionMock + .mockReset() + .mockImplementation(async (run: (tx: unknown) => unknown) => run({})); + reconcileMock.mockReset().mockImplementation( + async (_tx: unknown, input: { type: string; measuredAt: Date; externalId: string }) => ({ + status: "inserted", + row: { + id: `inserted:${input.externalId}`, + type: input.type, + measuredAt: input.measuredAt, + externalId: input.externalId, + }, + }), + ); createManyAndReturnMock.mockReset().mockImplementation( async ({ data, @@ -168,7 +192,7 @@ function err401() { } function createdRows() { - return createManyAndReturnMock.mock.calls.flatMap((call) => call[0].data); + return reconcileMock.mock.calls.map((call) => call[1]); } describe("syncUserOura", () => { @@ -327,7 +351,7 @@ describe("syncUserOura", () => { // Replace-then-write: the sweep runs before the insert statement. expect(updateManyMock.mock.invocationCallOrder[0]!).toBeLessThan( - createManyAndReturnMock.mock.invocationCallOrder[0]!, + reconcileMock.mock.invocationCallOrder[0]!, ); }); @@ -408,7 +432,7 @@ describe("syncUserOura", () => { ]); const imported = await syncUserOura("u1"); expect(imported).toBe(0); - expect(createManyAndReturnMock).not.toHaveBeenCalled(); + expect(reconcileMock).not.toHaveBeenCalled(); }); it("re-sync of the same day upserts RESILIENCE in place (overwrite, not duplicate)", async () => { @@ -421,26 +445,34 @@ describe("syncUserOura", () => { { id: "r", day: "2026-06-10", level: "adequate" }, ]) .mockResolvedValueOnce([{ id: "r", day: "2026-06-10", level: "solid" }]); - createManyAndReturnMock - .mockResolvedValueOnce([ - { + reconcileMock + .mockResolvedValueOnce({ + status: "inserted", + row: { id: "inserted-resilience", type: "RESILIENCE", measuredAt: new Date("2026-06-10T00:00:00.000Z"), externalId: "resilience:2026-06-10:resilience", }, - ]) - .mockResolvedValueOnce([]); + }) + .mockResolvedValueOnce({ + status: "updated", + row: { + id: "inserted-resilience", + type: "RESILIENCE", + measuredAt: new Date("2026-06-10T00:00:00.000Z"), + externalId: "resilience:2026-06-10:resilience", + }, + }); await syncUserOura("u1"); await syncUserOura("u1"); - const createCalls = createManyAndReturnMock.mock.calls; - expect(createCalls.map((call) => call[0].data[0].externalId)).toEqual([ + const writes = reconcileMock.mock.calls.map((call) => call[1]); + expect(writes.map((row) => row.externalId)).toEqual([ "resilience:2026-06-10:resilience", "resilience:2026-06-10:resilience", ]); - // The second sync carries the re-scored value through the update branch. - expect(createCalls[0]![0].data[0].value).toBe(2); - expect(updateMock.mock.calls[0]![0].data.value).toBe(3); + expect(writes[0].value).toBe(2); + expect(writes[1].value).toBe(3); }); it("does NOT refresh on a 403 (not an expiry case)", async () => { @@ -599,7 +631,17 @@ describe("upsertOuraMeasurements — exact insertion results", () => { measuredAt, externalId: readings[0].externalId, }; - createManyAndReturnMock.mockResolvedValue([inserted]); + reconcileMock + .mockResolvedValueOnce({ status: "inserted", row: inserted }) + .mockResolvedValueOnce({ + status: "updated", + row: { + id: "existing-2", + type: "HRV_RMSSD", + measuredAt, + externalId: readings[1].externalId, + }, + }); const onInserted = vi.fn(); expect(await upsertOuraMeasurements("u1", readings, { onInserted })).toBe( @@ -608,10 +650,7 @@ describe("upsertOuraMeasurements — exact insertion results", () => { expect(onInserted).toHaveBeenCalledWith([inserted]); expect(emitArrivalMock).toHaveBeenCalledWith("u1", [inserted], "oura"); - expect(updateMock).toHaveBeenCalledTimes(1); - expect( - updateMock.mock.calls[0]![0].where.userId_type_source_externalId, - ).toEqual({ + expect(reconcileMock.mock.calls[1]![1]).toMatchObject({ userId: "u1", type: "HRV_RMSSD", source: "OURA", @@ -627,7 +666,7 @@ describe("upsertOuraMeasurements — exact insertion results", () => { measuredAt, externalId: readings[0].externalId, }; - createManyAndReturnMock.mockResolvedValue([inserted]); + reconcileMock.mockResolvedValueOnce({ status: "inserted", row: inserted }); await upsertOuraMeasurements("u1", [readings[0]]); diff --git a/src/lib/oura/sync.ts b/src/lib/oura/sync.ts index 1fb2c5c95..c534fabc1 100644 --- a/src/lib/oura/sync.ts +++ b/src/lib/oura/sync.ts @@ -27,9 +27,9 @@ * Idempotency: `externalId = ::` for the day-keyed * collections; sleep rows carry a record-scoped `sleep::` * key (per-segment timeline + nightly scalars) so a nap and the main sleep on - * one day stay distinct instead of overwriting each other (B2). Upsert keyed on - * `(userId, type, source = OURA, externalId)`. Oura finalises a day's scores - * after the night, so the `update` branch overwrites in place (re-score). + * one day stay distinct instead of overwriting each other (B2). The shared + * reconciler protects both external and natural identity. Oura finalises a + * day's scores after the night, so an exact re-post overwrites in place. * * The measurement-write tail mirrors the shared WHOOP / Nightscout sync tail. */ @@ -56,6 +56,10 @@ import { sweepStaleSleepSegments, type SleepSegmentSweep, } from "@/lib/sleep/sweep-stale-segments"; +import { + MeasurementReconciliationError, + reconcileExternalMeasurement, +} from "@/lib/measurements/reconcile-external-measurement"; import { fetchCardiovascularAge, fetchDailyActivity, @@ -453,76 +457,46 @@ export async function upsertOuraMeasurements( let imported = 0; const touched: Array<{ type: MeasurementType; measuredAt: Date }> = []; - let insertedRows: Array< + const insertedRows: Array< InsertedMeasurementArrivalRow & { externalId: string | null } > = []; - try { - insertedRows = await prisma.measurement.createManyAndReturn({ - data: readings.map((r) => ({ - userId, - type: r.type as MeasurementType, - source: "OURA" as const, - value: r.value, - unit: r.unit, - measuredAt: r.measuredAt, - externalId: r.externalId, - sleepStage: r.sleepStage ?? null, - })), - skipDuplicates: true, - select: { - id: true, - type: true, - measuredAt: true, - externalId: true, - }, - }); - imported += insertedRows.length; - for (const row of insertedRows) { - touched.push({ type: row.type, measuredAt: row.measuredAt }); - } - } catch (err) { - getEvent()?.addWarning(`oura: failed to create measurements: ${err}`); - } - - const insertedIdentityCounts = new Map(); - for (const row of insertedRows) { - const key = `${row.type}:${row.externalId ?? ""}`; - insertedIdentityCounts.set(key, (insertedIdentityCounts.get(key) ?? 0) + 1); - } - - for (const r of readings) { - const type = r.type as MeasurementType; - const key = `${type}:${r.externalId}`; - const insertedCount = insertedIdentityCounts.get(key) ?? 0; - if (insertedCount > 0) { - insertedIdentityCounts.set(key, insertedCount - 1); - continue; - } - - try { - await prisma.measurement.update({ - where: { - userId_type_source_externalId: { - userId, - type, - source: "OURA", - externalId: r.externalId, - }, - }, - data: { + const verdicts = await prisma.$transaction(async (tx) => { + const outcomes = []; + for (const r of readings) { + const verdict = await reconcileExternalMeasurement( + tx, + { + userId, + type: r.type as MeasurementType, + source: "OURA", value: r.value, unit: r.unit, measuredAt: r.measuredAt, + externalId: r.externalId, sleepStage: r.sleepStage ?? null, - deletedAt: null, - syncVersion: { increment: 1 }, }, - }); - touched.push({ type, measuredAt: r.measuredAt }); - imported++; - } catch (err) { - getEvent()?.addWarning(`oura: failed to update measurement: ${err}`); + { exactExternalMatch: "update" }, + ); + if (verdict.status === "failed") { + throw new MeasurementReconciliationError(verdict); + } + outcomes.push(verdict); + } + return outcomes; + }, { timeout: 60_000 }); + + for (let index = 0; index < readings.length; index++) { + const reading = readings[index]!; + const verdict = verdicts[index]!; + for (const dirty of verdict.dirtyIdentities ?? []) { + touched.push(dirty); + } + const type = reading.type as MeasurementType; + imported++; + touched.push({ type, measuredAt: reading.measuredAt }); + if (verdict.status === "inserted") { + insertedRows.push(verdict.row); } } diff --git a/src/lib/polar/__tests__/sync.test.ts b/src/lib/polar/__tests__/sync.test.ts index c00c2bb3c..da00e02cd 100644 --- a/src/lib/polar/__tests__/sync.test.ts +++ b/src/lib/polar/__tests__/sync.test.ts @@ -15,6 +15,8 @@ const { recordSuccessMock, recordFailureMock, recomputeMock, + transactionMock, + reconcileMock, invalidateMock, } = vi.hoisted(() => ({ getConnMock: vi.fn(), @@ -31,6 +33,8 @@ const { recordSuccessMock: vi.fn(), recordFailureMock: vi.fn(), recomputeMock: vi.fn(), + transactionMock: vi.fn(), + reconcileMock: vi.fn(), invalidateMock: vi.fn(), })); @@ -46,9 +50,15 @@ vi.mock("@/lib/db", () => ({ findMany: findManyMock, updateMany: updateManyMock, }, + $transaction: transactionMock, }, })); +vi.mock("@/lib/measurements/reconcile-external-measurement", () => ({ + reconcileExternalMeasurement: reconcileMock, + MeasurementReconciliationError: class extends Error {}, +})); + vi.mock("@/lib/integrations/status", async (importOriginal) => ({ ...(await importOriginal()), recordSyncSuccess: recordSuccessMock, @@ -93,6 +103,20 @@ beforeEach(() => { fetchActivitiesMock.mockReset(); fetchCardioLoadsMock.mockReset(); fetchSpo2Mock.mockReset(); + transactionMock + .mockReset() + .mockImplementation(async (run: (tx: unknown) => unknown) => run({})); + reconcileMock.mockReset().mockImplementation( + async (_tx: unknown, input: { type: string; measuredAt: Date; externalId: string }) => ({ + status: "inserted", + row: { + id: `inserted:${input.externalId}`, + type: input.type, + measuredAt: input.measuredAt, + externalId: input.externalId, + }, + }), + ); createManyAndReturnMock.mockReset().mockImplementation( async ({ data, @@ -128,7 +152,7 @@ beforeEach(() => { afterEach(() => vi.clearAllMocks()); function createdRows() { - return createManyAndReturnMock.mock.calls.flatMap((call) => call[0].data); + return reconcileMock.mock.calls.map((call) => call[1]); } describe("syncUserPolar", () => { @@ -245,7 +269,7 @@ describe("syncUserPolar", () => { // Replace-then-write: the sweep runs before the insert statement. expect(updateManyMock.mock.invocationCallOrder[0]!).toBeLessThan( - createManyAndReturnMock.mock.invocationCallOrder[0]!, + reconcileMock.mock.invocationCallOrder[0]!, ); }); @@ -263,21 +287,29 @@ describe("syncUserPolar", () => { fetchRechargesMock.mockResolvedValue([ { date: "2026-06-10", nightly_recharge_status: 3 }, ]); - createManyAndReturnMock - .mockResolvedValueOnce([ - { + reconcileMock + .mockResolvedValueOnce({ + status: "inserted", + row: { id: "inserted-recharge", type: "RECOVERY_SCORE", measuredAt: new Date("2026-06-10T00:00:00.000Z"), externalId: "recharge:2026-06-10:recovery", }, - ]) - .mockResolvedValueOnce([]); + }) + .mockResolvedValueOnce({ + status: "updated", + row: { + id: "inserted-recharge", + type: "RECOVERY_SCORE", + measuredAt: new Date("2026-06-10T00:00:00.000Z"), + externalId: "recharge:2026-06-10:recovery", + }, + }); await syncUserPolar("u1"); await syncUserPolar("u1"); - expect(createManyAndReturnMock.mock.calls[0]![0].data[0].externalId).toBe( - updateMock.mock.calls[0]![0].where.userId_type_source_externalId - .externalId, + expect(reconcileMock.mock.calls[0]![1].externalId).toBe( + reconcileMock.mock.calls[1]![1].externalId, ); }); @@ -330,7 +362,17 @@ describe("upsertPolarMeasurements — exact insertion results", () => { measuredAt, externalId: readings[0].externalId, }; - createManyAndReturnMock.mockResolvedValue([inserted]); + reconcileMock + .mockResolvedValueOnce({ status: "inserted", row: inserted }) + .mockResolvedValueOnce({ + status: "updated", + row: { + id: "existing-2", + type: "CARDIO_LOAD", + measuredAt, + externalId: readings[1].externalId, + }, + }); const onInserted = vi.fn(); expect(await upsertPolarMeasurements("u1", readings, { onInserted })).toBe( @@ -339,10 +381,7 @@ describe("upsertPolarMeasurements — exact insertion results", () => { expect(onInserted).toHaveBeenCalledWith([inserted]); expect(emitArrivalMock).toHaveBeenCalledWith("u1", [inserted], "polar"); - expect(updateMock).toHaveBeenCalledTimes(1); - expect( - updateMock.mock.calls[0]![0].where.userId_type_source_externalId, - ).toEqual({ + expect(reconcileMock.mock.calls[1]![1]).toMatchObject({ userId: "u1", type: "CARDIO_LOAD", source: "POLAR", @@ -358,7 +397,7 @@ describe("upsertPolarMeasurements — exact insertion results", () => { measuredAt, externalId: readings[0].externalId, }; - createManyAndReturnMock.mockResolvedValue([inserted]); + reconcileMock.mockResolvedValueOnce({ status: "inserted", row: inserted }); await upsertPolarMeasurements("u1", [readings[0]]); diff --git a/src/lib/polar/sync.ts b/src/lib/polar/sync.ts index ddd8f4a50..f321239a5 100644 --- a/src/lib/polar/sync.ts +++ b/src/lib/polar/sync.ts @@ -12,10 +12,10 @@ * as a 401 → `reauth_required` on the shared integration ledger (`polar` key), * exactly like Nightscout's token-rejected path. * - * Idempotency: each row's `externalId` is `:` and the write is - * an upsert keyed on `(userId, type, source = POLAR, externalId)`. Polar - * re-scores a night for a short window after the fact, so the `update` branch - * overwrites in place (WHOOP-style re-score), not first-write-wins. + * Idempotency: each row's `externalId` is `:`. The shared + * reconciler protects both external and natural identity. Polar re-scores a + * night for a short window after the fact, so an exact re-post overwrites in + * place (WHOOP-style re-score), not first-write-wins. * * The measurement-write tail (per-row upsert → rollup fold → status-insight * invalidate) mirrors the shared WHOOP / Nightscout sync tail; it is NOT a new @@ -40,6 +40,10 @@ import { emitInsertedMeasurementArrivals, type InsertedMeasurementArrivalRow, } from "@/lib/arrivals/measurement-emit"; +import { + MeasurementReconciliationError, + reconcileExternalMeasurement, +} from "@/lib/measurements/reconcile-external-measurement"; import { fetchActivities, fetchCardioLoads, @@ -187,8 +191,9 @@ export async function syncUserPolar(userId: string): Promise { /** * Upsert a batch of mapped Polar readings, then fold the rollup tier + * invalidate status-insight caches once at the end (mirrors the WHOOP / - * Nightscout sync tail). Idempotent: the `(userId, type, source, externalId)` - * unique key makes a re-post overwrite in place. Best-effort on the rollup fold. + * Nightscout sync tail). The shared reconciler protects both external and + * natural identity; an exact re-post overwrites in place. Best-effort on the + * rollup fold. */ export async function upsertPolarMeasurements( userId: string, @@ -201,76 +206,46 @@ export async function upsertPolarMeasurements( let imported = 0; const touched: Array<{ type: MeasurementType; measuredAt: Date }> = []; - let insertedRows: Array< + const insertedRows: Array< InsertedMeasurementArrivalRow & { externalId: string | null } > = []; - try { - insertedRows = await prisma.measurement.createManyAndReturn({ - data: readings.map((r) => ({ - userId, - type: r.type as MeasurementType, - source: "POLAR" as const, - value: r.value, - unit: r.unit, - measuredAt: r.measuredAt, - externalId: r.externalId, - sleepStage: r.sleepStage ?? null, - })), - skipDuplicates: true, - select: { - id: true, - type: true, - measuredAt: true, - externalId: true, - }, - }); - imported += insertedRows.length; - for (const row of insertedRows) { - touched.push({ type: row.type, measuredAt: row.measuredAt }); - } - } catch (err) { - getEvent()?.addWarning(`polar: failed to create measurements: ${err}`); - } - - const insertedIdentityCounts = new Map(); - for (const row of insertedRows) { - const key = `${row.type}:${row.externalId ?? ""}`; - insertedIdentityCounts.set(key, (insertedIdentityCounts.get(key) ?? 0) + 1); - } - - for (const r of readings) { - const type = r.type as MeasurementType; - const key = `${type}:${r.externalId}`; - const insertedCount = insertedIdentityCounts.get(key) ?? 0; - if (insertedCount > 0) { - insertedIdentityCounts.set(key, insertedCount - 1); - continue; - } - - try { - await prisma.measurement.update({ - where: { - userId_type_source_externalId: { - userId, - type, - source: "POLAR", - externalId: r.externalId, - }, - }, - data: { + const verdicts = await prisma.$transaction(async (tx) => { + const outcomes = []; + for (const r of readings) { + const verdict = await reconcileExternalMeasurement( + tx, + { + userId, + type: r.type as MeasurementType, + source: "POLAR", value: r.value, unit: r.unit, measuredAt: r.measuredAt, + externalId: r.externalId, sleepStage: r.sleepStage ?? null, - deletedAt: null, - syncVersion: { increment: 1 }, }, - }); - touched.push({ type, measuredAt: r.measuredAt }); - imported++; - } catch (err) { - getEvent()?.addWarning(`polar: failed to update measurement: ${err}`); + { exactExternalMatch: "update" }, + ); + if (verdict.status === "failed") { + throw new MeasurementReconciliationError(verdict); + } + outcomes.push(verdict); + } + return outcomes; + }, { timeout: 60_000 }); + + for (let index = 0; index < readings.length; index++) { + const reading = readings[index]!; + const verdict = verdicts[index]!; + for (const dirty of verdict.dirtyIdentities ?? []) { + touched.push(dirty); + } + const type = reading.type as MeasurementType; + imported++; + touched.push({ type, measuredAt: reading.measuredAt }); + if (verdict.status === "inserted") { + insertedRows.push(verdict.row); } } diff --git a/src/lib/whoop/__tests__/sync.test.ts b/src/lib/whoop/__tests__/sync.test.ts index c9c1d0acf..91dff2c21 100644 --- a/src/lib/whoop/__tests__/sync.test.ts +++ b/src/lib/whoop/__tests__/sync.test.ts @@ -16,6 +16,7 @@ const { recordSyncSuccess, isReauthRequired, emitArrivalMock, + reconcileMock, } = vi.hoisted(() => ({ prismaMock: { whoopConnection: { @@ -30,6 +31,7 @@ const { findMany: vi.fn(), }, $executeRaw: vi.fn(), + $transaction: vi.fn(), }, refreshAccessTokenMock: vi.fn(), fetchRecoveriesMock: vi.fn(), @@ -38,6 +40,7 @@ const { isReauthRequired: vi.fn<(...a: unknown[]) => Promise>( async () => false, ), + reconcileMock: vi.fn(), emitArrivalMock: vi.fn(), })); @@ -57,6 +60,11 @@ vi.mock("../client", async (orig) => { }; }); +vi.mock("@/lib/measurements/reconcile-external-measurement", () => ({ + reconcileExternalMeasurement: reconcileMock, + MeasurementReconciliationError: class extends Error {}, +})); + vi.mock("../credentials", () => ({ getUserWhoopCredentials: vi.fn(async () => ({ clientId: "cid", @@ -105,6 +113,20 @@ import { syncUserRecovery } from "../sync-recovery"; beforeEach(() => { vi.clearAllMocks(); + prismaMock.$transaction.mockImplementation(async (run: (tx: unknown) => unknown) => + run({}), + ); + reconcileMock.mockImplementation( + async (_tx: unknown, input: { type: string; measuredAt: Date; externalId: string }) => ({ + status: "inserted", + row: { + id: `inserted:${input.externalId}`, + type: input.type, + measuredAt: input.measuredAt, + externalId: input.externalId, + }, + }), + ); isReauthRequired.mockResolvedValue(false); emitArrivalMock.mockResolvedValue(undefined); }); @@ -371,8 +393,17 @@ describe("upsertWhoopMeasurements — exact insertion results", () => { measuredAt, externalId: readings[0].externalId, }; - prismaMock.measurement.createManyAndReturn.mockResolvedValue([inserted]); - prismaMock.measurement.update.mockResolvedValue({}); + reconcileMock + .mockResolvedValueOnce({ status: "inserted", row: inserted }) + .mockResolvedValueOnce({ + status: "updated", + row: { + id: "existing-2", + type: "HRV_RMSSD", + measuredAt, + externalId: readings[1].externalId, + }, + }); const onInserted = vi.fn(); expect( @@ -381,11 +412,7 @@ describe("upsertWhoopMeasurements — exact insertion results", () => { expect(onInserted).toHaveBeenCalledWith([inserted]); expect(emitArrivalMock).toHaveBeenCalledWith("user1", [inserted], "whoop"); - expect(prismaMock.measurement.update).toHaveBeenCalledTimes(1); - expect( - prismaMock.measurement.update.mock.calls[0]![0].where - .userId_type_source_externalId, - ).toEqual({ + expect(reconcileMock.mock.calls[1]![1]).toMatchObject({ userId: "user1", type: "HRV_RMSSD", source: "WHOOP", @@ -403,7 +430,7 @@ describe("upsertWhoopMeasurements — exact insertion results", () => { measuredAt, externalId: readings[0].externalId, }; - prismaMock.measurement.createManyAndReturn.mockResolvedValue([inserted]); + reconcileMock.mockResolvedValueOnce({ status: "inserted", row: inserted }); await upsertWhoopMeasurements("user1", [readings[0]]); @@ -412,20 +439,23 @@ describe("upsertWhoopMeasurements — exact insertion results", () => { }); it("updates a re-post in place without reporting another insert", async () => { - prismaMock.measurement.createManyAndReturn.mockResolvedValue([]); - prismaMock.measurement.update.mockResolvedValue({}); + reconcileMock.mockResolvedValueOnce({ + status: "updated", + row: { + id: "existing-1", + type: "RECOVERY_SCORE", + measuredAt, + externalId: readings[0].externalId, + }, + }); expect(await upsertWhoopMeasurements("user1", [readings[0]])).toBe(1); expect(emitArrivalMock).toHaveBeenCalledWith("user1", [], "whoop"); - expect(prismaMock.measurement.update).toHaveBeenCalledWith( - expect.objectContaining({ - data: expect.objectContaining({ - value: 71, - syncVersion: { increment: 1 }, - }), - }), - ); + expect(reconcileMock.mock.calls[0]![1]).toMatchObject({ + value: 71, + externalId: readings[0].externalId, + }); }); }); diff --git a/src/lib/whoop/sync.ts b/src/lib/whoop/sync.ts index d7043ae0c..c3802ba14 100644 --- a/src/lib/whoop/sync.ts +++ b/src/lib/whoop/sync.ts @@ -45,6 +45,10 @@ import { type InsertedMeasurementArrivalRow, } from "@/lib/arrivals/measurement-emit"; import { persistRotatedToken } from "@/lib/integrations/oauth-refresh"; +import { + MeasurementReconciliationError, + reconcileExternalMeasurement, +} from "@/lib/measurements/reconcile-external-measurement"; import { refreshAccessToken } from "./client"; import { getUserWhoopCredentials } from "./credentials"; import { @@ -324,8 +328,8 @@ export interface WhoopMeasurementUpsert { /** * Upsert a batch of mapped WHOOP readings for one user and fold the rollup * tier + invalidate status-insight caches once at the end (mirrors the - * Withings sync tail). Idempotent: the `(userId, type, source, externalId)` - * unique key makes a re-post (WHOOP re-score) overwrite in place. Returns the + * Withings sync tail). The shared reconciler protects both external and + * natural identity; an exact WHOOP re-score overwrites in place. Returns the * count of rows written. * * Best-effort on the rollup fold + insight invalidate — a populator hiccup @@ -342,76 +346,46 @@ export async function upsertWhoopMeasurements( let imported = 0; const touched: Array<{ type: MeasurementType; measuredAt: Date }> = []; - let insertedRows: Array< + const insertedRows: Array< InsertedMeasurementArrivalRow & { externalId: string | null } > = []; - try { - insertedRows = await prisma.measurement.createManyAndReturn({ - data: readings.map((r) => ({ - userId, - type: r.type as MeasurementType, - source: "WHOOP" as const, - value: r.value, - unit: r.unit, - measuredAt: r.measuredAt, - externalId: r.externalId, - sleepStage: r.sleepStage ?? null, - })), - skipDuplicates: true, - select: { - id: true, - type: true, - measuredAt: true, - externalId: true, - }, - }); - imported += insertedRows.length; - for (const row of insertedRows) { - touched.push({ type: row.type, measuredAt: row.measuredAt }); - } - } catch (err) { - getEvent()?.addWarning(`WHOOP: failed to create measurements: ${err}`); - } - - const insertedIdentityCounts = new Map(); - for (const row of insertedRows) { - const key = `${row.type}:${row.externalId ?? ""}`; - insertedIdentityCounts.set(key, (insertedIdentityCounts.get(key) ?? 0) + 1); - } - - for (const r of readings) { - const type = r.type as MeasurementType; - const key = `${type}:${r.externalId}`; - const insertedCount = insertedIdentityCounts.get(key) ?? 0; - if (insertedCount > 0) { - insertedIdentityCounts.set(key, insertedCount - 1); - continue; - } - - try { - await prisma.measurement.update({ - where: { - userId_type_source_externalId: { - userId, - type, - source: "WHOOP", - externalId: r.externalId, - }, - }, - data: { + const verdicts = await prisma.$transaction(async (tx) => { + const outcomes = []; + for (const r of readings) { + const verdict = await reconcileExternalMeasurement( + tx, + { + userId, + type: r.type as MeasurementType, + source: "WHOOP", value: r.value, unit: r.unit, measuredAt: r.measuredAt, + externalId: r.externalId, sleepStage: r.sleepStage ?? null, - deletedAt: null, - syncVersion: { increment: 1 }, }, - }); - touched.push({ type, measuredAt: r.measuredAt }); - imported++; - } catch (err) { - getEvent()?.addWarning(`WHOOP: failed to update measurement: ${err}`); + { exactExternalMatch: "update" }, + ); + if (verdict.status === "failed") { + throw new MeasurementReconciliationError(verdict); + } + outcomes.push(verdict); + } + return outcomes; + }, { timeout: 60_000 }); + + for (let index = 0; index < readings.length; index++) { + const reading = readings[index]!; + const verdict = verdicts[index]!; + for (const dirty of verdict.dirtyIdentities ?? []) { + touched.push(dirty); + } + const type = reading.type as MeasurementType; + imported++; + touched.push({ type, measuredAt: reading.measuredAt }); + if (verdict.status === "inserted") { + insertedRows.push(verdict.row); } } diff --git a/src/lib/withings/__tests__/sync-sleep.test.ts b/src/lib/withings/__tests__/sync-sleep.test.ts index f2c6b8912..dd5271afd 100644 --- a/src/lib/withings/__tests__/sync-sleep.test.ts +++ b/src/lib/withings/__tests__/sync-sleep.test.ts @@ -8,6 +8,11 @@ */ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +const { reconcileMock, transactionMock } = vi.hoisted(() => ({ + reconcileMock: vi.fn(), + transactionMock: vi.fn(), +})); + vi.mock("@/lib/db", () => ({ prisma: { measurement: { @@ -18,9 +23,15 @@ vi.mock("@/lib/db", () => ({ updateMany: vi.fn(), }, withingsConnection: { findUnique: vi.fn() }, + $transaction: transactionMock, }, })); +vi.mock("@/lib/measurements/reconcile-external-measurement", () => ({ + reconcileExternalMeasurement: reconcileMock, + MeasurementReconciliationError: class extends Error {}, +})); + vi.mock("@/lib/integrations/status", () => ({ isReauthRequired: vi.fn().mockResolvedValue(false), parkIntegrationAtReauth: vi.fn(), @@ -96,6 +107,20 @@ function installFetchMock(segments: FakeSegment[]) { beforeEach(() => { vi.resetAllMocks(); + transactionMock.mockImplementation( + async (run: (tx: unknown) => unknown) => run({}), + ); + reconcileMock.mockImplementation( + async (_tx: unknown, input: { type: string; measuredAt: Date; externalId: string }) => ({ + status: "inserted", + row: { + id: `inserted:${input.externalId}`, + type: input.type, + measuredAt: input.measuredAt, + externalId: input.externalId, + }, + }), + ); // v1.4.26 — every syncUserSleep call now reads the connection's // scope to short-circuit legacy `user.metrics`-only connections. // Default-mock to "scope is fine" so the existing segment-mapping @@ -249,16 +274,14 @@ describe("syncUserSleep — segment writes + idempotency", () => { const imported = await syncUserSleep("user-1"); expect(imported).toBe(4); - const stages = vi - .mocked(prisma.measurement.create) - .mock.calls.map((c) => (c[0].data as { sleepStage: string }).sleepStage); - expect(stages).toEqual(["CORE", "DEEP", "REM", "AWAKE"]); - - // First segment: 60 minutes (3600s / 60). - const firstDuration = vi.mocked(prisma.measurement.create).mock - .calls[0][0] as { data: { value: number; unit: string } }; - expect(firstDuration.data.value).toBe(60); - expect(firstDuration.data.unit).toBe("minutes"); + const writes = reconcileMock.mock.calls.map((call) => call[1]); + expect(writes.map((row) => row.sleepStage)).toEqual([ + "CORE", + "DEEP", + "REM", + "AWAKE", + ]); + expect(writes[0]).toMatchObject({ value: 60, unit: "minutes" }); }); it("stamps measuredAt at the segment END (enddate, unix seconds → Date)", async () => { @@ -269,13 +292,11 @@ describe("syncUserSleep — segment writes + idempotency", () => { vi.mocked(prisma.measurement.create).mockResolvedValue({} as never); await syncUserSleep("user-1"); - const arg = vi.mocked(prisma.measurement.create).mock.calls[0][0] as { - data: { measuredAt: Date }; - }; + const row = reconcileMock.mock.calls[0]![1] as { measuredAt: Date }; // measuredAt is the segment END — every reader treats it as the END and // resolves the span as start = end − duration. Stamping the START shifted // the night one segment-length earlier. - expect(arg.data.measuredAt.getTime()).toBe(1715003600 * 1000); + expect(row.measuredAt.getTime()).toBe(1715003600 * 1000); }); it("skips state 4 (synthetic marker) without throwing", async () => { @@ -288,9 +309,7 @@ describe("syncUserSleep — segment writes + idempotency", () => { const imported = await syncUserSleep("user-1"); expect(imported).toBe(1); - const stages = vi - .mocked(prisma.measurement.create) - .mock.calls.map((c) => (c[0].data as { sleepStage: string }).sleepStage); + const stages = reconcileMock.mock.calls.map((call) => call[1].sleepStage); expect(stages).toEqual(["DEEP"]); }); @@ -298,27 +317,23 @@ describe("syncUserSleep — segment writes + idempotency", () => { installFetchMock([ { startdate: 1715000000, enddate: 1715003600, state: 2, id: 1 }, ]); - // findFirst returns an existing row → writer takes the update path. - vi.mocked(prisma.measurement.findFirst).mockResolvedValue({ - id: "row-1", - } as never); - vi.mocked(prisma.measurement.update).mockResolvedValue({} as never); + reconcileMock.mockResolvedValueOnce({ + status: "updated", + row: { + id: "row-1", + type: "SLEEP_DURATION", + measuredAt: new Date(1715003600 * 1000), + externalId: "withings:sleep:user-1:1:1715000000", + }, + }); const imported = await syncUserSleep("user-1"); expect(imported).toBe(1); - expect(prisma.measurement.create).not.toHaveBeenCalled(); - // Re-sync keys on the stable externalId and refreshes value, the - // END-stamped measuredAt (enddate may shift) and the stage. - expect(prisma.measurement.update).toHaveBeenCalledWith({ - where: { id: "row-1" }, - data: { - value: 60, - measuredAt: new Date(1715003600 * 1000), - sleepStage: "DEEP", - // No-op on a live row, resurrection on a tombstoned one — the - // source-owned resurrect rule. - deletedAt: null, - }, + expect(reconcileMock.mock.calls[0]![1]).toMatchObject({ + value: 60, + measuredAt: new Date(1715003600 * 1000), + sleepStage: "DEEP", + externalId: "withings:sleep:user-1:1:1715000000", }); }); @@ -333,32 +348,13 @@ describe("syncUserSleep — segment writes + idempotency", () => { installFetchMock([ { startdate: 1715000500, enddate: 1715003600, state: 2, id: 42 }, ]); - vi.mocked(prisma.measurement.findFirst).mockImplementation((async (args: { - where: Record; - }) => { - const where = args.where; - // externalId probe (START shifted ⇒ new id) → miss. - if ("externalId" in where) return null; - // natural-key probe (END + stage unchanged) → the surviving prior row. - if ("measuredAt" in where) return { id: "old-row" }; - return null; - }) as never); - vi.mocked(prisma.measurement.update).mockResolvedValue({} as never); - vi.mocked(prisma.measurement.create).mockResolvedValue({} as never); - const imported = await syncUserSleep("user-1"); expect(imported).toBe(1); - - // No collision: the survivor is re-keyed in place, never re-created. - expect(prisma.measurement.create).not.toHaveBeenCalled(); - expect(prisma.measurement.update).toHaveBeenCalledWith({ - where: { id: "old-row" }, - data: { - value: expect.any(Number), - // Re-keyed onto the shifted START-based externalId and resurrected. - externalId: "withings:sleep:user-1:42:1715000500", - deletedAt: null, - }, + expect(reconcileMock.mock.calls[0]![1]).toMatchObject({ + value: expect.any(Number), + measuredAt: new Date(1715003600 * 1000), + sleepStage: "DEEP", + externalId: "withings:sleep:user-1:42:1715000500", }); }); @@ -370,14 +366,12 @@ describe("syncUserSleep — segment writes + idempotency", () => { vi.mocked(prisma.measurement.create).mockResolvedValue({} as never); await syncUserSleep("user-1"); - const arg = vi.mocked(prisma.measurement.create).mock.calls[0][0] as { - data: { externalId: string }; - }; + const row = reconcileMock.mock.calls[0]![1] as { externalId: string }; // Session id + the segment's own startdate — both stable within a session. // The retired running index counted across the whole rolling 30-day fetch // window, so the window slide renumbered every segment on each sync and a // re-aggregated night inserted a duplicate set. - expect(arg.data.externalId).toBe("withings:sleep:user-1:42:1715000000"); + expect(row.externalId).toBe("withings:sleep:user-1:42:1715000000"); }); it("mints IDENTICAL externalIds when the fetch window slides (re-fetch of the same night)", async () => { @@ -395,19 +389,18 @@ describe("syncUserSleep — segment writes + idempotency", () => { installFetchMock(night); await syncUserSleep("user-1"); - const firstIds = vi - .mocked(prisma.measurement.create) - .mock.calls.map((c) => (c[0].data as { externalId: string }).externalId); + const firstIds = reconcileMock.mock.calls.map( + (call) => call[1].externalId as string, + ); - vi.mocked(prisma.measurement.create).mockClear(); + reconcileMock.mockClear(); installFetchMock([ { startdate: 1714990000, enddate: 1714993600, state: 1, id: 7 }, ...night, ]); await syncUserSleep("user-1"); - const secondIds = vi - .mocked(prisma.measurement.create) - .mock.calls.map((c) => (c[0].data as { externalId: string }).externalId) + const secondIds = reconcileMock.mock.calls + .map((call) => call[1].externalId as string) .filter((id) => id.includes(":42:")); expect(secondIds).toEqual(firstIds); @@ -496,27 +489,22 @@ describe("syncUserSleep — segment writes + idempotency", () => { }, ], ); - vi.mocked(prisma.measurement.upsert).mockResolvedValue({} as never); + reconcileMock.mockResolvedValue({ status: "updated", row: {} }); const imported = await syncUserSleep("user-1"); // Five vitals upserted (no stage segments in this fixture). expect(imported).toBe(5); - const calls = vi.mocked(prisma.measurement.upsert).mock.calls; const byType = new Map( - calls.map((c) => { - const w = c[0] as { - where: { - userId_type_source_externalId: { type: string; externalId: string }; - }; - create: { value: number }; + reconcileMock.mock.calls.map((call) => { + const row = call[1] as { + type: string; + externalId: string; + value: number; }; return [ - w.where.userId_type_source_externalId.type, - { - externalId: w.where.userId_type_source_externalId.externalId, - value: w.create.value, - }, + row.type, + { externalId: row.externalId, value: row.value }, ]; }), ); @@ -565,7 +553,7 @@ describe("syncUserSleep — segment writes + idempotency", () => { const imported = await syncUserSleep("user-1"); // The one stage segment is still written; the summary failure is swallowed. expect(imported).toBe(1); - expect(prisma.measurement.create).toHaveBeenCalledTimes(1); + expect(reconcileMock).toHaveBeenCalledTimes(1); expect(recordSyncSuccess).toHaveBeenCalledWith("user-1", "withings"); }); @@ -582,9 +570,7 @@ describe("syncUserSleep — segment writes + idempotency", () => { const imported = await syncUserSleep("user-1"); expect(imported).toBe(2); - const stages = vi - .mocked(prisma.measurement.create) - .mock.calls.map((c) => (c[0].data as { sleepStage: string }).sleepStage); + const stages = reconcileMock.mock.calls.map((call) => call[1].sleepStage); expect(stages).not.toContain("REM"); }); }); @@ -601,7 +587,7 @@ describe("syncUserSleep — scope-skip guard (v1.4.26)", () => { expect(imported).toBe(0); expect(fetchMock).not.toHaveBeenCalled(); - expect(prisma.measurement.create).not.toHaveBeenCalled(); + expect(reconcileMock).not.toHaveBeenCalled(); expect(recordSyncSuccess).not.toHaveBeenCalled(); }); diff --git a/src/lib/withings/sync-sleep.ts b/src/lib/withings/sync-sleep.ts index 00e14c5ee..6ba1cb710 100644 --- a/src/lib/withings/sync-sleep.ts +++ b/src/lib/withings/sync-sleep.ts @@ -54,6 +54,10 @@ import { sweepStaleSleepSegments, type SleepSegmentSweep, } from "@/lib/sleep/sweep-stale-segments"; +import { + MeasurementReconciliationError, + reconcileExternalMeasurement, +} from "@/lib/measurements/reconcile-external-measurement"; import { hasActivityScope } from "./client"; import { @@ -408,131 +412,48 @@ export async function syncUserSleep( // and are excluded (their `no-id` prefix would span every id-less night in // history, not just this fetch). const freshBySession = new Map(); - for (const segment of segments) { - const stage = mapWithingsSleepState(segment.state); - if (!stage) { - // State 4 (manual / synthetic marker) is ignored — see file - // header. - continue; - } - // `measuredAt` is the segment END (`enddate`). Every reader treats - // `measuredAt` as the END instant — `segmentOf` resolves the span as - // `start = measuredAt − value·60_000` (sleep-night.ts). Stamping the START - // here shifted the whole night one segment-length earlier; END is correct. - const measuredAt = new Date(segment.enddate * 1000); - const durationSec = Math.max(0, segment.enddate - segment.startdate); - // Canonical SLEEP_DURATION unit is minutes (Apple Health alignment, - // v1.4.23 schema note). Round so re-sync doesn't flip values on - // floating-point noise. - const minutes = Math.round(durationSec / 60); + await prisma.$transaction(async (tx) => { + for (const segment of segments) { + const stage = mapWithingsSleepState(segment.state); + if (!stage) continue; - // Keyed by the segment's own START instant — a coordinate that is stable - // within a session — NOT a positional index. The old running index counted - // across the whole 30-day fetch window, so the sliding window renumbered - // every segment on each sync: the probe below never matched, the update - // branch was dead code, and a re-aggregated night inserted a second set of - // rows next to the first (the same double-count bug class Google Health - // fixed with its stable session anchor). Legacy indexed rows are swept by - // the session-scoped cleanup below, so the format change self-heals. - const externalId = `withings:sleep:${userId}:${segment.id ?? "no-id"}:${segment.startdate}`; - if (typeof segment.id === "number") { - const fresh = freshBySession.get(segment.id) ?? []; - fresh.push(externalId); - freshBySession.set(segment.id, fresh); - } + const measuredAt = new Date(segment.enddate * 1000); + const durationSec = Math.max(0, segment.enddate - segment.startdate); + const minutes = Math.round(durationSec / 60); + const externalId = `withings:sleep:${userId}:${segment.id ?? "no-id"}:${segment.startdate}`; + if (typeof segment.id === "number") { + const fresh = freshBySession.get(segment.id) ?? []; + fresh.push(externalId); + freshBySession.set(segment.id, fresh); + } - try { - // Key the re-sync lookup on the STABLE `externalId` (session id + - // segment start), not `measuredAt`. Withings re-aggregates a night - // between syncs, so a segment's `enddate` — and therefore its - // END-stamped `measuredAt` — shifts; keying on `measuredAt` would miss - // the prior row and then collide with the unique `externalId`, leaving - // the night stuck at the stale value. Update both `value` and - // `measuredAt`. - const existing = await prisma.measurement.findFirst({ - where: { + const verdict = await reconcileExternalMeasurement( + tx, + { userId, type: SLEEP_TYPE, + value: minutes, + unit: getUnitForType(SLEEP_TYPE), + measuredAt, source: "WITHINGS", + sleepStage: stage, externalId, }, - select: { id: true }, - }); - if (existing) { - await prisma.measurement.update({ - where: { id: existing.id }, - // `deletedAt: null` is a no-op on a live row, a deliberate - // RESURRECTION on a tombstoned one — Withings is the source of - // truth for its own rows, so a re-fetched segment brings the row - // back (mirrors Google / Fitbit). - data: { - value: minutes, - measuredAt, - sleepStage: stage, - deletedAt: null, - }, - }); - } else { - // Natural-key rescue — mirrors the measure path's findFirst in - // `sync.ts`. When Withings re-scores a night it can shift a segment's - // START (its externalId) while the END (`measuredAt`) and the stage - // stay put: the externalId probe above then MISSES, but the 0055 - // natural key `(userId, type, measuredAt, source, sleepStage)` - // (NULLS NOT DISTINCT) is already occupied by the prior row. A blind - // `create` P2002s on that key; the collision is swallowed by the catch - // below and the sweep then tombstones the old row — so the re-keyed - // segment is lost and, because a tombstoned row still holds the natural - // key, every later sync re-collides: the night wedges permanently (the - // "0055 erased re-keyed sleep" class). Probe the natural key - // (tombstone-inclusive — no `deletedAt` filter) and re-key that row in - // place, so a re-scored segment updates rather than colliding. - const naturalKeyRow = await prisma.measurement.findFirst({ - where: { - userId, - type: SLEEP_TYPE, - source: "WITHINGS", - measuredAt, - sleepStage: stage, - }, - select: { id: true }, - }); - if (naturalKeyRow) { - await prisma.measurement.update({ - where: { id: naturalKeyRow.id }, - // Re-key the surviving row onto the new START-based externalId and - // resurrect it if the sweep had tombstoned it. `measuredAt` and - // `sleepStage` already match by construction (they ARE the natural - // key we matched on), so only value + externalId + deletedAt move. - data: { - value: minutes, - externalId, - deletedAt: null, - }, - }); - } else { - await prisma.measurement.create({ - data: { - userId, - type: SLEEP_TYPE, - value: minutes, - unit: getUnitForType(SLEEP_TYPE), - measuredAt, - source: "WITHINGS", - sleepStage: stage, - externalId, - }, - }); - insertedSleepMeasuredAts.push(measuredAt); - } + { exactExternalMatch: "update" }, + ); + if (verdict.status === "failed") { + throw new MeasurementReconciliationError(verdict); + } + for (const dirty of verdict.dirtyIdentities ?? []) { + touched.push(dirty); + } + if (verdict.status === "inserted") { + insertedSleepMeasuredAts.push(measuredAt); } touched.push({ type: SLEEP_TYPE, measuredAt }); imported++; - } catch (err) { - getEvent()?.addWarning( - `Failed to upsert sleep segment (${stage}, ${measuredAt.toISOString()}): ${err}`, - ); } - } + }, { timeout: 60_000 }); // Session-scoped sweep (mirrors Google Health's replace-by-window): for each // session this fetch re-produced, soft-delete any live WITHINGS @@ -550,69 +471,51 @@ export async function syncUserSleep( await sweepStaleSleepSegments(userId, "WITHINGS", sweeps); // v1.18.10 P0 — nightly sleep vitals (avg HR / respiratory rate / SDNN HRV - // / avg SpO2 / sleep score) from the per-night summary. A summary-fetch - // failure must not lose the stage segments already written, so it runs in - // its own try/catch and only annotates on error. Each vital upserts on the - // stable `(userId, type, source, externalId)` composite so a re-sync - // overwrites the night's value rather than inserting a duplicate. + // / avg SpO2 / sleep score) from the per-night summary. Fetch failure stays + // isolated from the stage writes; database failures are hard sync failures. + let summaries: WithingsSleepSummary[] = []; try { - const startYmd = ymdUtc(start); - const endYmd = ymdUtc(now); - const summaries = await fetchWithingsSleepSummary( + summaries = await fetchWithingsSleepSummary( tokenInfo.accessToken, - startYmd, - endYmd, + ymdUtc(start), + ymdUtc(now), ); - for (const summary of summaries) { - const measuredAt = new Date(summary.enddate * 1000); - const sessionId = summary.id ?? "no-id"; - const vitals = mapWithingsSleepSummary(summary.data); - for (const vital of vitals) { - const externalId = `withings:sleep:${userId}:${sessionId}:${vital.fieldTag}`; - try { - await prisma.measurement.upsert({ - where: { - userId_type_source_externalId: { - userId, - type: vital.type, - source: "WITHINGS", - externalId, - }, - }, - create: { - userId, - type: vital.type, - value: vital.value, - unit: getUnitForType(vital.type), - measuredAt, - source: "WITHINGS", - externalId, - }, - update: { - value: vital.value, - measuredAt, - // No-op on a live row, a deliberate RESURRECTION on a - // tombstoned one — Withings is the source of truth for its - // own rows (mirrors Google / Fitbit). - deletedAt: null, - syncVersion: { increment: 1 }, - }, - }); - touched.push({ type: vital.type, measuredAt }); - imported++; - } catch (err) { - getEvent()?.addWarning( - `withings sleep: failed to upsert vital (${vital.type}, ${measuredAt.toISOString()}): ${err}`, - ); - } - } - } } catch (err) { getEvent()?.addWarning( `withings sleep: summary fetch failed for ${userId}: ${err}`, ); } + await prisma.$transaction(async (tx) => { + for (const summary of summaries) { + const measuredAt = new Date(summary.enddate * 1000); + const sessionId = summary.id ?? "no-id"; + for (const vital of mapWithingsSleepSummary(summary.data)) { + const verdict = await reconcileExternalMeasurement( + tx, + { + userId, + type: vital.type, + value: vital.value, + unit: getUnitForType(vital.type), + measuredAt, + source: "WITHINGS", + externalId: `withings:sleep:${userId}:${sessionId}:${vital.fieldTag}`, + }, + { exactExternalMatch: "update" }, + ); + if (verdict.status === "failed") { + throw new MeasurementReconciliationError(verdict); + } + for (const dirty of verdict.dirtyIdentities ?? []) { + touched.push(dirty); + } + touched.push({ type: vital.type, measuredAt }); + imported++; + } + } + }, { timeout: 60_000 }); + // v1.4.39.1 — refresh the persistent rollup table for every distinct // (type, day) the sync touched. Sleep segments collapse heavily — // ~10 stage rows per night land in one DAY recompute. diff --git a/tests/integration/measurement-identity-reconcile.test.ts b/tests/integration/measurement-identity-reconcile.test.ts new file mode 100644 index 000000000..451c4b0f6 --- /dev/null +++ b/tests/integration/measurement-identity-reconcile.test.ts @@ -0,0 +1,224 @@ +import { beforeEach, describe, expect, it } from "vitest"; + +import { + reconcileExternalMeasurement, + type ExternalMeasurementWrite, +} from "@/lib/measurements/reconcile-external-measurement"; +import { getPrismaClient, truncateAllTables } from "./setup"; + +beforeEach(async () => { + await truncateAllTables(getPrismaClient()); +}); + +async function createUser(id: string) { + return getPrismaClient().user.create({ + data: { + id, + username: id, + email: `${id}@example.test`, + role: "USER", + }, + }); +} + +const measuredAt = new Date("2026-07-20T06:30:00.000Z"); + +function desired( + userId: string, + externalId: string, + overrides: Partial = {}, +): ExternalMeasurementWrite { + return { + userId, + type: "SLEEP_DURATION" as const, + value: 42, + unit: "minutes", + source: "OURA" as const, + measuredAt, + sleepStage: "DEEP" as const, + externalId, + externalSourceVersion: "v2", + deviceType: "ring", + ...overrides, + }; +} + +async function reconcile( + input: ExternalMeasurementWrite, + options?: { exactExternalMatch?: "update" | "duplicate" }, +) { + const prisma = getPrismaClient(); + return prisma.$transaction((tx) => + reconcileExternalMeasurement(tx, input, options), + ); +} + +describe("external measurement identity reconciliation (real Postgres)", () => { + it("moves an external-id hit onto an occupied natural key and retires the redundant row", async () => { + const prisma = getPrismaClient(); + const user = await createUser("identity-external-move"); + const external = await prisma.measurement.create({ + data: { + ...desired(user.id, "sleep:new"), + value: 30, + measuredAt: new Date("2026-07-20T06:00:00.000Z"), + externalSourceVersion: null, + }, + }); + const collision = await prisma.measurement.create({ + data: { + ...desired(user.id, "sleep:legacy"), + value: 35, + }, + }); + + const verdict = await reconcile(desired(user.id, "sleep:new")); + + expect(verdict).toMatchObject({ + status: "updated", + row: { id: external.id, externalId: "sleep:new" }, + retiredCollisionId: collision.id, + }); + const rows = await prisma.measurement.findMany({ + where: { userId: user.id }, + }); + expect(rows).toHaveLength(2); + expect(rows.find((row) => row.id === external.id)).toMatchObject({ + value: 42, + measuredAt, + externalId: "sleep:new", + externalSourceVersion: "v2", + deviceType: "ring", + deletedAt: null, + }); + expect(rows.find((row) => row.id === collision.id)).toMatchObject({ + externalId: "sleep:legacy", + deletedAt: expect.any(Date), + }); + }); + + it("adopts a live natural-key row when a new external id collides", async () => { + const prisma = getPrismaClient(); + const user = await createUser("identity-natural-live"); + const occupied = await prisma.measurement.create({ + data: { ...desired(user.id, "sleep:legacy"), value: 30 }, + }); + + const verdict = await reconcile(desired(user.id, "sleep:new")); + + expect(verdict).toMatchObject({ + status: "updated", + row: { id: occupied.id, externalId: "sleep:new" }, + }); + expect( + await prisma.measurement.count({ where: { userId: user.id } }), + ).toBe(1); + }); + + it("resurrects and re-keys a tombstone occupying the natural key", async () => { + const prisma = getPrismaClient(); + const user = await createUser("identity-natural-tombstone"); + const occupied = await prisma.measurement.create({ + data: { + ...desired(user.id, "sleep:legacy"), + value: 30, + deletedAt: new Date("2026-07-20T08:00:00.000Z"), + }, + }); + + const verdict = await reconcile(desired(user.id, "sleep:new")); + + expect(verdict).toMatchObject({ + status: "resurrected", + row: { id: occupied.id, externalId: "sleep:new" }, + }); + const stored = await prisma.measurement.findUniqueOrThrow({ + where: { id: occupied.id }, + }); + expect(stored.deletedAt).toBeNull(); + expect(stored.value).toBe(42); + }); + + it("never merges a row owned by another user", async () => { + const prisma = getPrismaClient(); + const first = await createUser("identity-owner-first"); + const second = await createUser("identity-owner-second"); + const other = await prisma.measurement.create({ + data: desired(first.id, "sleep:shared"), + }); + + const verdict = await reconcile(desired(second.id, "sleep:shared")); + + expect(verdict.status).toBe("inserted"); + expect(await prisma.measurement.count()).toBe(2); + expect( + await prisma.measurement.findUniqueOrThrow({ where: { id: other.id } }), + ).toMatchObject({ userId: first.id, value: 42 }); + }); + + it("serializes concurrent attempts that share the same natural identity", async () => { + const prisma = getPrismaClient(); + const user = await createUser("identity-concurrent"); + + const verdicts = await Promise.all([ + reconcile(desired(user.id, "sleep:race-a", { value: 41 })), + reconcile(desired(user.id, "sleep:race-b", { value: 43 })), + ]); + + expect(verdicts.every((verdict) => verdict.status !== "failed")).toBe(true); + expect(verdicts.map((verdict) => verdict.status).sort()).toEqual([ + "inserted", + "updated", + ]); + const rows = await prisma.measurement.findMany({ + where: { userId: user.id }, + }); + expect(rows).toHaveLength(1); + expect(["sleep:race-a", "sleep:race-b"]).toContain(rows[0]?.externalId); + expect([41, 43]).toContain(rows[0]?.value); + }); + + it("serializes crossing external and natural identity moves", async () => { + const prisma = getPrismaClient(); + const user = await createUser("identity-crossing-race"); + const firstTime = new Date("2026-07-20T06:00:00.000Z"); + const secondTime = new Date("2026-07-20T07:00:00.000Z"); + await prisma.measurement.create({ + data: desired(user.id, "sleep:race-a", { measuredAt: firstTime }), + }); + + const verdicts = await Promise.all([ + reconcile( + desired(user.id, "sleep:race-a", { + measuredAt: secondTime, + value: 41, + }), + ), + reconcile( + desired(user.id, "sleep:race-b", { + measuredAt: firstTime, + value: 43, + }), + ), + ]); + + expect(verdicts.every((verdict) => verdict.status !== "failed")).toBe(true); + const rows = await prisma.measurement.findMany({ + where: { userId: user.id, deletedAt: null }, + orderBy: { measuredAt: "asc" }, + }); + expect(rows).toHaveLength(2); + expect(rows).toEqual([ + expect.objectContaining({ + externalId: "sleep:race-b", + measuredAt: firstTime, + value: 43, + }), + expect.objectContaining({ + externalId: "sleep:race-a", + measuredAt: secondTime, + value: 41, + }), + ]); + }); +}); diff --git a/tests/integration/measurements-batch.test.ts b/tests/integration/measurements-batch.test.ts index 2daded94f..cd2c24d40 100644 --- a/tests/integration/measurements-batch.test.ts +++ b/tests/integration/measurements-batch.test.ts @@ -326,27 +326,17 @@ describe("POST /api/measurements/batch (real Postgres)", () => { expect(stored.every((r) => r.unit === "minutes")).toBe(true); }); - // v1.4.25 W10 reconcile (senior-dev H-1): under contention the - // per-entry envelope MUST stay in sync with the aggregate - // `inserted` / `duplicate` counts. The previous reconciliation - // block was an effective no-op (it looked for rows that were - // "marked inserted but missing from the DB" — an impossibility - // because the raced row IS in the DB, written by the other - // batch). The corrected block downgrades enough "inserted" - // statuses to "duplicate" so the per-entry envelope and the - // aggregate counts agree, which the iOS sync cursor depends on - // to checkpoint correctly. + // Under contention every per-entry verdict must stay in sync with its + // aggregate counter. Inserted, updated, duplicate, skipped, and failed are + // all explicit because only the hard failure verdict must block checkpoint + // advancement. it("keeps per-entry status in sync with aggregate counts under a concurrent-write race", async () => { const { POST } = await import("@/app/api/measurements/batch/route"); - // Two batches with overlapping externalIds posted in parallel. - // The composite unique index ensures only one row per - // (user, type, source, externalId) lands; `skipDuplicates` - // absorbs duplicate-key conflicts. The exact split between - // "won the race" and "got absorbed" depends on Postgres's - // commit order, so we assert only the invariant the iOS sync - // cursor relies on: per-entry envelope sums equal the aggregate - // counts for each response. + // Two batches with overlapping external ids and the same natural identity + // are posted in parallel. The reconciler serializes both identity indexes; + // the exact inserted/updated split depends on commit order, but every + // response must remain internally consistent. const sharedEntries = () => Array.from({ length: 6 }, (_, i) => ({ hkIdentifier: "HKQuantityTypeIdentifierBodyMass", @@ -368,6 +358,8 @@ describe("POST /api/measurements/batch (real Postgres)", () => { data: { processed: number; inserted: number; + updated: number; + failed: number; duplicates: number; entries: Array<{ status: string }>; }; @@ -379,19 +371,31 @@ describe("POST /api/measurements/batch (real Postgres)", () => { // no-op reconciliation left the envelope out of sync. for (const json of [firstJson, secondJson]) { const insertedEntries = json.data.entries.filter( - (e) => e.status === "inserted", + (entry) => entry.status === "inserted", + ).length; + const updatedEntries = json.data.entries.filter( + (entry) => entry.status === "updated", ).length; const duplicateEntries = json.data.entries.filter( - (e) => e.status === "duplicate", + (entry) => entry.status === "duplicate", ).length; const skippedEntries = json.data.entries.filter( - (e) => e.status === "skipped", + (entry) => entry.status === "skipped", + ).length; + const failedEntries = json.data.entries.filter( + (entry) => entry.status === "failed", ).length; expect(insertedEntries).toBe(json.data.inserted); + expect(updatedEntries).toBe(json.data.updated); expect(duplicateEntries).toBe(json.data.duplicates); - expect(insertedEntries + duplicateEntries + skippedEntries).toBe( - json.data.processed, - ); + expect(failedEntries).toBe(json.data.failed); + expect( + insertedEntries + + updatedEntries + + duplicateEntries + + skippedEntries + + failedEntries, + ).toBe(json.data.processed); } // Invariant 2 — aggregate counts are non-negative. The previous @@ -400,14 +404,12 @@ describe("POST /api/measurements/batch (real Postgres)", () => { expect(firstJson.data.inserted).toBeGreaterThanOrEqual(0); expect(secondJson.data.inserted).toBeGreaterThanOrEqual(0); - // Invariant 3 — the DB ends up with at most 6 rows because the - // composite unique index enforces single-copy, and the - // `inserted` counts across both requests cannot exceed the - // number of rows actually present. + // Invariant 3 — all samples share one natural identity, so exactly one + // canonical row survives and only one request can report its insertion. const stored = await getPrismaClient().measurement.findMany({ where: { userId: TEST_USER_ID }, }); - expect(stored.length).toBeLessThanOrEqual(6); + expect(stored).toHaveLength(1); expect(firstJson.data.inserted + secondJson.data.inserted).toBe( stored.length, );