From 94842406889bb9505850691f7306d02dfcca4f89 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Mon, 20 Jul 2026 13:24:49 +0200 Subject: [PATCH] fix(whoop): settle backfill side effects --- .../whoop/__tests__/sync-sleep-sweep.test.ts | 41 ++++++++++++++++++ src/lib/whoop/__tests__/sync.test.ts | 3 +- src/lib/whoop/sync-sleep.ts | 43 ++++++++++++------- src/lib/whoop/sync.ts | 29 +++++++++---- 4 files changed, 90 insertions(+), 26 deletions(-) diff --git a/src/lib/whoop/__tests__/sync-sleep-sweep.test.ts b/src/lib/whoop/__tests__/sync-sleep-sweep.test.ts index 2582fc6b2..8e7cf6dab 100644 --- a/src/lib/whoop/__tests__/sync-sleep-sweep.test.ts +++ b/src/lib/whoop/__tests__/sync-sleep-sweep.test.ts @@ -164,6 +164,47 @@ describe("syncUserSleep — record-scoped stale-segment sweep", () => { expect(upsertMeasurementsMock).toHaveBeenCalledTimes(1); }); + it("enqueues one refresh after every committed chunk settles", async () => { + fetchSleepsMock.mockResolvedValue([NIGHT]); + const firstMeasuredAt = new Date("2026-06-01T06:30:00.000Z"); + const secondMeasuredAt = new Date("2026-06-01T07:00:00.000Z"); + upsertMeasurementsMock.mockImplementationOnce( + async ( + _userId: string, + _readings: Array<{ type: string; measuredAt: Date }>, + opts?: { + onInserted?: ( + rows: Array<{ id: string; type: string; measuredAt: Date }>, + ) => void; + }, + ) => { + opts?.onInserted?.([ + { + id: "first-committed-segment", + type: "SLEEP_DURATION", + measuredAt: firstMeasuredAt, + }, + ]); + opts?.onInserted?.([ + { + id: "second-committed-segment", + type: "SLEEP_DURATION", + measuredAt: secondMeasuredAt, + }, + ]); + return 2; + }, + ); + + await expect(syncUserSleep("user-1")).resolves.toBe(2); + + expect(morningRefreshMock).toHaveBeenCalledTimes(1); + expect(morningRefreshMock).toHaveBeenCalledWith("user-1", [ + firstMeasuredAt, + secondMeasuredAt, + ]); + }); + it("enqueues refreshes for a committed chunk before a later chunk fails", async () => { fetchSleepsMock.mockResolvedValue([NIGHT]); const measuredAt = new Date("2026-06-01T07:00:00.000Z"); diff --git a/src/lib/whoop/__tests__/sync.test.ts b/src/lib/whoop/__tests__/sync.test.ts index c9ab76082..8d09b9239 100644 --- a/src/lib/whoop/__tests__/sync.test.ts +++ b/src/lib/whoop/__tests__/sync.test.ts @@ -603,7 +603,7 @@ describe("syncUserRecovery — durable cursor ordering", () => { expect(prismaMock.$transaction.mock.calls.length).toBeGreaterThan(1); expect(prismaMock.$executeRaw).not.toHaveBeenCalled(); expect(recomputeBucketsMock).toHaveBeenCalled(); - expect(invalidateStatusMock).toHaveBeenCalled(); + expect(invalidateStatusMock).toHaveBeenCalledTimes(1); failLaterChunk = false; await expect(syncUserRecovery("user1")).resolves.toBe(105); @@ -625,6 +625,7 @@ describe("syncUserRecovery — durable cursor ordering", () => { expect(committedInsertCount).toBe(105); expect([...committedRows.keys()]).toEqual(expectedExternalIds); expect(emittedExternalIds).toEqual(expectedExternalIds); + expect(invalidateStatusMock).toHaveBeenCalledTimes(2); expect(prismaMock.$executeRaw).toHaveBeenCalledTimes(1); }); diff --git a/src/lib/whoop/sync-sleep.ts b/src/lib/whoop/sync-sleep.ts index 2a263dc79..251ecc96a 100644 --- a/src/lib/whoop/sync-sleep.ts +++ b/src/lib/whoop/sync-sleep.ts @@ -40,15 +40,11 @@ import { } from "./sync"; import { prisma } from "@/lib/db"; -function enqueueMorningRefreshForInsertedSleepRows( +function enqueueMorningRefreshForMeasuredAts( userId: string, - rows: ReadonlyArray<{ type: string; measuredAt: Date }>, + measuredAts: Date[], ): void { - const measuredAts = rows - .filter((row) => row.type === "SLEEP_DURATION") - .map((row) => row.measuredAt); if (measuredAts.length === 0) return; - void maybeEnqueueMorningRefresh(userId, measuredAts).catch(() => {}); } @@ -106,13 +102,24 @@ export async function syncUserSleep( // the fresh set upserts (mirrors Google Health's replace-by-window order). await sweepStaleSleepSegments(userId, "WHOOP", sweeps); - // Each committed chunk triggers independently; a later chunk failure cannot - // strand inserted sleep rows without their debounced morning refresh. - const imported = await upsertWhoopMeasurements(userId, readings, { - onInserted: (rows) => { - enqueueMorningRefreshForInsertedSleepRows(userId, rows); - }, - }); + // Accumulate committed chunks, then trigger once after the write pass settles. + // The finally path also covers a later chunk failure without generating from + // an intermediate prefix while more chunks are still being written. + const insertedSleepMeasuredAts: Date[] = []; + let imported: number; + try { + imported = await upsertWhoopMeasurements(userId, readings, { + onInserted: (rows) => { + insertedSleepMeasuredAts.push( + ...rows + .filter((row) => row.type === "SLEEP_DURATION") + .map((row) => row.measuredAt), + ); + }, + }); + } finally { + enqueueMorningRefreshForMeasuredAts(userId, insertedSleepMeasuredAts); + } await markResourceSynced(userId, "sleep"); return imported; @@ -162,12 +169,16 @@ export async function syncWhoopSleepById( { prefix: `${record.id}:`, keepIds }, ]); - // Dispatch the targeted refresh as soon as its insert commits, matching the - // chunked collection path above. + const insertedSleepMeasuredAts: Date[] = []; const imported = await upsertWhoopMeasurements(userId, readings, { onInserted: (rows) => { - enqueueMorningRefreshForInsertedSleepRows(userId, rows); + insertedSleepMeasuredAts.push( + ...rows + .filter((row) => row.type === "SLEEP_DURATION") + .map((row) => row.measuredAt), + ); }, }); + enqueueMorningRefreshForMeasuredAts(userId, insertedSleepMeasuredAts); return imported; } diff --git a/src/lib/whoop/sync.ts b/src/lib/whoop/sync.ts index cfe5403fa..a058d6cb7 100644 --- a/src/lib/whoop/sync.ts +++ b/src/lib/whoop/sync.ts @@ -345,6 +345,18 @@ export async function upsertWhoopMeasurements( if (readings.length === 0) return 0; let imported = 0; + const touchedTypes = new Set(); + const invalidateTouchedTypes = (): void => { + if (touchedTypes.size === 0) return; + + void invalidateStatusInsightsForTypes(userId, [...touchedTypes]).catch( + (err) => { + getEvent()?.addWarning( + `whoop: status-insight invalidate failed for ${userId}: ${err}`, + ); + }, + ); + }; for ( let chunkStart = 0; @@ -382,7 +394,10 @@ export async function upsertWhoopMeasurements( return outcomes; }, { maxWait: 10_000, timeout: 60_000 }, - ); + ).catch((err) => { + invalidateTouchedTypes(); + throw err; + }); const touched: Array<{ type: MeasurementType; measuredAt: Date }> = []; const insertedRows: Array< @@ -393,8 +408,10 @@ export async function upsertWhoopMeasurements( const verdict = verdicts[index - chunkStart]!; for (const dirty of verdict.dirtyIdentities ?? []) { touched.push(dirty); + touchedTypes.add(dirty.type); } const type = reading.type as MeasurementType; + touchedTypes.add(type); imported++; touched.push({ type, measuredAt: reading.measuredAt }); if (verdict.status === "inserted") { @@ -411,14 +428,6 @@ export async function upsertWhoopMeasurements( for (const k of keys) { await recomputeBucketsForMeasurement(userId, k.type, k.measuredAt); } - invalidateStatusInsightsForTypes( - userId, - keys.map((k) => k.type), - ).catch((err) => { - getEvent()?.addWarning( - `whoop: status-insight invalidate failed for ${userId}: ${err}`, - ); - }); } catch (err) { getEvent()?.addWarning( `whoop: rollup recompute failed for ${userId}: ${err}`, @@ -426,6 +435,8 @@ export async function upsertWhoopMeasurements( } } + invalidateTouchedTypes(); + return imported; }