From 621b6734d0db440d4f72d95596ee82f7f0aaf4dc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Sun, 21 Jun 2026 00:54:16 +0200 Subject: [PATCH] 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 | 197 +++++++++++++ src/lib/withings/sync-ecg.ts | 301 ++++++++++++++++++++ src/lib/withings/webhook-handler.ts | 16 ++ 3 files changed, 514 insertions(+) create mode 100644 src/lib/withings/__tests__/sync-ecg.test.ts create mode 100644 src/lib/withings/sync-ecg.ts diff --git a/src/lib/withings/__tests__/sync-ecg.test.ts b/src/lib/withings/__tests__/sync-ecg.test.ts new file mode 100644 index 000000000..57d4ead02 --- /dev/null +++ b/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("../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" }), + ); + }); +}); diff --git a/src/lib/withings/sync-ecg.ts b/src/lib/withings/sync-ecg.ts new file mode 100644 index 000000000..71dee7d05 --- /dev/null +++ b/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::` (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 { + 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 { + 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; +} diff --git a/src/lib/withings/webhook-handler.ts b/src/lib/withings/webhook-handler.ts index 1bfc4ca77..31770e9d9 100644 --- a/src/lib/withings/webhook-handler.ts +++ b/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 });