fix(withings): sync ECG on a schedule and surface write failures

Add an hourly ECG catch-net cron so a watch-only account with no scale pulls its
recordings without a webhook: the queue was registered and bound but never
scheduled, and the handler now grows a cohort arm that walks every connection on
a payload-less tick, mirroring the activity and sleep handlers.

Make the activity and sleep write paths honest. A swallowed activity batch-write
fell through to the success stamp — a green pill over lost data, green forever on
a persistent write bug; it now records a transient failure and refuses the
stamp. A reconcile throw out of either sleep transaction propagated past the
stamp and was swallowed by both callers, leaving neither success nor failure
recorded; the source now records a transient failure and rethrows so the cron
and webhook paths stay honest without changing their warn-and-continue
behaviour.

src/lib/jobs/__tests__/withings-ecg-worker.test.ts

@@ -52,9 +52,15 @@ vi.mock("@/lib/logging/background", () => ({
 vi.mock("@/lib/logging/fire-and-forget", () => ({
   fireAndForget: vi.fn(),
 }));
+const { ecgFindMany } = vi.hoisted(() => ({
+  ecgFindMany: vi.fn<(...a: unknown[]) => Promise<Array<{ userId: string }>>>(
+    async () => [],
+  ),
+}));
+
 vi.mock("../reminder/shared", () => ({
   getWorkerPrisma: vi.fn(() => ({
-    withingsConnection: { findMany: vi.fn(async () => []) },
+    withingsConnection: { findMany: (...a: unknown[]) => ecgFindMany(...a) },
   })),
 }));
 
@@ -69,6 +75,7 @@ function job(data: WithingsEcgSyncPayload): Job<WithingsEcgSyncPayload> {
 
 beforeEach(() => {
   vi.clearAllMocks();
+  ecgFindMany.mockResolvedValue([]);
 });
 const registrarSource = readFileSync(
   join(process.cwd(), "src/lib/jobs/reminder/register-integration-sync.ts"),
@@ -76,10 +83,8 @@ const registrarSource = readFileSync(
 );
 
 describe("handleWithingsEcgSync", () => {
-  it("rejects a failed source write so pg-boss can retry the same job", async () => {
-    syncUserEcg
-      .mockRejectedValueOnce(new Error("source write failed"))
-      .mockResolvedValueOnce(1);
+  it("a webhook job (userId + window) does a targeted date-windowed sync", async () => {
+    syncUserEcg.mockResolvedValue(1);
     const queued = job({
       userId: "user-1",
       eventId: "wu-1:1:1715000000:1715000060",
@@ -88,20 +93,42 @@ describe("handleWithingsEcgSync", () => {
       enddate: 1715000060,
     });
 
-    await expect(handleWithingsEcgSync([queued])).rejects.toThrow(
-      "source write failed",
-    );
     await expect(handleWithingsEcgSync([queued])).resolves.toBeUndefined();
 
+    expect(ecgFindMany).not.toHaveBeenCalled();
+    expect(syncUserEcg).toHaveBeenCalledTimes(1);
+    expect(syncUserEcg).toHaveBeenCalledWith("user-1", {
+      startdate: 1715000000,
+      enddate: 1715000060,
+    });
+  });
+
+  it("a payload-less cron tick walks the connection cohort with the default window", async () => {
+    // W-1 — the catch-net that lets a watch-only account (no scale, appli 54
+    // unsubscribed) pull its ECG strips: no userId → iterate every connection
+    // with `syncUserEcg(userId)` (default trailing-30-day window).
+    ecgFindMany.mockResolvedValue([{ userId: "a" }, { userId: "b" }]);
+    syncUserEcg.mockResolvedValue(0);
+
+    await handleWithingsEcgSync([job({} as WithingsEcgSyncPayload)]);
+
+    expect(ecgFindMany).toHaveBeenCalledTimes(1);
+    expect(syncUserEcg).toHaveBeenCalledTimes(2);
+    expect(syncUserEcg).toHaveBeenNthCalledWith(1, "a", {});
+    expect(syncUserEcg).toHaveBeenNthCalledWith(2, "b", {});
+  });
+
+  it("one user's failure never starves the rest of the cohort", async () => {
+    ecgFindMany.mockResolvedValue([{ userId: "a" }, { userId: "b" }]);
+    syncUserEcg.mockRejectedValueOnce(new Error("boom for a"));
+
+    // Per-user catch-and-warn (mirrors the activity / sleep handlers): the
+    // handler resolves, both users attempted, the leaf records its own ledger
+    // failure so honesty is preserved without rejecting the whole batch.
+    await expect(
+      handleWithingsEcgSync([job({} as WithingsEcgSyncPayload)]),
+    ).resolves.toBeUndefined();
     expect(syncUserEcg).toHaveBeenCalledTimes(2);
-    expect(syncUserEcg).toHaveBeenNthCalledWith(1, "user-1", {
-      startdate: 1715000000,
-      enddate: 1715000060,
-    });
-    expect(syncUserEcg).toHaveBeenNthCalledWith(2, "user-1", {
-      startdate: 1715000000,
-      enddate: 1715000060,
-    });
   });
 });
 

src/lib/jobs/__tests__/withings-queues.test.ts

@@ -76,4 +76,15 @@ describe("reminder-worker — Withings activity + sleep v2 queues", () => {
     );
     expect(source).toMatch(/\[WITHINGS_SLEEP_QUEUE,\s*WITHINGS_SLEEP_CRON\]/);
   });
+
+  it("schedules the ECG catch-net cron at :41 so watch-only accounts pull strips", () => {
+    // The queue is already in allQueues + boss.work-bound; W-1 adds the missing
+    // schedule entry so a payload-less cron tick walks the connection cohort.
+    expect(source).toMatch(
+      /WITHINGS_ECG_SYNC_CRON\s*=\s*["']41 \* \* \* \*["']/,
+    );
+    expect(source).toMatch(
+      /\[WITHINGS_ECG_SYNC_QUEUE,\s*WITHINGS_ECG_SYNC_CRON\]/,
+    );
+  });
 });

src/lib/jobs/reminder/register-integration-sync.ts

@@ -137,6 +137,14 @@ const WITHINGS_ACTIVITY_CRON = "0 * * * *"; // every hour at :00
 const WITHINGS_SLEEP_QUEUE = "withings-sleep-sync";
 
 const WITHINGS_SLEEP_CRON = "15 * * * *"; // every hour at :15
+// v1.32.23 — Withings ECG cron catch-net. The queue is webhook-primary
+// (appli=54 → `enqueueWithingsEcgSync`), but appli 54 is not in the
+// subscription set and the manual sync route measures only, so a ScanWatch-only
+// account (no scale) never had its ECG strips pulled. An hourly cohort walk
+// (one `v2/heart list` call per connected user, empty + cheap for non-ECG
+// devices) closes the gap. :41 is a free minute (used: 0,5,8,11,13,15,17,18,
+// 20,30,35,50) staggered off every other hourly sync tick.
+const WITHINGS_ECG_SYNC_CRON = "41 * * * *"; // every hour at :41
 
 const MOODLOG_SYNC_QUEUE = "moodlog-sync";
 
@@ -310,6 +318,9 @@ const schedules: ScheduleEntry[] = [
   [WITHINGS_SYNC_QUEUE, WITHINGS_SYNC_CRON],
   [WITHINGS_ACTIVITY_QUEUE, WITHINGS_ACTIVITY_CRON],
   [WITHINGS_SLEEP_QUEUE, WITHINGS_SLEEP_CRON],
+  // v1.32.23 — hourly ECG catch-net (:41) so watch-only accounts pull their
+  // strips without a webhook or a scale.
+  [WITHINGS_ECG_SYNC_QUEUE, WITHINGS_ECG_SYNC_CRON],
   [MOODLOG_SYNC_QUEUE, MOODLOG_SYNC_CRON],
   [WITHINGS_OAUTH_STATE_CLEANUP_QUEUE, WITHINGS_OAUTH_STATE_CLEANUP_CRON],
   // v1.11.0 — WHOOP poll-fallback crons. Recovery/sleep/workout catch

src/lib/jobs/reminder/withings-sync.ts

@@ -108,25 +108,72 @@ export async function handleWithingsFallbackSync(
 }
 
 /**
- * Drain webhook-owned ECG events. A failed source read or write is allowed to
- * reject the handler so pg-boss retains the job and applies its retry policy.
+ * ECG sync handler. Two enqueue paths feed this queue, mirroring the
+ * activity / sleep handlers:
+ *
+ *   1. Webhook (appli=54) — payload carries `userId` (+ the source
+ *      `startdate`/`enddate` window); the handler runs `syncUserEcg` for that
+ *      one user over the windowed range.
+ *   2. Cron (`withings-ecg-sync` at :41) — payload has no `userId`, the handler
+ *      iterates every Withings connection and runs `syncUserEcg` over its
+ *      default trailing 30-day window. This is the catch-net that lets a
+ *      watch-only account (no scale, appli 54 unsubscribed) pull its ECG strips
+ *      without a webhook.
+ *
+ * Per-user failures are warned and the cohort carries on so one user's
+ * parked-at-reauth state (or a transient device outage) never starves the rest;
+ * the ECG leaf records its own classified failure on the shared `withings`
+ * ledger, so honesty is preserved without rejecting the whole batch.
  */
 export async function handleWithingsEcgSync(
   jobs: Job<WithingsEcgSyncPayload>[],
 ) {
   await withBackgroundEvent("job.withings_ecg_sync", async (evt) => {
+    const prisma = getWorkerPrisma();
     try {
-      let recordingsImported = 0;
+      const targets: Array<{
+        userId: string;
+        startdate?: number;
+        enddate?: number;
+      }> = [];
       for (const job of jobs) {
-        recordingsImported += await syncUserEcg(job.data.userId, {
-          startdate: job.data.startdate,
-          enddate: job.data.enddate,
-        });
+        if (job.data?.userId) {
+          targets.push({
+            userId: job.data.userId,
+            startdate: job.data.startdate,
+            enddate: job.data.enddate,
+          });
+        }
       }
+      // No user-specific enqueue → cron fallback iterating every connection.
+      if (targets.length === 0) {
+        const connections = await prisma.withingsConnection.findMany({
+          select: { userId: true },
+        });
+        targets.push(...connections);
+      }
+      if (targets.length === 0) return;
+
+      let usersSynced = 0;
+      let recordingsImported = 0;
+      for (const { userId, startdate, enddate } of targets) {
+        try {
+          const options =
+            startdate === undefined && enddate === undefined
+              ? {}
+              : { startdate, enddate };
+          recordingsImported += await syncUserEcg(userId, options);
+          usersSynced++;
+        } catch (err) {
+          evt.addWarning(`Withings ECG sync failed for user ${userId}: ${err}`);
+        }
+      }
+
       evt.setBackground({
         task_name: "job.withings_ecg_sync",
         result: {
-          events_processed: jobs.length,
+          users_synced: usersSynced,
+          total: targets.length,
           recordings_imported: recordingsImported,
         },
       });

src/lib/withings/__tests__/sync-activity.test.ts

@@ -373,6 +373,31 @@ describe("syncUserActivity — field mapping + idempotency", () => {
     expect(recordSyncSuccess).toHaveBeenCalledWith("user-1", "withings");
   });
 
+  it("W-2 — records a transient failure and refuses the success stamp when the batch write throws", async () => {
+    // A swallowed batch-write used to fall through to recordSyncSuccess: green
+    // pill over lost data, and a PERSISTENT write bug green forever. The write
+    // failure now records a transient failure and the success stamp is skipped.
+    installFetchMock([
+      { date: "2026-05-12", steps: 8420, distance: 6720, calories: 412 },
+    ]);
+    vi.mocked(prisma.measurement.findMany).mockResolvedValue([] as never);
+    vi.mocked(prisma.measurement.createMany).mockRejectedValue(
+      new Error("db write blew up"),
+    );
+
+    // No rethrow — sibling chunks + the rollup fold still run (matches Google).
+    await expect(syncUserActivity("user-1")).resolves.toBeGreaterThanOrEqual(0);
+
+    expect(recordSyncSuccess).not.toHaveBeenCalled();
+    expect(recordSyncFailure).toHaveBeenCalledWith(
+      expect.objectContaining({
+        integration: "withings",
+        kind: "transient",
+        message: expect.stringContaining("activity batch write failed"),
+      }),
+    );
+  });
+
   it("tags every row with the canonical externalId so future replays dedup", async () => {
     installFetchMock([{ date: "2026-05-12", steps: 8420 }]);
     vi.mocked(prisma.measurement.findMany).mockResolvedValue([] as never);

src/lib/withings/__tests__/sync-sleep.test.ts

@@ -473,6 +473,32 @@ describe("syncUserSleep — segment writes + idempotency", () => {
     expect(recordSyncSuccess).toHaveBeenCalledWith("user-1", "withings");
   });
 
+  it("W-3 — records a transient failure and rethrows when a sleep write throws", async () => {
+    // A reconcile throw out of the stage transaction used to propagate past the
+    // success stamp and get swallowed by both callers: neither success NOR
+    // failure recorded, `lastAttemptAt` quietly stale. Now the source records a
+    // transient failure and rethrows (callers keep their warn-and-continue).
+    installFetchMock([
+      { startdate: 1715000000, enddate: 1715003600, state: 2, id: 1 },
+    ]);
+    reconcileMock.mockReset().mockResolvedValueOnce({
+      status: "failed",
+      reason: "db_error",
+      error: new Error("reconcile boom"),
+    });
+
+    await expect(syncUserSleep("user-1")).rejects.toThrow();
+
+    expect(recordSyncFailure).toHaveBeenCalledWith(
+      expect.objectContaining({
+        integration: "withings",
+        kind: "transient",
+        message: expect.stringContaining("sleep write failed"),
+      }),
+    );
+    expect(recordSyncSuccess).not.toHaveBeenCalled();
+  });
+
   it("upserts nightly vitals with stable per-vital externalIds (v1.18.10 P0)", async () => {
     const nightEnd = Math.floor(Date.UTC(2026, 4, 13, 6) / 1000);
     installSegmentThenSummaryFetch(

src/lib/withings/sync-activity.ts

@@ -311,6 +311,11 @@ export async function syncUserActivity(
   }
 
   let imported = 0;
+  // W-2 — a swallowed batch-write must not read as a clean sync. The catch
+  // below sets this so the tail stamps success ONLY when the write landed; a
+  // persistent write bug then accrues strikes instead of being reset by its
+  // own success stamp every tick.
+  let writeFailed = false;
   // v1.4.39.1 — track every (type, measuredAt) we touched so the
   // persistent rollup tier can be re-folded at the end of the sync.
   // See sync.ts header for the full rationale; the chart's
@@ -446,9 +451,24 @@ export async function syncUserActivity(
         );
       }
     } catch (err) {
+      // Data is lost for this tick and the pill would otherwise read green. Keep
+      // the no-rethrow (sibling day-chunks + the rollup fold still run) but
+      // record a transient failure and refuse the success stamp below.
+      // `transient` because the next hourly tick re-walks the fixed 30-day
+      // window, so a transient DB error self-heals while a persistent one now
+      // surfaces on the status card instead of resetting itself.
+      writeFailed = true;
       getEvent()?.addWarning(
         `Failed to batch-upsert ${planned.length} activity rows for ${userId}: ${err}`,
       );
+      await recordSyncFailure({
+        userId,
+        integration: "withings",
+        kind: "transient",
+        message: `activity batch write failed: ${
+          err instanceof Error ? err.message : String(err)
+        }`,
+      });
     }
   }
 
@@ -477,6 +497,10 @@ export async function syncUserActivity(
     );
   }
 
-  await recordSyncSuccess(userId, "withings");
+  // W-2 — only a clean write stamps success. A swallowed batch-write left the
+  // ledger honest via the transient failure recorded above.
+  if (!writeFailed) {
+    await recordSyncSuccess(userId, "withings");
+  }
   return imported;
 }

src/lib/withings/sync-sleep.ts

@@ -412,51 +412,74 @@ export async function syncUserSleep(
   // and are excluded (their `no-id` prefix would span every id-less night in
   // history, not just this fetch).
   const freshBySession = new Map<number, string[]>();
-  await prisma.$transaction(
-    async (tx) => {
-      for (const segment of segments) {
-        const stage = mapWithingsSleepState(segment.state);
-        if (!stage) continue;
 
-        const measuredAt = new Date(segment.enddate * 1000);
-        const durationSec = Math.max(0, segment.enddate - segment.startdate);
-        const minutes = Math.round(durationSec / 60);
-        const externalId = `withings:sleep:${userId}:${segment.id ?? "no-id"}:${segment.startdate}`;
-        if (typeof segment.id === "number") {
-          const fresh = freshBySession.get(segment.id) ?? [];
-          fresh.push(externalId);
-          freshBySession.set(segment.id, fresh);
-        }
+  // W-3 — a reconcile throw out of either sleep transaction propagated past the
+  // success stamp and both callers (cron + webhook) swallowed it: neither
+  // success NOR failure was recorded, so `lastAttemptAt` sat quietly stale
+  // during a persistent write failure. Record a transient failure at the source
+  // AND rethrow — preserving the callers' existing warn-and-continue cohort
+  // semantics — so the write failure surfaces on the status card without
+  // touching either caller. The fetch-side `WithingsApiError` path already
+  // records + rethrows above, so it never reaches these blocks.
+  const recordSleepWriteFailure = async (err: unknown): Promise<never> => {
+    await recordSyncFailure({
+      userId,
+      integration: "withings",
+      kind: "transient",
+      message: `sleep write failed: ${
+        err instanceof Error ? err.message : String(err)
+      }`,
+    });
+    throw err;
+  };
 
-        const verdict = await reconcileExternalMeasurement(
-          tx,
-          {
-            userId,
-            type: SLEEP_TYPE,
-            value: minutes,
-            unit: getUnitForType(SLEEP_TYPE),
-            measuredAt,
-            source: "WITHINGS",
-            sleepStage: stage,
-            externalId,
-          },
-          { exactExternalMatch: "update" },
-        );
-        if (verdict.status === "failed") {
-          throw new MeasurementReconciliationError(verdict);
+  await prisma
+    .$transaction(
+      async (tx) => {
+        for (const segment of segments) {
+          const stage = mapWithingsSleepState(segment.state);
+          if (!stage) continue;
+
+          const measuredAt = new Date(segment.enddate * 1000);
+          const durationSec = Math.max(0, segment.enddate - segment.startdate);
+          const minutes = Math.round(durationSec / 60);
+          const externalId = `withings:sleep:${userId}:${segment.id ?? "no-id"}:${segment.startdate}`;
+          if (typeof segment.id === "number") {
+            const fresh = freshBySession.get(segment.id) ?? [];
+            fresh.push(externalId);
+            freshBySession.set(segment.id, fresh);
+          }
+
+          const verdict = await reconcileExternalMeasurement(
+            tx,
+            {
+              userId,
+              type: SLEEP_TYPE,
+              value: minutes,
+              unit: getUnitForType(SLEEP_TYPE),
+              measuredAt,
+              source: "WITHINGS",
+              sleepStage: stage,
+              externalId,
+            },
+            { exactExternalMatch: "update" },
+          );
+          if (verdict.status === "failed") {
+            throw new MeasurementReconciliationError(verdict);
+          }
+          for (const dirty of verdict.dirtyIdentities ?? []) {
+            touched.push(dirty);
+          }
+          if (verdict.status === "inserted") {
+            insertedSleepMeasuredAts.push(measuredAt);
+          }
+          touched.push({ type: SLEEP_TYPE, measuredAt });
+          imported++;
         }
-        for (const dirty of verdict.dirtyIdentities ?? []) {
-          touched.push(dirty);
-        }
-        if (verdict.status === "inserted") {
-          insertedSleepMeasuredAts.push(measuredAt);
-        }
-        touched.push({ type: SLEEP_TYPE, measuredAt });
-        imported++;
-      }
-    },
-    { timeout: 60_000 },
-  );
+      },
+      { timeout: 60_000 },
+    )
+    .catch(recordSleepWriteFailure);
 
   // Session-scoped sweep (mirrors Google Health's replace-by-window): for each
   // session this fetch re-produced, soft-delete any live WITHINGS
@@ -489,38 +512,40 @@ export async function syncUserSleep(
     );
   }
 
-  await prisma.$transaction(
-    async (tx) => {
-      for (const summary of summaries) {
-        const measuredAt = new Date(summary.enddate * 1000);
-        const sessionId = summary.id ?? "no-id";
-        for (const vital of mapWithingsSleepSummary(summary.data)) {
-          const verdict = await reconcileExternalMeasurement(
-            tx,
-            {
-              userId,
-              type: vital.type,
-              value: vital.value,
-              unit: getUnitForType(vital.type),
-              measuredAt,
-              source: "WITHINGS",
-              externalId: `withings:sleep:${userId}:${sessionId}:${vital.fieldTag}`,
-            },
-            { exactExternalMatch: "update" },
-          );
-          if (verdict.status === "failed") {
-            throw new MeasurementReconciliationError(verdict);
+  await prisma
+    .$transaction(
+      async (tx) => {
+        for (const summary of summaries) {
+          const measuredAt = new Date(summary.enddate * 1000);
+          const sessionId = summary.id ?? "no-id";
+          for (const vital of mapWithingsSleepSummary(summary.data)) {
+            const verdict = await reconcileExternalMeasurement(
+              tx,
+              {
+                userId,
+                type: vital.type,
+                value: vital.value,
+                unit: getUnitForType(vital.type),
+                measuredAt,
+                source: "WITHINGS",
+                externalId: `withings:sleep:${userId}:${sessionId}:${vital.fieldTag}`,
+              },
+              { exactExternalMatch: "update" },
+            );
+            if (verdict.status === "failed") {
+              throw new MeasurementReconciliationError(verdict);
+            }
+            for (const dirty of verdict.dirtyIdentities ?? []) {
+              touched.push(dirty);
+            }
+            touched.push({ type: vital.type, measuredAt });
+            imported++;
           }
-          for (const dirty of verdict.dirtyIdentities ?? []) {
-            touched.push(dirty);
-          }
-          touched.push({ type: vital.type, measuredAt });
-          imported++;
         }
-      }
-    },
-    { timeout: 60_000 },
-  );
+      },
+      { timeout: 60_000 },
+    )
+    .catch(recordSleepWriteFailure);
 
   // v1.4.39.1 — refresh the persistent rollup table for every distinct
   // (type, day) the sync touched. Sleep segments collapse heavily —