fix(withings): queue ECG webhooks durably

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

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)", () => {

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<number>
+  >(),
+}));
+
+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<void>,
+  ) =>
+    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<WithingsEcgSyncPayload> {
+  return { data } as Job<WithingsEcgSyncPayload>;
+}
+
+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<WithingsEcgSyncPayload>\([\s\S]*?WITHINGS_ECG_SYNC_QUEUE[\s\S]*?handleWithingsEcgSync/,
+    );
+  });
+});

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<WithingsEcgSyncPayload>(
+    WITHINGS_ECG_SYNC_QUEUE,
+    { localConcurrency: 1 },
+    handleWithingsEcgSync,
+  );
   await boss.work<WithingsOAuthStateCleanupPayload>(
     WITHINGS_OAUTH_STATE_CLEANUP_QUEUE,
     { localConcurrency: 1 },

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<WithingsEcgSyncPayload>[],
+) {
+  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.
  *

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

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

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<number> {
+export async function syncUserEcg(
+  userId: string,
+  options: { startdate?: number; enddate?: number } = {},
+): Promise<number> {
   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<number> {
   }
 
   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<number> {
   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<number> {
       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<number> {
     }
   }
 
+
   annotate({
     action: { name: "withings.ecg.sync" },
     meta: {
@@ -500,6 +526,11 @@ export async function syncUserEcg(userId: string): Promise<number> {
     );
   }
 
+  if (sourceWriteError) {
+    await recordWithingsSyncFailure(userId, sourceWriteError.cause);
+    throw sourceWriteError.cause;
+  }
+
   await recordSyncSuccess(userId, "withings");
   return imported;
 }

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