From e47200505e26ec6c96f35c7f4f23d25de8331c59 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Fri, 24 Jul 2026 20:05:57 +0200 Subject: [PATCH] fix(withings): sync ECG on a schedule and surface write failures MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- .../__tests__/withings-ecg-worker.test.ts | 59 ++++-- .../jobs/__tests__/withings-queues.test.ts | 11 ++ .../reminder/register-integration-sync.ts | 11 ++ src/lib/jobs/reminder/withings-sync.ts | 63 ++++++- .../withings/__tests__/sync-activity.test.ts | 25 +++ src/lib/withings/__tests__/sync-sleep.test.ts | 26 +++ src/lib/withings/sync-activity.ts | 26 ++- src/lib/withings/sync-sleep.ts | 169 ++++++++++-------- 8 files changed, 293 insertions(+), 97 deletions(-) diff --git a/src/lib/jobs/__tests__/withings-ecg-worker.test.ts b/src/lib/jobs/__tests__/withings-ecg-worker.test.ts index 05de54a0a..cb11bde9e 100644 --- a/src/lib/jobs/__tests__/withings-ecg-worker.test.ts +++ b/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>>( + 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 { 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, - }); }); }); diff --git a/src/lib/jobs/__tests__/withings-queues.test.ts b/src/lib/jobs/__tests__/withings-queues.test.ts index c525e1a04..5dc7dac08 100644 --- a/src/lib/jobs/__tests__/withings-queues.test.ts +++ b/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\]/, + ); + }); }); diff --git a/src/lib/jobs/reminder/register-integration-sync.ts b/src/lib/jobs/reminder/register-integration-sync.ts index 6fe050510..eaaf55544 100644 --- a/src/lib/jobs/reminder/register-integration-sync.ts +++ b/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 diff --git a/src/lib/jobs/reminder/withings-sync.ts b/src/lib/jobs/reminder/withings-sync.ts index fbfd828a8..96be0b354 100644 --- a/src/lib/jobs/reminder/withings-sync.ts +++ b/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[], ) { 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, }, }); diff --git a/src/lib/withings/__tests__/sync-activity.test.ts b/src/lib/withings/__tests__/sync-activity.test.ts index c34b0d512..0f9c2a80d 100644 --- a/src/lib/withings/__tests__/sync-activity.test.ts +++ b/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); diff --git a/src/lib/withings/__tests__/sync-sleep.test.ts b/src/lib/withings/__tests__/sync-sleep.test.ts index 565455c02..98fcff22d 100644 --- a/src/lib/withings/__tests__/sync-sleep.test.ts +++ b/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( diff --git a/src/lib/withings/sync-activity.ts b/src/lib/withings/sync-activity.ts index ca21ad6ca..05431f7ef 100644 --- a/src/lib/withings/sync-activity.ts +++ b/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; } diff --git a/src/lib/withings/sync-sleep.ts b/src/lib/withings/sync-sleep.ts index f878d39a1..37ba27262 100644 --- a/src/lib/withings/sync-sleep.ts +++ b/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(); - 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 => { + 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 —