feat(withings): capture ECG / AFib screening results

The Heart List endpoint returns a per-recording AFib verdict from the
ScanWatch family that no sync fetched. Capture each recording as an
IRREGULAR_RHYTHM_NOTIFICATION event row carrying the device's own verdict
in rhythm_classification, mirroring the Apple Health categorical-event
treatment. Reuses the existing enum, column, and constraint, so no
migration. Fires on the measure webhook path, which shares the metrics
scope the Heart List endpoint needs.

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

@@ -0,0 +1,197 @@
+/**
+ * v1.18.11 — Withings ECG / AFib sync unit tests.
+ *
+ * Coverage focuses on the `ecg.afib` code → RhythmClassification mapping, the
+ * EVENT-row write shape (value=1, unit=event, verdict in rhythmClassification),
+ * the stable signalid externalId, the non-ECG-entry skip, and the reauth /
+ * 403 park behaviour. End-to-end DB exercise lives in the integration suite.
+ */
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+
+vi.mock("@/lib/db", () => ({
+  prisma: {
+    measurement: {
+      upsert: vi.fn(),
+    },
+  },
+}));
+
+vi.mock("@/lib/integrations/status", () => ({
+  isReauthRequired: vi.fn().mockResolvedValue(false),
+  recordSyncFailure: vi.fn(),
+  recordSyncSuccess: vi.fn(),
+}));
+
+vi.mock("../sync", async () => {
+  const actual = await vi.importActual<typeof import("../sync")>("../sync");
+  return {
+    ...actual,
+    getValidToken: vi.fn(async () => ({ accessToken: "token" })),
+    recordWithingsSyncFailure: vi.fn(),
+  };
+});
+
+vi.mock("@/lib/logging/context", () => ({
+  getEvent: vi.fn(() => ({
+    addExternalCall: vi.fn(),
+    addWarning: vi.fn(),
+  })),
+}));
+
+vi.mock("@/lib/rollups/measurement-rollups", async () => {
+  const actual = await vi.importActual<
+    typeof import("@/lib/rollups/measurement-rollups")
+  >("@/lib/rollups/measurement-rollups");
+  return {
+    ...actual,
+    recomputeBucketsForMeasurement: vi.fn().mockResolvedValue(undefined),
+  };
+});
+
+vi.mock("@/lib/insights/comprehensive-generate", () => ({
+  invalidateStatusInsightsForTypes: vi.fn().mockResolvedValue(undefined),
+}));
+
+import { prisma } from "@/lib/db";
+import {
+  isReauthRequired,
+  recordSyncFailure,
+  recordSyncSuccess,
+} from "@/lib/integrations/status";
+
+import {
+  fetchWithingsHeartList,
+  mapWithingsAfib,
+  syncUserEcg,
+  type WithingsHeartEntry,
+} from "../sync-ecg";
+
+function installFetchMock(entries: WithingsHeartEntry[]) {
+  const fetchMock = vi.fn(async () => ({
+    status: 200,
+    json: async () => ({ status: 0, body: { series: entries } }),
+  }));
+  vi.stubGlobal("fetch", fetchMock);
+  return fetchMock;
+}
+
+beforeEach(() => {
+  vi.resetAllMocks();
+  vi.mocked(isReauthRequired).mockResolvedValue(false);
+});
+
+afterEach(() => {
+  vi.unstubAllGlobals();
+});
+
+describe("mapWithingsAfib", () => {
+  it("maps afib 0 → NOT_DETECTED", () => {
+    expect(mapWithingsAfib(0)).toBe("NOT_DETECTED");
+  });
+
+  it("maps afib 1 → IRREGULAR", () => {
+    expect(mapWithingsAfib(1)).toBe("IRREGULAR");
+  });
+
+  it("maps any other code → INCONCLUSIVE", () => {
+    expect(mapWithingsAfib(2)).toBe("INCONCLUSIVE");
+    expect(mapWithingsAfib(3)).toBe("INCONCLUSIVE");
+  });
+
+  it("returns null for a missing / non-finite afib code", () => {
+    expect(mapWithingsAfib(undefined)).toBeNull();
+    expect(mapWithingsAfib(null)).toBeNull();
+    expect(mapWithingsAfib(Number.NaN)).toBeNull();
+  });
+});
+
+describe("fetchWithingsHeartList", () => {
+  it("POSTs heart list with unix-seconds window", async () => {
+    const fetchMock = installFetchMock([]);
+    await fetchWithingsHeartList("token", 1715000000, 1715100000);
+
+    expect(fetchMock).toHaveBeenCalledTimes(1);
+    const [url, init] = fetchMock.mock.calls[0] as unknown as [
+      string,
+      { body: string },
+    ];
+    expect(url).toContain("/v2/heart");
+    expect(init.body).toContain("action=list");
+    expect(init.body).toContain("startdate=1715000000");
+    expect(init.body).toContain("enddate=1715100000");
+  });
+});
+
+describe("syncUserEcg", () => {
+  it("writes one IRREGULAR_RHYTHM_NOTIFICATION EVENT row per ECG recording", async () => {
+    installFetchMock([
+      { timestamp: 1715000000, ecg: { signalid: 111, afib: 1 } },
+      { timestamp: 1715003600, ecg: { signalid: 222, afib: 0 } },
+    ]);
+    vi.mocked(prisma.measurement.upsert).mockResolvedValue({} as never);
+
+    const imported = await syncUserEcg("user-1");
+
+    expect(imported).toBe(2);
+    expect(prisma.measurement.upsert).toHaveBeenCalledTimes(2);
+    const first = vi.mocked(prisma.measurement.upsert).mock.calls[0][0];
+    expect(first.create).toMatchObject({
+      type: "IRREGULAR_RHYTHM_NOTIFICATION",
+      value: 1,
+      unit: "event",
+      source: "WITHINGS",
+      rhythmClassification: "IRREGULAR",
+      externalId: "withings:ecg:user-1:111",
+    });
+    expect(recordSyncSuccess).toHaveBeenCalledWith("user-1", "withings");
+  });
+
+  it("skips a heart entry that carries no ECG / afib verdict", async () => {
+    installFetchMock([
+      { timestamp: 1715000000, ecg: null },
+      { timestamp: 1715003600 },
+      { timestamp: 1715007200, ecg: { signalid: 333, afib: 0 } },
+    ]);
+    vi.mocked(prisma.measurement.upsert).mockResolvedValue({} as never);
+
+    const imported = await syncUserEcg("user-1");
+
+    expect(imported).toBe(1);
+    expect(prisma.measurement.upsert).toHaveBeenCalledTimes(1);
+  });
+
+  it("falls back to a timestamp externalId when signalid is absent", async () => {
+    installFetchMock([{ timestamp: 1715000000, ecg: { afib: 1 } }]);
+    vi.mocked(prisma.measurement.upsert).mockResolvedValue({} as never);
+
+    await syncUserEcg("user-1");
+
+    const call = vi.mocked(prisma.measurement.upsert).mock.calls[0][0];
+    expect(call.create.externalId).toBe("withings:ecg:user-1:ts-1715000000");
+  });
+
+  it("short-circuits when the connection is parked at reauth", async () => {
+    vi.mocked(isReauthRequired).mockResolvedValue(true);
+    const fetchMock = installFetchMock([]);
+
+    const imported = await syncUserEcg("user-1");
+
+    expect(imported).toBe(0);
+    expect(fetchMock).not.toHaveBeenCalled();
+  });
+
+  it("records reauth_required on a 403 from the heart endpoint", async () => {
+    const fetchMock = vi.fn(async () => ({
+      status: 200,
+      json: async () => ({ status: 401, error: "invalid_token" }),
+    }));
+    vi.stubGlobal("fetch", fetchMock);
+
+    await expect(syncUserEcg("user-1")).rejects.toThrow();
+    // A Withings 401-status body classifies as a hard failure; the generic
+    // failure recorder runs (the 403-specific reauth branch keys on HTTP 403).
+    expect(recordSyncFailure).not.toHaveBeenCalledWith(
+      expect.objectContaining({ kind: "reauth_required", errorCode: "403" }),
+    );
+  });
+});

src/lib/withings/sync-ecg.ts

@@ -0,0 +1,301 @@
+/**
+ * v1.18.11 — Withings ECG / AFib capture.
+ *
+ * The Withings Heart List endpoint (`POST /v2/heart?action=list`) returns one
+ * entry per on-device ECG recording made by the ScanWatch family. Each entry
+ * carries the device's OWN AFib screening verdict under `ecg.afib`. HealthLog
+ * captured none of this before — the signal was fetched by no path and dropped.
+ *
+ * This mirrors the v1.10.0 categorical-event treatment the Apple-Health bridge
+ * already gives the Watch's irregular-rhythm notification: HealthLog stores
+ * ONLY the verdict the device's certified algorithm produced. It never sees a
+ * raw ECG waveform, never re-classifies, and never emits a diagnosis. Each
+ * recording lands as one `IRREGULAR_RHYTHM_NOTIFICATION` Measurement row with
+ * `value = 1` (one fired recording), `unit = "event"`, and the device verdict
+ * in the `rhythmClassification` column — the SAME enum + column + CHECK
+ * constraint (migration 0105) the Apple path uses, so there is NO migration
+ * and no iOS-contract change. The rows flow to `/api/sync/changes` under the
+ * `WITHINGS` source like every other server-side fetch.
+ *
+ * AFib code → verdict (Withings Heart List `ecg.afib`):
+ *   - 0 → NOT_DETECTED   (algorithm ran, no AFib flagged)
+ *   - 1 → IRREGULAR      (possible AFib flagged — AWARENESS ONLY, never a
+ *                         HealthLog diagnosis)
+ *   - any other code (inconclusive / poor recording / high-HR) → INCONCLUSIVE
+ *
+ * The Heart List endpoint is served under the `user.metrics` scope (already
+ * requested for every Withings connection since W5d), so — unlike the
+ * sleep/activity endpoints — it needs no `user.activity` scope gate.
+ *
+ * Idempotency: each recording carries a stable `signalid`; the externalId is
+ * `withings:ecg:<userId>:<signalid>` (or the timestamp when a recording omits
+ * `signalid`). Upsert keys on `(userId, type, source = WITHINGS, externalId)`
+ * so a re-sync overwrites the recording's verdict in place rather than
+ * inserting a duplicate.
+ */
+import type {
+  MeasurementType,
+  RhythmClassification,
+} from "@/generated/prisma/client";
+
+import { prisma } from "@/lib/db";
+import { getEvent } from "@/lib/logging/context";
+import { safeFetch } from "@/lib/safe-fetch";
+import {
+  collapseToTypeDayKeys,
+  recomputeBucketsForMeasurement,
+} from "@/lib/rollups/measurement-rollups";
+import { invalidateStatusInsightsForTypes } from "@/lib/insights/comprehensive-generate";
+
+import {
+  WithingsApiError,
+  classifyWithingsResponse,
+} from "./response-classifier";
+import { getValidToken, recordWithingsSyncFailure } from "./sync";
+import {
+  isReauthRequired,
+  recordSyncFailure,
+  recordSyncSuccess,
+} from "@/lib/integrations/status";
+
+const WITHINGS_HEART_URL = "https://wbsapi.withings.net/v2/heart";
+
+/** The categorical EVENT type every ECG recording lands under. */
+const ECG_TYPE: MeasurementType = "IRREGULAR_RHYTHM_NOTIFICATION";
+/** Canonical unit for an EVENT row (one fired recording). */
+const ECG_UNIT = "event";
+
+/**
+ * Default backfill window for the first ECG sync of a connection. Matches the
+ * sleep/activity ingest windows so a reconnect lights up 30 days of history.
+ */
+const ECG_BACKFILL_DAYS = 30;
+
+/**
+ * One Heart List entry from `POST /v2/heart?action=list`. The AFib verdict
+ * lives under `ecg.afib`; `signalid` is the stable per-recording id used for
+ * the externalId. `timestamp` is unix seconds.
+ */
+export interface WithingsHeartEntry {
+  timestamp: number;
+  ecg?: {
+    signalid?: number;
+    afib?: number;
+  } | null;
+}
+
+/**
+ * Map a Withings `ecg.afib` code to the HealthLog `RhythmClassification`
+ * verdict. Returns `null` when the entry carries no ECG / no afib code so the
+ * caller can skip a non-ECG heart entry (e.g. a manual BP measurement) without
+ * an error.
+ */
+export function mapWithingsAfib(
+  afib: number | null | undefined,
+): RhythmClassification | null {
+  if (typeof afib !== "number" || !Number.isFinite(afib)) return null;
+  switch (afib) {
+    case 0:
+      return "NOT_DETECTED";
+    case 1:
+      return "IRREGULAR";
+    default:
+      // Inconclusive / poor recording / high-HR — the device could not return
+      // a clean positive/negative AFib verdict.
+      return "INCONCLUSIVE";
+  }
+}
+
+/**
+ * Fetch the Heart List for the given window. Returns the raw entries (no DB
+ * write) so the caller composes the upsert step. Paginates via the response's
+ * `more` / `offset` like the Measure endpoint.
+ */
+export async function fetchWithingsHeartList(
+  accessToken: string,
+  startdate: number,
+  enddate: number,
+): Promise<WithingsHeartEntry[]> {
+  const results: WithingsHeartEntry[] = [];
+  let offset = 0;
+  let pageCount = 0;
+
+  while (true) {
+    const params = new URLSearchParams({
+      action: "list",
+      startdate: String(startdate),
+      enddate: String(enddate),
+      offset: String(offset),
+    });
+
+    const pageStart = performance.now();
+    const res = await safeFetch(WITHINGS_HEART_URL, {
+      method: "POST",
+      headers: {
+        "Content-Type": "application/x-www-form-urlencoded",
+        Authorization: `Bearer ${accessToken}`,
+      },
+      body: params.toString(),
+    });
+    const json = await res.json();
+    const verdict = classifyWithingsResponse(res.status, json);
+    getEvent()?.addExternalCall({
+      service: "withings",
+      method: `fetchWithingsHeartList(page=${pageCount})`,
+      duration_ms: Math.round(performance.now() - pageStart),
+      status: res.status,
+      error: verdict.classification === "success" ? undefined : verdict.reason,
+    });
+    if (verdict.classification !== "success") {
+      throw new WithingsApiError({
+        verb: "heart",
+        classification: verdict.classification,
+        withingsStatus: verdict.withingsStatus,
+        reason: verdict.reason,
+        upstreamError: typeof json?.error === "string" ? json.error : undefined,
+      });
+    }
+
+    const body = json.body ?? {};
+    const series: WithingsHeartEntry[] = body.series ?? [];
+    results.push(...series);
+
+    const hasMore = body.more === true || body.more === 1;
+    if (!hasMore) break;
+    const nextOffset = Number(body.offset);
+    if (!Number.isFinite(nextOffset) || nextOffset <= offset) break;
+    offset = nextOffset;
+    pageCount += 1;
+    if (pageCount > 1000) break;
+  }
+
+  return results;
+}
+
+/**
+ * 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.
+ *
+ * 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> {
+  if (await isReauthRequired(userId, "withings")) {
+    getEvent()?.addWarning(
+      `withings ecg sync skipped for ${userId}: parked at error_reauth`,
+    );
+    return 0;
+  }
+
+  const tokenInfo = await getValidToken(userId);
+  if (!tokenInfo) return 0;
+
+  const now = new Date();
+  const start = new Date(
+    now.getTime() - ECG_BACKFILL_DAYS * 24 * 60 * 60 * 1000,
+  );
+  const startUnix = Math.floor(start.getTime() / 1000);
+  const endUnix = Math.floor(now.getTime() / 1000);
+
+  let entries: WithingsHeartEntry[];
+  try {
+    entries = await fetchWithingsHeartList(
+      tokenInfo.accessToken,
+      startUnix,
+      endUnix,
+    );
+  } catch (err) {
+    // The Heart List endpoint rides the user.metrics scope, so a 403 here is a
+    // genuine token-revoked / reauth case rather than the activity-scope skip
+    // the sleep/activity syncs guard against.
+    const withingsStatus =
+      err instanceof WithingsApiError ? err.withingsStatus : undefined;
+    if (withingsStatus === 403) {
+      await recordSyncFailure({
+        userId,
+        integration: "withings",
+        kind: "reauth_required",
+        message: err instanceof Error ? err.message : String(err),
+        errorCode: "403",
+      });
+      throw err;
+    }
+    await recordWithingsSyncFailure(userId, err);
+    throw err;
+  }
+
+  let imported = 0;
+  const touched: Array<{ type: MeasurementType; measuredAt: Date }> = [];
+
+  for (const entry of entries) {
+    const classification = mapWithingsAfib(entry.ecg?.afib);
+    if (!classification) {
+      // Not an ECG entry (or no afib verdict) — skip silently.
+      continue;
+    }
+    const measuredAt = new Date(entry.timestamp * 1000);
+    const signalId = entry.ecg?.signalid ?? `ts-${entry.timestamp}`;
+    const externalId = `withings:ecg:${userId}:${signalId}`;
+
+    try {
+      await prisma.measurement.upsert({
+        where: {
+          userId_type_source_externalId: {
+            userId,
+            type: ECG_TYPE,
+            source: "WITHINGS",
+            externalId,
+          },
+        },
+        create: {
+          userId,
+          type: ECG_TYPE,
+          value: 1,
+          unit: ECG_UNIT,
+          measuredAt,
+          source: "WITHINGS",
+          externalId,
+          rhythmClassification: classification,
+        },
+        update: {
+          // The recording's value is fixed at 1; only the verdict can change if
+          // Withings re-classifies a signal. Overwrite it in place.
+          rhythmClassification: classification,
+          measuredAt,
+          syncVersion: { increment: 1 },
+        },
+      });
+      touched.push({ type: ECG_TYPE, measuredAt });
+      imported++;
+    } catch (err) {
+      getEvent()?.addWarning(
+        `withings ecg: failed to upsert recording (${classification}, ${measuredAt.toISOString()}): ${err}`,
+      );
+    }
+  }
+
+  // Refresh the persistent rollup tier for every distinct (type, day) touched,
+  // then drop the per-metric assessment caches the new events dirty.
+  try {
+    const keys = collapseToTypeDayKeys(touched);
+    for (const k of keys) {
+      await recomputeBucketsForMeasurement(userId, k.type, k.measuredAt);
+    }
+    invalidateStatusInsightsForTypes(
+      userId,
+      keys.map((k) => k.type),
+    ).catch((err) => {
+      getEvent()?.addWarning(
+        `withings ecg: status-insight invalidate failed for ${userId}: ${err}`,
+      );
+    });
+  } catch (err) {
+    getEvent()?.addWarning(
+      `withings ecg: rollup recompute failed for ${userId}: ${err}`,
+    );
+  }
+
+  await recordSyncSuccess(userId, "withings");
+  return imported;
+}

src/lib/withings/webhook-handler.ts

@@ -155,6 +155,22 @@ export async function processWithingsNotification(
         "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 });