fix(whoop): fold committed backfill chunks

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]);
+  });
 });

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<void>>(
+    async () => {},
+  ),
+  invalidateStatusMock: vi.fn<(...a: unknown[]) => Promise<void>>(
+    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);

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;
 }

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;