From 1b717b4ca02685d682bfa1f04449aa1277b44065 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Mon, 20 Jul 2026 13:16:34 +0200 Subject: [PATCH] fix(whoop): fold committed backfill chunks --- .../whoop/__tests__/sync-sleep-sweep.test.ts | 69 ++++++++++++++----- src/lib/whoop/__tests__/sync.test.ts | 16 ++++- src/lib/whoop/sync-sleep.ts | 40 +++++------ src/lib/whoop/sync.ts | 34 ++++----- 4 files changed, 101 insertions(+), 58 deletions(-) diff --git a/src/lib/whoop/__tests__/sync-sleep-sweep.test.ts b/src/lib/whoop/__tests__/sync-sleep-sweep.test.ts index d824579a0..2582fc6b2 100644 --- a/src/lib/whoop/__tests__/sync-sleep-sweep.test.ts +++ b/src/lib/whoop/__tests__/sync-sleep-sweep.test.ts @@ -9,24 +9,31 @@ */ import { beforeEach, describe, expect, it, vi } from "vitest"; -const { updateManyMock, upsertMeasurementsMock } = vi.hoisted(() => ({ - updateManyMock: vi.fn(async () => ({ count: 0 })), - upsertMeasurementsMock: vi.fn( - async ( - _userId: string, - readings: Array<{ type: string; measuredAt: Date }>, - opts?: { - onInserted?: ( - rows: Array<{ id: string; type: string; measuredAt: Date }>, - ) => void; +const { updateManyMock, upsertMeasurementsMock, morningRefreshMock } = + vi.hoisted(() => ({ + updateManyMock: vi.fn(async () => ({ count: 0 })), + upsertMeasurementsMock: vi.fn( + async ( + _userId: string, + readings: Array<{ type: string; measuredAt: Date }>, + opts?: { + onInserted?: ( + rows: Array<{ id: string; type: string; measuredAt: Date }>, + ) => void; + }, + ) => { + opts?.onInserted?.( + readings.map((row, index) => ({ ...row, id: `inserted-${index}` })), + ); + return 1; }, - ) => { - opts?.onInserted?.( - readings.map((row, index) => ({ ...row, id: `inserted-${index}` })), - ); - return 1; - }, - ), + ), + morningRefreshMock: vi.fn(async () => {}), + })); + +vi.mock("@/lib/daily/morning-refresh-trigger", () => ({ + maybeEnqueueMorningRefresh: (...args: unknown[]) => + morningRefreshMock(...args), })); vi.mock("@/lib/db", () => ({ @@ -92,6 +99,7 @@ beforeEach(() => { updateManyMock.mockClear().mockResolvedValue({ count: 0 }); upsertMeasurementsMock.mockClear(); fetchSleepsMock.mockReset(); + morningRefreshMock.mockClear(); }); describe("syncUserSleep — record-scoped stale-segment sweep", () => { @@ -155,4 +163,31 @@ describe("syncUserSleep — record-scoped stale-segment sweep", () => { await expect(syncUserSleep("user-1")).resolves.toBe(1); expect(upsertMeasurementsMock).toHaveBeenCalledTimes(1); }); + + 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"); + 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: "committed-segment", type: "SLEEP_DURATION", measuredAt }, + ]); + throw new Error("injected later-chunk failure"); + }, + ); + + await expect(syncUserSleep("user-1")).rejects.toThrow( + "injected later-chunk failure", + ); + + expect(morningRefreshMock).toHaveBeenCalledWith("user-1", [measuredAt]); + }); }); diff --git a/src/lib/whoop/__tests__/sync.test.ts b/src/lib/whoop/__tests__/sync.test.ts index bbdf84e29..c9ab76082 100644 --- a/src/lib/whoop/__tests__/sync.test.ts +++ b/src/lib/whoop/__tests__/sync.test.ts @@ -17,6 +17,8 @@ const { isReauthRequired, emitArrivalMock, reconcileMock, + recomputeBucketsMock, + invalidateStatusMock, } = vi.hoisted(() => ({ prismaMock: { whoopConnection: { @@ -43,6 +45,12 @@ const { ), reconcileMock: vi.fn(), emitArrivalMock: vi.fn(), + recomputeBucketsMock: vi.fn<(...a: unknown[]) => Promise>( + async () => {}, + ), + invalidateStatusMock: vi.fn<(...a: unknown[]) => Promise>( + async () => {}, + ), })); vi.mock("@/lib/db", () => ({ prisma: prismaMock })); @@ -82,11 +90,13 @@ vi.mock("@/lib/integrations/status", () => ({ vi.mock("@/lib/rollups/measurement-rollups", () => ({ collapseToTypeDayKeys: (rows: Array<{ type: string; measuredAt: Date }>) => rows.map((r) => ({ type: r.type, measuredAt: r.measuredAt })), - recomputeBucketsForMeasurement: vi.fn(async () => {}), + recomputeBucketsForMeasurement: (...a: unknown[]) => + recomputeBucketsMock(...a), })); vi.mock("@/lib/insights/comprehensive-generate", () => ({ - invalidateStatusInsightsForTypes: vi.fn(async () => {}), + invalidateStatusInsightsForTypes: (...a: unknown[]) => + invalidateStatusMock(...a), })); vi.mock("@/lib/arrivals/measurement-emit", () => ({ @@ -592,6 +602,8 @@ describe("syncUserRecovery — durable cursor ordering", () => { expect(committedRows.size).toBeLessThan(105); expect(prismaMock.$transaction.mock.calls.length).toBeGreaterThan(1); expect(prismaMock.$executeRaw).not.toHaveBeenCalled(); + expect(recomputeBucketsMock).toHaveBeenCalled(); + expect(invalidateStatusMock).toHaveBeenCalled(); failLaterChunk = false; await expect(syncUserRecovery("user1")).resolves.toBe(105); diff --git a/src/lib/whoop/sync-sleep.ts b/src/lib/whoop/sync-sleep.ts index 18711f9c3..2a263dc79 100644 --- a/src/lib/whoop/sync-sleep.ts +++ b/src/lib/whoop/sync-sleep.ts @@ -40,6 +40,18 @@ import { } from "./sync"; import { prisma } from "@/lib/db"; +function enqueueMorningRefreshForInsertedSleepRows( + userId: string, + rows: ReadonlyArray<{ type: string; measuredAt: 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(() => {}); +} + export async function syncUserSleep( userId: string, opts: { fullSync?: boolean } = {}, @@ -94,23 +106,15 @@ export async function syncUserSleep( // the fresh set upserts (mirrors Google Health's replace-by-window order). await sweepStaleSleepSegments(userId, "WHOOP", sweeps); - const insertedSleepMeasuredAts: Date[] = []; + // 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) => { - insertedSleepMeasuredAts.push( - ...rows - .filter((row) => row.type === "SLEEP_DURATION") - .map((row) => row.measuredAt), - ); + enqueueMorningRefreshForInsertedSleepRows(userId, rows); }, }); await markResourceSynced(userId, "sleep"); - // S4 — trigger the debounced morning refresh on a last-night segment landing. - void maybeEnqueueMorningRefresh(userId, insertedSleepMeasuredAts).catch( - () => {}, - ); - return imported; } @@ -158,20 +162,12 @@ export async function syncWhoopSleepById( { prefix: `${record.id}:`, keepIds }, ]); - let insertedSleepMeasuredAts: Date[] = []; + // Dispatch the targeted refresh as soon as its insert commits, matching the + // chunked collection path above. const imported = await upsertWhoopMeasurements(userId, readings, { onInserted: (rows) => { - insertedSleepMeasuredAts = rows - .filter((row) => row.type === "SLEEP_DURATION") - .map((row) => row.measuredAt); + enqueueMorningRefreshForInsertedSleepRows(userId, rows); }, }); - - // S4 — a webhook-driven single-record refresh is the freshest possible - // last-night signal; kick the debounced morning refresh on its segments. - void maybeEnqueueMorningRefresh(userId, insertedSleepMeasuredAts).catch( - () => {}, - ); - return imported; } diff --git a/src/lib/whoop/sync.ts b/src/lib/whoop/sync.ts index 820fd15aa..cfe5403fa 100644 --- a/src/lib/whoop/sync.ts +++ b/src/lib/whoop/sync.ts @@ -345,7 +345,6 @@ export async function upsertWhoopMeasurements( if (readings.length === 0) return 0; let imported = 0; - const touched: Array<{ type: MeasurementType; measuredAt: Date }> = []; for ( let chunkStart = 0; @@ -385,6 +384,7 @@ export async function upsertWhoopMeasurements( { maxWait: 10_000, timeout: 60_000 }, ); + const touched: Array<{ type: MeasurementType; measuredAt: Date }> = []; const insertedRows: Array< InsertedMeasurementArrivalRow & { externalId: string | null } > = []; @@ -406,24 +406,24 @@ export async function upsertWhoopMeasurements( void emitInsertedMeasurementArrivals(userId, insertedRows, "whoop").catch( () => {}, ); - } - try { - const keys = collapseToTypeDayKeys(touched); - for (const k of keys) { - await recomputeBucketsForMeasurement(userId, k.type, k.measuredAt); - } - invalidateStatusInsightsForTypes( - userId, - keys.map((k) => k.type), - ).catch((err) => { + try { + const keys = collapseToTypeDayKeys(touched); + 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: status-insight invalidate failed for ${userId}: ${err}`, + `whoop: rollup recompute failed for ${userId}: ${err}`, ); - }); - } catch (err) { - getEvent()?.addWarning( - `whoop: rollup recompute failed for ${userId}: ${err}`, - ); + } } return imported;