fix(whoop): settle backfill side effects

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");

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

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

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