fix(whoop): persist backfill chunks durably

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<string, StoredRow> };
+
+    let committedRows = new Map<string, StoredRow>();
+    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);

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

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) {