From 5f4017d0391d249e54dc9a34e9ea75f1ac3574fb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Mon, 20 Jul 2026 11:37:23 +0200 Subject: [PATCH] fix(withings): queue ECG webhooks durably --- .../webhook/[token]/__tests__/route.test.ts | 4 + .../withings/webhook/__tests__/route.test.ts | 81 +++++++++++- .../__tests__/withings-ecg-worker.test.ts | 124 ++++++++++++++++++ .../reminder/register-integration-sync.ts | 14 ++ src/lib/jobs/reminder/withings-sync.ts | 35 +++++ src/lib/jobs/withings-ecg-queue.ts | 66 ++++++++++ src/lib/withings/__tests__/sync-ecg.test.ts | 88 +++++++++++++ src/lib/withings/sync-ecg.ts | 49 +++++-- src/lib/withings/webhook-handler.ts | 45 ++++--- 9 files changed, 477 insertions(+), 29 deletions(-) create mode 100644 src/lib/jobs/__tests__/withings-ecg-worker.test.ts create mode 100644 src/lib/jobs/withings-ecg-queue.ts diff --git a/src/app/api/withings/webhook/[token]/__tests__/route.test.ts b/src/app/api/withings/webhook/[token]/__tests__/route.test.ts index 480f34946..96e2402a8 100644 --- a/src/app/api/withings/webhook/[token]/__tests__/route.test.ts +++ b/src/app/api/withings/webhook/[token]/__tests__/route.test.ts @@ -48,6 +48,7 @@ import { prisma } from "@/lib/db"; import { syncUserMeasurements } from "@/lib/withings/sync"; import { checkRateLimit } from "@/lib/rate-limit"; import { getEvent } from "@/lib/logging/context"; +import { getGlobalBoss } from "@/lib/jobs/boss-instance"; const ORIGINAL_SECRET = process.env.WITHINGS_WEBHOOK_SECRET; @@ -64,6 +65,9 @@ beforeEach(() => { addWarning: vi.fn(), } as never); vi.mocked(syncUserMeasurements).mockResolvedValue(undefined as never); + vi.mocked(getGlobalBoss).mockReturnValue({ + send: vi.fn().mockResolvedValue("ecg-job"), + } as never); }); afterEach(() => { diff --git a/src/app/api/withings/webhook/__tests__/route.test.ts b/src/app/api/withings/webhook/__tests__/route.test.ts index f3372c968..9c9efb37b 100644 --- a/src/app/api/withings/webhook/__tests__/route.test.ts +++ b/src/app/api/withings/webhook/__tests__/route.test.ts @@ -69,9 +69,9 @@ beforeEach(() => { addWarning: vi.fn(), } as never); vi.mocked(syncUserMeasurements).mockResolvedValue(undefined as never); - // Default: pg-boss not available → activity / sleep paths fall back - // to inline sync. Individual tests override with a fake boss. - vi.mocked(getGlobalBoss).mockReturnValue(null); + vi.mocked(getGlobalBoss).mockReturnValue({ + send: vi.fn().mockResolvedValue("ecg-job"), + } as never); vi.mocked(syncUserActivity).mockResolvedValue(0 as never); vi.mocked(syncUserSleep).mockResolvedValue(0 as never); }); @@ -301,7 +301,7 @@ describe("POST /api/withings/webhook — appli dispatch (W17b/c)", () => { userId: "user-act-inline", withingsUserId: "wu-act-inline", } as never); - // Default beforeEach already sets `getGlobalBoss → null`. + vi.mocked(getGlobalBoss).mockReturnValue(null); const res = await POST( formRequest( @@ -319,6 +319,7 @@ describe("POST /api/withings/webhook — appli dispatch (W17b/c)", () => { userId: "user-sleep-inline", withingsUserId: "wu-sleep-inline", } as never); + vi.mocked(getGlobalBoss).mockReturnValue(null); const res = await POST( formRequest( @@ -364,6 +365,78 @@ describe("POST /api/withings/webhook — appli dispatch (W17b/c)", () => { expect(res.status).toBe(200); expect(syncUserMeasurements).toHaveBeenCalledWith("user-legacy"); }); + it("returns retryable non-2xx when the ECG event cannot be durably enqueued", async () => { + vi.mocked(prisma.withingsConnection.findFirst).mockResolvedValue({ + userId: "user-ecg", + withingsUserId: "wu-ecg", + } as never); + const send = vi.fn().mockRejectedValue(new Error("queue unavailable")); + vi.mocked(getGlobalBoss).mockReturnValue({ send } as never); + + const res = await POST( + formRequest( + { + userid: "wu-ecg", + appli: "1", + startdate: "1715000000", + enddate: "1715000060", + }, + { "x-withings-webhook-secret": "test-secret" }, + ), + ); + + expect(res.status).toBe(503); + expect(syncUserMeasurements).not.toHaveBeenCalled(); + }); + + it("uses one logical ECG job identity for a provider replay", async () => { + vi.mocked(prisma.withingsConnection.findFirst).mockResolvedValue({ + userId: "user-ecg", + withingsUserId: "wu-ecg", + } as never); + const send = vi + .fn() + .mockResolvedValueOnce("job-1") + .mockResolvedValueOnce(null); + vi.mocked(getGlobalBoss).mockReturnValue({ send } as never); + const delivery = { + userid: "wu-ecg", + appli: "1", + startdate: "1715000000", + enddate: "1715000060", + }; + + const first = await POST( + formRequest(delivery, { + "x-withings-webhook-secret": "test-secret", + }), + ); + const replay = await POST( + formRequest(delivery, { + "x-withings-webhook-secret": "test-secret", + }), + ); + + expect(first.status).toBe(200); + expect(replay.status).toBe(200); + expect(send).toHaveBeenCalledTimes(2); + expect(send.mock.calls[0][0]).toBe("withings-ecg-sync"); + expect(send.mock.calls[1][0]).toBe("withings-ecg-sync"); + expect(send.mock.calls[0][1]).toMatchObject({ + userId: "user-ecg", + startdate: 1715000000, + enddate: 1715000060, + }); + expect(send.mock.calls[0][2]).toMatchObject({ + singletonKey: expect.any(String), + retryLimit: 3, + retryDelay: 60, + retryBackoff: true, + }); + expect(send.mock.calls[0][2].singletonKey).toBe( + send.mock.calls[1][2].singletonKey, + ); + }); }); describe("HEAD /api/withings/webhook (Withings URL verification)", () => { diff --git a/src/lib/jobs/__tests__/withings-ecg-worker.test.ts b/src/lib/jobs/__tests__/withings-ecg-worker.test.ts new file mode 100644 index 000000000..c873d070d --- /dev/null +++ b/src/lib/jobs/__tests__/withings-ecg-worker.test.ts @@ -0,0 +1,124 @@ +import { beforeEach, describe, expect, it, vi } from "vitest"; +import type { Job } from "pg-boss"; +import { readFileSync } from "node:fs"; +import { join } from "node:path"; +const { syncUserEcg } = vi.hoisted(() => ({ + syncUserEcg: vi.fn< + ( + userId: string, + options?: { startdate?: number; enddate?: number }, + ) => Promise + >(), +})); + +vi.mock("@/lib/withings/sync-ecg", () => ({ + syncUserEcg: ( + userId: string, + options?: { startdate?: number; enddate?: number }, + ) => syncUserEcg(userId, options), +})); +vi.mock("@/lib/withings/sync", () => ({ + syncUserMeasurements: vi.fn(async () => 0), +})); +vi.mock("@/lib/withings/sync-activity", () => ({ + syncUserActivity: vi.fn(async () => 0), +})); +vi.mock("@/lib/withings/sync-sleep", () => ({ + syncUserSleep: vi.fn(async () => 0), +})); +vi.mock("@/lib/jobs/reminder-satisfy", () => ({ + enqueueReminderSatisfy: vi.fn(async () => undefined), +})); +vi.mock("@/lib/jobs/worker-status", () => ({ + recordError: vi.fn(), + recordWithingsSync: vi.fn(), +})); +vi.mock("@/lib/logging/background", () => ({ + withBackgroundEvent: async ( + _name: string, + fn: (event: { + setBackground: (value: unknown) => void; + setError: (error: unknown) => void; + addWarning: (warning: string) => void; + }) => Promise, + ) => + fn({ + setBackground: vi.fn(), + setError: vi.fn(), + addWarning: vi.fn(), + }), +})); +vi.mock("@/lib/logging/fire-and-forget", () => ({ + fireAndForget: vi.fn(), +})); +vi.mock("../reminder/shared", () => ({ + getWorkerPrisma: vi.fn(() => ({ + withingsConnection: { findMany: vi.fn(async () => []) }, + })), +})); + +import { + handleWithingsEcgSync, + type WithingsEcgSyncPayload, +} from "../reminder/withings-sync"; + +function job(data: WithingsEcgSyncPayload): Job { + return { data } as Job; +} + +beforeEach(() => { + vi.clearAllMocks(); +}); +const registrarSource = readFileSync( + join( + process.cwd(), + "src/lib/jobs/reminder/register-integration-sync.ts", + ), + "utf8", +); + +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); + const queued = job({ + userId: "user-1", + eventId: "wu-1:1:1715000000:1715000060", + triggeredAt: "2026-07-20T12:00:00.000Z", + startdate: 1715000000, + enddate: 1715000060, + }); + + await expect(handleWithingsEcgSync([queued])).rejects.toThrow( + "source write failed", + ); + await expect(handleWithingsEcgSync([queued])).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, + }); + }); +}); + +describe("withings ECG queue registration", () => { + it("deduplicates only queued replays and admits a rescue after work starts", () => { + const allQueues = registrarSource.match( + /const allQueues\s*=\s*\[([\s\S]*?)\];/, + ); + expect(allQueues).not.toBeNull(); + expect(allQueues![1]).toMatch(/\bWITHINGS_ECG_SYNC_QUEUE\b/); + expect(registrarSource).toMatch( + /\[WITHINGS_ECG_SYNC_QUEUE\]:\s*\{[\s\S]*?policy:\s*"short"/, + ); + expect(registrarSource).toMatch( + /boss\.work\([\s\S]*?WITHINGS_ECG_SYNC_QUEUE[\s\S]*?handleWithingsEcgSync/, + ); + }); +}); diff --git a/src/lib/jobs/reminder/register-integration-sync.ts b/src/lib/jobs/reminder/register-integration-sync.ts index 1d3256e17..988f4c040 100644 --- a/src/lib/jobs/reminder/register-integration-sync.ts +++ b/src/lib/jobs/reminder/register-integration-sync.ts @@ -69,13 +69,16 @@ import { type QueuePolicyTable, type ScheduleEntry, } from "./registrar-shared"; +import { WITHINGS_ECG_SYNC_QUEUE } from "@/lib/jobs/withings-ecg-queue"; import { WithingsSyncPayload, WithingsActivitySyncPayload, WithingsSleepSyncPayload, + WithingsEcgSyncPayload, handleWithingsFallbackSync, handleWithingsActivitySync, handleWithingsSleepSync, + handleWithingsEcgSync, } from "./withings-sync"; import { WhoopSyncPayload, @@ -242,6 +245,7 @@ const allQueues = [ WITHINGS_SYNC_QUEUE, WITHINGS_ACTIVITY_QUEUE, WITHINGS_SLEEP_QUEUE, + WITHINGS_ECG_SYNC_QUEUE, MOODLOG_SYNC_QUEUE, WITHINGS_OAUTH_STATE_CLEANUP_QUEUE, // v1.11.0 — WHOOP sync queues. Webhook-primary + cron-safety-net for @@ -349,6 +353,11 @@ const schedules: ScheduleEntry[] = [ * the work is still outstanding, a suppressed send can never lose work. */ const queuePolicies: QueuePolicyTable = { + [WITHINGS_ECG_SYNC_QUEUE]: { + policy: "short", + reason: + "Webhook retries share one identity while queued; once work starts, a later replay remains admissible as a rescue if the original exhausts retries.", + }, [WHOOP_BACKFILL_QUEUE]: { policy: "exclusive", reason: @@ -410,6 +419,11 @@ export async function registerIntegrationSyncQueues( { localConcurrency: 1 }, handleWithingsSleepSync, ); + await boss.work( + WITHINGS_ECG_SYNC_QUEUE, + { localConcurrency: 1 }, + handleWithingsEcgSync, + ); await boss.work( WITHINGS_OAUTH_STATE_CLEANUP_QUEUE, { localConcurrency: 1 }, diff --git a/src/lib/jobs/reminder/withings-sync.ts b/src/lib/jobs/reminder/withings-sync.ts index 24320b06d..f42305b0b 100644 --- a/src/lib/jobs/reminder/withings-sync.ts +++ b/src/lib/jobs/reminder/withings-sync.ts @@ -11,8 +11,12 @@ import { withBackgroundEvent } from "@/lib/logging/background"; import { syncUserMeasurements } from "@/lib/withings/sync"; import { syncUserActivity } from "@/lib/withings/sync-activity"; import { syncUserSleep } from "@/lib/withings/sync-sleep"; +import { syncUserEcg } from "@/lib/withings/sync-ecg"; +import type { WithingsEcgSyncPayload } from "@/lib/jobs/withings-ecg-queue"; import { enqueueReminderSatisfy } from "@/lib/jobs/reminder-satisfy"; import { getWorkerPrisma } from "./shared"; +export type { WithingsEcgSyncPayload } from "@/lib/jobs/withings-ecg-queue"; + export interface WithingsSyncPayload { triggeredAt: string; @@ -96,6 +100,37 @@ 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. + */ +export async function handleWithingsEcgSync( + jobs: Job[], +) { + await withBackgroundEvent("job.withings_ecg_sync", async (evt) => { + try { + let recordingsImported = 0; + for (const job of jobs) { + recordingsImported += await syncUserEcg(job.data.userId, { + startdate: job.data.startdate, + enddate: job.data.enddate, + }); + } + evt.setBackground({ + task_name: "job.withings_ecg_sync", + result: { + events_processed: jobs.length, + recordings_imported: recordingsImported, + }, + }); + } catch (err) { + evt.setError(err); + recordError(); + throw err; + } + }); +} + /** * v1.4.25 W17b — activity-sync handler. * diff --git a/src/lib/jobs/withings-ecg-queue.ts b/src/lib/jobs/withings-ecg-queue.ts new file mode 100644 index 000000000..9026d1b20 --- /dev/null +++ b/src/lib/jobs/withings-ecg-queue.ts @@ -0,0 +1,66 @@ +import { getGlobalBoss } from "@/lib/jobs/boss-instance"; +import { z } from "zod/v4"; + +const unixSecondsSchema = z.coerce.number().int().nonnegative().safe(); + +export const WITHINGS_ECG_SYNC_QUEUE = "withings-ecg-sync"; + +export interface WithingsEcgSyncPayload { + userId: string; + eventId: string; + triggeredAt: string; + startdate?: number; + enddate?: number; +} + +export interface WithingsEcgEventIdentity { + userId: string; + withingsUserId: string; + appli: number | null; + startdate: string | null; + enddate: string | null; +} + +export async function enqueueWithingsEcgSync( + identity: WithingsEcgEventIdentity, +): Promise<"enqueued" | "known"> { + const boss = getGlobalBoss(); + if (!boss) { + throw new Error("pg-boss producer is unavailable"); + } + + const eventId = [ + identity.withingsUserId, + identity.appli ?? "unknown", + identity.startdate ?? "unknown", + identity.enddate ?? "unknown", + ].join(":"); + const parsedStartdate = + identity.startdate === null + ? null + : unixSecondsSchema.safeParse(identity.startdate); + const parsedEnddate = + identity.enddate === null + ? null + : unixSecondsSchema.safeParse(identity.enddate); + const jobId = await boss.send( + WITHINGS_ECG_SYNC_QUEUE, + { + userId: identity.userId, + eventId, + triggeredAt: new Date().toISOString(), + ...(parsedStartdate?.success + ? { startdate: parsedStartdate.data } + : {}), + ...(parsedEnddate?.success ? { enddate: parsedEnddate.data } : {}), + } satisfies WithingsEcgSyncPayload, + { + singletonKey: `withings-ecg:${identity.userId}:${eventId}`, + retryLimit: 3, + retryDelay: 60, + retryBackoff: true, + }, + ); + + return jobId ? "enqueued" : "known"; +} diff --git a/src/lib/withings/__tests__/sync-ecg.test.ts b/src/lib/withings/__tests__/sync-ecg.test.ts index ba4d306af..92317bd21 100644 --- a/src/lib/withings/__tests__/sync-ecg.test.ts +++ b/src/lib/withings/__tests__/sync-ecg.test.ts @@ -16,6 +16,9 @@ vi.mock("@/lib/db", () => ({ ecgRecording: { upsert: vi.fn(), }, + withingsConnection: { + findUnique: vi.fn(), + }, }, })); @@ -68,6 +71,8 @@ import { recordSyncFailure, recordSyncSuccess, } from "@/lib/integrations/status"; +import { getValidToken, recordWithingsSyncFailure } from "../sync"; +import { recomputeBucketsForMeasurement } from "@/lib/rollups/measurement-rollups"; import { fetchWithingsHeartList, @@ -112,6 +117,12 @@ function installFetchMock( beforeEach(() => { vi.resetAllMocks(); vi.mocked(isReauthRequired).mockResolvedValue(false); + vi.mocked(getValidToken).mockResolvedValue({ + accessToken: "token", + } as never); + vi.mocked(prisma.withingsConnection.findUnique).mockResolvedValue({ + id: "connection-1", + } as never); vi.mocked(prisma.measurement.upsert).mockResolvedValue({ id: "m-1", } as never); @@ -120,6 +131,7 @@ beforeEach(() => { afterEach(() => { vi.unstubAllGlobals(); + vi.useRealTimers(); }); describe("mapWithingsAfib", () => { @@ -207,6 +219,30 @@ describe("syncUserEcg", () => { expect(recordSyncSuccess).toHaveBeenCalledWith("user-1", "withings"); }); + it("never narrows the broad catch-up to an ordinary webhook window", async () => { + vi.useFakeTimers(); + vi.setSystemTime(new Date("2026-07-20T12:00:00.000Z")); + const fetchMock = installFetchMock([]); + const recentStart = Math.floor( + new Date("2026-07-19T12:00:00.000Z").getTime() / 1000, + ); + const recentEnd = Math.floor( + new Date("2026-07-20T12:00:00.000Z").getTime() / 1000, + ); + + await syncUserEcg("user-1", { + startdate: recentStart, + enddate: recentEnd, + }); + + const init = fetchMock.mock.calls[0][1] as { body: string }; + const broadStart = Math.floor( + new Date("2026-06-20T12:00:00.000Z").getTime() / 1000, + ); + expect(init.body).toContain(`startdate=${broadStart}`); + expect(init.body).toContain(`enddate=${recentEnd}`); + }); + it("skips a heart entry that carries no ECG / afib verdict", async () => { installFetchMock([ { timestamp: 1715000000, ecg: null }, @@ -321,6 +357,58 @@ describe("syncUserEcg", () => { expect(keys[0]).toEqual(keys[1]); }); + it("retries token acquisition failures while the owned connection still exists", async () => { + installFetchMock([]); + vi.mocked(getValidToken).mockResolvedValue(null); + + await expect(syncUserEcg("user-1")).rejects.toThrow( + "Withings ECG token unavailable", + ); + expect(prisma.withingsConnection.findUnique).toHaveBeenCalledWith({ + where: { userId: "user-1" }, + select: { id: true }, + }); + expect(recordSyncSuccess).not.toHaveBeenCalled(); + }); + + it("does not mark the source complete when an ECG event write fails", async () => { + installFetchMock([ + { timestamp: 1715000000, ecg: { signalid: 111, afib: 1 } }, + ]); + vi.mocked(prisma.measurement.upsert).mockRejectedValue( + new Error("source write failed"), + ); + + await expect(syncUserEcg("user-1")).rejects.toThrow("source write failed"); + expect(recordSyncSuccess).not.toHaveBeenCalled(); + expect(recordWithingsSyncFailure).toHaveBeenCalledWith( + "user-1", + expect.objectContaining({ message: "source write failed" }), + ); + }); + + it("continues later ECG rows before retrying a failed source write", async () => { + installFetchMock([ + { timestamp: 1715000000, ecg: { signalid: 111, afib: 1 } }, + { timestamp: 1715003600, ecg: { signalid: 222, afib: 0 } }, + ]); + const poison = new Error("poison row"); + vi.mocked(prisma.measurement.upsert) + .mockRejectedValueOnce(poison) + .mockResolvedValueOnce({ id: "m-2" } as never); + + await expect(syncUserEcg("user-1")).rejects.toThrow("poison row"); + + expect(prisma.measurement.upsert).toHaveBeenCalledTimes(2); + expect(recomputeBucketsForMeasurement).toHaveBeenCalledWith( + "user-1", + "IRREGULAR_RHYTHM_NOTIFICATION", + new Date("2024-05-06T00:00:00.000Z"), + ); + expect(recordWithingsSyncFailure).toHaveBeenCalledWith("user-1", poison); + expect(recordSyncSuccess).not.toHaveBeenCalled(); + }); + it("still captures the AFib verdict when the waveform fetch fails", async () => { // List returns an entry, but the signal GET returns a hard-fail body. const fetchMock = vi.fn(async (_url: string, init: { body: string }) => { diff --git a/src/lib/withings/sync-ecg.ts b/src/lib/withings/sync-ecg.ts index 636d73f1c..61cfb92ec 100644 --- a/src/lib/withings/sync-ecg.ts +++ b/src/lib/withings/sync-ecg.ts @@ -340,14 +340,19 @@ async function captureEcgWaveform(params: { } /** - * Sync ECG / AFib recordings for a single user. Walks the trailing 30-day - * window, writing one `IRREGULAR_RHYTHM_NOTIFICATION` row per ECG recording - * tagged with the device's AFib verdict. Returns the number of upserted rows. + * Sync ECG / AFib recordings for a single user. Always walks at least the + * trailing 30 days and widens that catch-up to include an older webhook source + * window when supplied. Writes + * one `IRREGULAR_RHYTHM_NOTIFICATION` row per ECG recording tagged with the + * device's AFib verdict. Returns the number of upserted rows. * * Park behaviour mirrors the other Withings syncs — a connection at * `error_reauth` short-circuits before the upstream call. */ -export async function syncUserEcg(userId: string): Promise { +export async function syncUserEcg( + userId: string, + options: { startdate?: number; enddate?: number } = {}, +): Promise { if (await isReauthRequired(userId, "withings")) { getEvent()?.addWarning( `withings ecg sync skipped for ${userId}: parked at error_reauth`, @@ -356,14 +361,32 @@ export async function syncUserEcg(userId: string): Promise { } const tokenInfo = await getValidToken(userId); - if (!tokenInfo) return 0; + if (!tokenInfo) { + if (await isReauthRequired(userId, "withings")) return 0; + const connection = await prisma.withingsConnection.findUnique({ + where: { userId }, + select: { id: true }, + }); + if (!connection) return 0; + throw new Error("Withings ECG token unavailable"); + } const now = new Date(); - const start = new Date( - now.getTime() - ECG_BACKFILL_DAYS * 24 * 60 * 60 * 1000, + const defaultStartUnix = Math.floor( + (now.getTime() - ECG_BACKFILL_DAYS * 24 * 60 * 60 * 1000) / 1000, ); - const startUnix = Math.floor(start.getTime() / 1000); - const endUnix = Math.floor(now.getTime() / 1000); + const defaultEndUnix = Math.floor(now.getTime() / 1000); + const hasProviderWindow = + Number.isSafeInteger(options.startdate) && + Number.isSafeInteger(options.enddate) && + options.startdate! >= 0 && + options.startdate! <= options.enddate!; + const startUnix = hasProviderWindow + ? Math.min(defaultStartUnix, options.startdate!) + : defaultStartUnix; + const endUnix = hasProviderWindow + ? Math.max(defaultEndUnix, options.enddate!) + : defaultEndUnix; let entries: WithingsHeartEntry[]; try { @@ -395,6 +418,7 @@ export async function syncUserEcg(userId: string): Promise { let imported = 0; let waveformsCaptured = 0; const touched: Array<{ type: MeasurementType; measuredAt: Date }> = []; + let sourceWriteError: { cause: unknown } | null = null; for (const entry of entries) { const classification = mapWithingsAfib(entry.ecg?.afib); @@ -444,6 +468,7 @@ export async function syncUserEcg(userId: string): Promise { getEvent()?.addWarning( `withings ecg: failed to upsert recording (${classification}, ${measuredAt.toISOString()}): ${err}`, ); + sourceWriteError ??= { cause: err }; continue; } @@ -471,6 +496,7 @@ export async function syncUserEcg(userId: string): Promise { } } + annotate({ action: { name: "withings.ecg.sync" }, meta: { @@ -500,6 +526,11 @@ export async function syncUserEcg(userId: string): Promise { ); } + if (sourceWriteError) { + await recordWithingsSyncFailure(userId, sourceWriteError.cause); + throw sourceWriteError.cause; + } + await recordSyncSuccess(userId, "withings"); return imported; } diff --git a/src/lib/withings/webhook-handler.ts b/src/lib/withings/webhook-handler.ts index 31770e9d9..136f7c0a1 100644 --- a/src/lib/withings/webhook-handler.ts +++ b/src/lib/withings/webhook-handler.ts @@ -30,6 +30,7 @@ import { checkRateLimit, rateLimitHeaders } from "@/lib/rate-limit"; import { getClientIp, safeJson } from "@/lib/api-response"; import { annotate, getEvent } from "@/lib/logging/context"; import { getGlobalBoss } from "@/lib/jobs/boss-instance"; +import { enqueueWithingsEcgSync } from "@/lib/jobs/withings-ecg-queue"; /** * v1.4.25 W17b/c — appli categories that have their own sync routine @@ -94,6 +95,8 @@ export async function processWithingsNotification( const contentType = request.headers.get("content-type") ?? ""; let withingsUserId: string | null = null; let appli: number | null = null; + let startdate: string | null = null; + let enddate: string | null = null; if (contentType.includes("application/x-www-form-urlencoded")) { const formData = await request.formData(); @@ -103,10 +106,16 @@ export async function processWithingsNotification( const parsed = Number.parseInt(String(rawAppli), 10); if (Number.isFinite(parsed)) appli = parsed; } + const rawStartdate = formData.get("startdate"); + const rawEnddate = formData.get("enddate"); + startdate = rawStartdate == null ? null : String(rawStartdate); + enddate = rawEnddate == null ? null : String(rawEnddate); } else { const { data: body, error: jsonError } = await safeJson<{ userid?: string | number; appli?: string | number; + startdate?: string | number; + enddate?: string | number; }>(request); if (jsonError) return jsonError; withingsUserId = body.userid?.toString() ?? null; @@ -114,6 +123,8 @@ export async function processWithingsNotification( const parsed = Number.parseInt(String(body.appli), 10); if (Number.isFinite(parsed)) appli = parsed; } + startdate = body.startdate?.toString() ?? null; + enddate = body.enddate?.toString() ?? null; } if (!withingsUserId) { @@ -150,27 +161,29 @@ export async function processWithingsNotification( } else if (appli === WITHINGS_SLEEP_APPLI) { await enqueueWithingsSync(WITHINGS_SLEEP_QUEUE, connection.userId); } else { + try { + await enqueueWithingsEcgSync({ + userId: connection.userId, + withingsUserId, + appli, + startdate, + enddate, + }); + } catch (err) { + getEvent()?.addWarning( + `Failed to durably enqueue Withings ECG event for ${connection.userId}: ${err}`, + ); + return NextResponse.json( + { status: "unavailable" }, + { status: 503, headers: { "retry-after": "60" } }, + ); + } + syncUserMeasurements(connection.userId).catch((err) => { getEvent()?.addWarning( "Sync failed for user " + connection.userId + ": " + err, ); }); - // v1.18.11 — ECG / AFib capture rides the measure path. The Heart List - // endpoint shares the `user.metrics` scope with the measure family, so a - // measure-family notification (which is what a ScanWatch ECG recording - // delivers) is the right trigger. Fire it non-blocking and lazy-loaded so - // the webhook response stays fast and the cold path doesn't pull the ECG - // module in until a measure notification actually lands. - void (async () => { - try { - const { syncUserEcg } = await import("./sync-ecg"); - await syncUserEcg(connection.userId); - } catch (err) { - getEvent()?.addWarning( - "ECG sync failed for user " + connection.userId + ": " + err, - ); - } - })(); } return NextResponse.json({ status: "ok" }, { status: 200 });