From 197f7bf5a11614026a6fbbfe7df4b1af3a8a9d4c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Mon, 20 Jul 2026 13:08:48 +0200 Subject: [PATCH] fix(whoop): persist backfill chunks durably --- src/lib/whoop/__tests__/sync.test.ts | 86 +++++++++++++++++++++ src/lib/whoop/sync-sleep.ts | 10 ++- src/lib/whoop/sync.ts | 109 +++++++++++++++------------ 3 files changed, 153 insertions(+), 52 deletions(-) diff --git a/src/lib/whoop/__tests__/sync.test.ts b/src/lib/whoop/__tests__/sync.test.ts index 6416e109c..bbdf84e29 100644 --- a/src/lib/whoop/__tests__/sync.test.ts +++ b/src/lib/whoop/__tests__/sync.test.ts @@ -530,6 +530,92 @@ describe("syncUserRecovery — durable cursor ordering", () => { expect(prismaMock.$executeRaw).not.toHaveBeenCalled(); }); + it("keeps earlier chunks durable and resumes without duplicate rows after a later chunk fails", async () => { + const records = Array.from({ length: 21 }, (_, index) => ({ + ...recovery, + cycle_id: index + 1, + sleep_id: `sleep-${index.toString().padStart(2, "0")}`, + })); + fetchRecoveriesMock.mockResolvedValue(records); + + type StoredRow = { + id: string; + type: string; + measuredAt: Date; + externalId: string; + }; + type TransactionState = { rows: Map }; + + let committedRows = new Map(); + let committedInsertCount = 0; + let failLaterChunk = true; + + prismaMock.$transaction.mockImplementation( + async (run: (tx: TransactionState) => unknown) => { + const stagedRows = new Map(committedRows); + const result = await run({ rows: stagedRows }); + committedInsertCount += [...stagedRows.keys()].filter( + (key) => !committedRows.has(key), + ).length; + committedRows = stagedRows; + return result; + }, + ); + reconcileMock.mockImplementation( + async ( + tx: TransactionState, + input: { type: string; measuredAt: Date; externalId: string }, + ) => { + if (failLaterChunk && input.externalId === "sleep-20:recovery") { + return { + status: "failed", + reason: "db_error", + error: new Error("injected later-chunk failure"), + }; + } + + const existing = tx.rows.get(input.externalId); + const row = { + id: existing?.id ?? `row:${input.externalId}`, + type: input.type, + measuredAt: input.measuredAt, + externalId: input.externalId, + }; + tx.rows.set(input.externalId, row); + return { status: existing ? "updated" : "inserted", row }; + }, + ); + + await expect(syncUserRecovery("user1")).rejects.toThrow(); + + expect(committedRows.size).toBeGreaterThan(0); + expect(committedRows.size).toBeLessThan(105); + expect(prismaMock.$transaction.mock.calls.length).toBeGreaterThan(1); + expect(prismaMock.$executeRaw).not.toHaveBeenCalled(); + + failLaterChunk = false; + await expect(syncUserRecovery("user1")).resolves.toBe(105); + + const expectedExternalIds = records.flatMap(({ sleep_id }) => + ["recovery", "hrv_rmssd", "rhr", "spo2", "skin_temp"].map( + (fieldTag) => `${sleep_id}:${fieldTag}`, + ), + ); + const emittedExternalIds = emitArrivalMock.mock.calls.flatMap((call) => + ( + call[1] as Array<{ + externalId: string; + }> + ).map((row) => row.externalId), + ); + + expect(committedRows).toHaveLength(105); + expect(committedInsertCount).toBe(105); + expect([...committedRows.keys()]).toEqual(expectedExternalIds); + expect(emittedExternalIds).toEqual(expectedExternalIds); + expect(prismaMock.$executeRaw).toHaveBeenCalledTimes(1); + }); + it("advances after success and replay updates identities without new arrivals", async () => { expect(await syncUserRecovery("user1")).toBe(5); expect(prismaMock.$executeRaw).toHaveBeenCalledTimes(1); diff --git a/src/lib/whoop/sync-sleep.ts b/src/lib/whoop/sync-sleep.ts index 89fdbe36d..18711f9c3 100644 --- a/src/lib/whoop/sync-sleep.ts +++ b/src/lib/whoop/sync-sleep.ts @@ -94,12 +94,14 @@ export async function syncUserSleep( // the fresh set upserts (mirrors Google Health's replace-by-window order). await sweepStaleSleepSegments(userId, "WHOOP", sweeps); - let insertedSleepMeasuredAts: Date[] = []; + const insertedSleepMeasuredAts: Date[] = []; const imported = await upsertWhoopMeasurements(userId, readings, { onInserted: (rows) => { - insertedSleepMeasuredAts = rows - .filter((row) => row.type === "SLEEP_DURATION") - .map((row) => row.measuredAt); + insertedSleepMeasuredAts.push( + ...rows + .filter((row) => row.type === "SLEEP_DURATION") + .map((row) => row.measuredAt), + ); }, }); await markResourceSynced(userId, "sleep"); diff --git a/src/lib/whoop/sync.ts b/src/lib/whoop/sync.ts index c275a49da..820fd15aa 100644 --- a/src/lib/whoop/sync.ts +++ b/src/lib/whoop/sync.ts @@ -323,10 +323,12 @@ export interface WhoopMeasurementUpsert { sleepStage?: "CORE" | "DEEP" | "REM" | "AWAKE" | "IN_BED" | null; } +const WHOOP_MEASUREMENT_TRANSACTION_CHUNK_SIZE = 100; + /** - * 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). The shared reconciler protects both external and + * Upsert mapped WHOOP readings for one user in bounded transactions, then fold + * the rollup tier + invalidate status-insight caches once at the end (mirrors + * 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. * @@ -344,56 +346,67 @@ export async function upsertWhoopMeasurements( let imported = 0; const touched: Array<{ type: MeasurementType; measuredAt: Date }> = []; - const insertedRows: Array< - InsertedMeasurementArrivalRow & { externalId: string | null } - > = []; - 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, - }, - { exactExternalMatch: "update" }, - ); - if (verdict.status === "failed") { - throw new MeasurementReconciliationError(verdict); + for ( + let chunkStart = 0; + chunkStart < readings.length; + chunkStart += WHOOP_MEASUREMENT_TRANSACTION_CHUNK_SIZE + ) { + const chunkEnd = Math.min( + chunkStart + WHOOP_MEASUREMENT_TRANSACTION_CHUNK_SIZE, + readings.length, + ); + const verdicts = await prisma.$transaction( + async (tx) => { + const outcomes = []; + for (let index = chunkStart; index < chunkEnd; index++) { + const reading = readings[index]!; + const verdict = await reconcileExternalMeasurement( + tx, + { + userId, + type: reading.type as MeasurementType, + source: "WHOOP", + value: reading.value, + unit: reading.unit, + measuredAt: reading.measuredAt, + externalId: reading.externalId, + sleepStage: reading.sleepStage ?? null, + }, + { exactExternalMatch: "update" }, + ); + if (verdict.status === "failed") { + throw new MeasurementReconciliationError(verdict); + } + outcomes.push(verdict); } - outcomes.push(verdict); + return outcomes; + }, + { maxWait: 10_000, timeout: 60_000 }, + ); + + const insertedRows: Array< + InsertedMeasurementArrivalRow & { externalId: string | null } + > = []; + for (let index = chunkStart; index < chunkEnd; index++) { + const reading = readings[index]!; + const verdict = verdicts[index - chunkStart]!; + for (const dirty of verdict.dirtyIdentities ?? []) { + touched.push(dirty); } - return outcomes; - }, - { maxWait: 10_000, timeout: 60_000 }, - ); + const type = reading.type as MeasurementType; + imported++; + touched.push({ type, measuredAt: reading.measuredAt }); + if (verdict.status === "inserted") { + insertedRows.push(verdict.row); + } + } - 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); - } + opts.onInserted?.(insertedRows); + void emitInsertedMeasurementArrivals(userId, insertedRows, "whoop").catch( + () => {}, + ); } - - opts.onInserted?.(insertedRows); - void emitInsertedMeasurementArrivals(userId, insertedRows, "whoop").catch( - () => {}, - ); try { const keys = collapseToTypeDayKeys(touched); for (const k of keys) {