From 462948b33b0d40dc7b6d457fe280c0e0611413c9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Sun, 21 Jun 2026 10:22:29 +0200 Subject: [PATCH] feat(withings): capture the full ECG waveform, not just the AFib verdict The Heart sync recorded only the device's atrial-fibrillation verdict per on-device ECG. Fetch each recording's full waveform via the Heart get endpoint and store the raw micro-volt samples so a real trace can later be rendered. The sample array is sensitive health data: it is held AES-256-GCM encrypted at rest in the new EcgRecording.waveformEncrypted Bytes column (fail-closed crypto), with the sampling frequency, sample count, duration, lead, average heart rate, and AFib verdict kept as plaintext descriptors. A re-sync upserts on (userId, source, externalRecordingId) so a recording is never duplicated; a waveform fetch failure never fails the verdict ingest. Rendering UI is a follow-up. --- .../0185_v1190_ecg_waveform/migration.sql | 70 ++++++ prisma/schema.prisma | 77 +++++++ .../__tests__/ecg-waveform-codec.test.ts | 60 +++++ src/lib/withings/__tests__/sync-ecg.test.ts | 171 +++++++++++++- src/lib/withings/ecg-waveform-codec.ts | 40 ++++ src/lib/withings/sync-ecg.ts | 210 +++++++++++++++++- 6 files changed, 617 insertions(+), 11 deletions(-) create mode 100644 prisma/migrations/0185_v1190_ecg_waveform/migration.sql create mode 100644 src/lib/withings/__tests__/ecg-waveform-codec.test.ts create mode 100644 src/lib/withings/ecg-waveform-codec.ts diff --git a/prisma/migrations/0185_v1190_ecg_waveform/migration.sql b/prisma/migrations/0185_v1190_ecg_waveform/migration.sql new file mode 100644 index 000000000..9ac812065 --- /dev/null +++ b/prisma/migrations/0185_v1190_ecg_waveform/migration.sql @@ -0,0 +1,70 @@ +-- v1.19.0 — full Withings ECG waveform capture. +-- +-- v1.18.11 captured only the device's atrial-fibrillation *verdict* per +-- on-device ECG (one IRREGULAR_RHYTHM_NOTIFICATION measurement EVENT row). +-- This adds the raw ECG **waveform** so a real trace can later be rendered. +-- +-- One additive table, `ecg_recordings`: +-- +-- * `waveform_encrypted` (BYTEA) holds the AES-256-GCM ciphertext of the +-- JSON-encoded micro-volt sample array (Withings Heart `get` `body.signal`) +-- — the `*Encrypted` Bytes convention (the coach_messages.encrypted_content +-- precedent). The waveform NEVER lands as plaintext (fail-closed crypto). +-- * `sampling_frequency` (Hz), `sample_count`, `duration_seconds`, `lead`, +-- and `average_heart_rate` stay plaintext descriptors so a renderer can +-- read them without a per-row decrypt. +-- * `rhythm_classification` snapshots the paired EVENT row's AFib verdict; +-- `measurement_id` FKs back to that row (ON DELETE SET NULL — the trace +-- survives a re-keyed/removed EVENT row). +-- +-- Idempotency: a re-sync of the same recording upserts in place via the +-- unique `(user_id, source, external_recording_id)` (Withings `signalid`, +-- or a `ts-` fallback). +-- +-- Additive; no existing row touched. Idempotent guards (`IF NOT EXISTS` / +-- `DO $$`) so reruns are safe on prod. +-- +-- Reversibility (down): +-- DROP TABLE IF EXISTS "ecg_recordings"; +-- The waveform store is self-contained — no other domain depends on it. + +-- CreateTable +CREATE TABLE IF NOT EXISTS "ecg_recordings" ( + "id" TEXT NOT NULL, + "user_id" TEXT NOT NULL, + "source" "measurement_source" NOT NULL DEFAULT 'WITHINGS', + "external_recording_id" TEXT NOT NULL, + "recorded_at" TIMESTAMP(3) NOT NULL, + "waveform_encrypted" BYTEA NOT NULL, + "sampling_frequency" INTEGER NOT NULL, + "sample_count" INTEGER NOT NULL, + "duration_seconds" DOUBLE PRECISION, + "lead" TEXT, + "average_heart_rate" INTEGER, + "rhythm_classification" "rhythm_classification", + "measurement_id" TEXT, + "created_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + "updated_at" TIMESTAMP(3) NOT NULL, + + CONSTRAINT "ecg_recordings_pkey" PRIMARY KEY ("id") +); + +-- CreateIndex +CREATE UNIQUE INDEX IF NOT EXISTS "ecg_recordings_user_id_source_external_recording_id_key" ON "ecg_recordings"("user_id", "source", "external_recording_id"); + +-- CreateIndex +CREATE INDEX IF NOT EXISTS "ecg_recordings_user_id_recorded_at_idx" ON "ecg_recordings"("user_id", "recorded_at" DESC); + +-- AddForeignKey +DO $$ BEGIN + ALTER TABLE "ecg_recordings" ADD CONSTRAINT "ecg_recordings_user_id_fkey" FOREIGN KEY ("user_id") REFERENCES "users"("id") ON DELETE CASCADE ON UPDATE CASCADE; +EXCEPTION + WHEN duplicate_object THEN null; +END $$; + +-- AddForeignKey +DO $$ BEGIN + ALTER TABLE "ecg_recordings" ADD CONSTRAINT "ecg_recordings_measurement_id_fkey" FOREIGN KEY ("measurement_id") REFERENCES "measurements"("id") ON DELETE SET NULL ON UPDATE CASCADE; +EXCEPTION + WHEN duplicate_object THEN null; +END $$; diff --git a/prisma/schema.prisma b/prisma/schema.prisma index b98b6f10d..f07426b66 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -657,6 +657,10 @@ model User { // per-day timeline mirroring the CycleDayLog shape. illnessEpisodes IllnessEpisode[] illnessDayLogs IllnessDayLog[] + // v1.19.0 — full Withings ECG waveform recordings (encrypted at rest). + // The paired AFib verdict still lands as an IRREGULAR_RHYTHM_NOTIFICATION + // Measurement EVENT row (v1.18.11); this holds the raw signal samples. + ecgRecordings EcgRecording[] // v1.7.0 profile — optional patient-identity fields surfaced on the // health-record export cover (PDF) and the FHIR `Patient` resource. @@ -1070,6 +1074,10 @@ model Measurement { updatedAt DateTime @updatedAt @map("updated_at") user User @relation(fields: [userId], references: [id], onDelete: Cascade) + // v1.19.0 — optional back-relation to the full ECG waveform recording(s) + // paired with this EVENT row. SET NULL on the FK side keeps the trace if + // the EVENT row is re-keyed / removed. + ecgRecordings EcgRecording[] // Original measuredAt-based dedup — kept for legacy Withings ingest // and the manual UI's "no duplicate at the same wall-clock" guard. @@ -1130,6 +1138,75 @@ model Measurement { @@map("measurements") } +// ─── ECG waveform recordings (v1.19.0) ───────────────────── +// +// v1.18.11 captured only the device's AFib *verdict* for each on-device +// ECG (one IRREGULAR_RHYTHM_NOTIFICATION Measurement EVENT row). This model +// stores the full raw ECG **waveform** so a real trace can later be rendered. +// +// The waveform is sensitive health data, so the sample array is held +// AES-256-GCM at rest in `waveformEncrypted` (Bytes) — the same Bytes-codec +// precedent as `CoachMessage.encryptedContent`. The plaintext is the JSON +// encoding of the integer micro-volt sample array returned by the Withings +// Heart `get` endpoint (`body.signal`); decrypt → `JSON.parse` → number[]. +// It NEVER lands as plaintext (fail-closed crypto in `src/lib/crypto.ts`). +// +// The non-sensitive descriptors stay plaintext so a future renderer can read +// `samplingFrequency` / `sampleCount` / `durationSeconds` without a per-row +// decrypt: `samplingFrequency` (Hz) pairs with the sample index to recover +// each sample's time offset; `sampleCount` is the array length; `durationSeconds` +// is `sampleCount / samplingFrequency`; `lead` is the recording lead/pathway +// when the source reports it; `averageHeartRate` mirrors the source's reported +// average BPM for the strip when present. +// +// `rhythmClassification` snapshots the same AFib verdict captured on the paired +// Measurement EVENT row so the waveform self-describes without a join, and +// `measurementId` links back to that row (SET NULL — the trace survives a +// re-keyed/removed EVENT row). +// +// Idempotency: `externalRecordingId` is the source's stable per-recording id +// (Withings `signalid`, or a `ts-` fallback when absent). The unique +// `(userId, source, externalRecordingId)` lets a re-sync overwrite the same +// recording in place rather than inserting a duplicate. +// +// RENDERING UI IS OUT OF SCOPE FOR v1.19.0 — this is ingest-only. A trace +// view (and any analytics over the waveform) is a follow-up. +model EcgRecording { + id String @id @default(cuid()) + userId String @map("user_id") + source MeasurementSource @default(WITHINGS) + /// Source's stable per-recording id (Withings `signalid`, or `ts-`). + externalRecordingId String @map("external_recording_id") + /// When the strip was recorded on-device (source timestamp). + recordedAt DateTime @map("recorded_at") + /// AES-256-GCM ciphertext of the JSON-encoded micro-volt sample array. + waveformEncrypted Bytes @map("waveform_encrypted") + /// Sampling rate of the signal in Hz (sample index → time offset). + samplingFrequency Int @map("sampling_frequency") + /// Number of samples in the (decrypted) waveform array. + sampleCount Int @map("sample_count") + /// `sampleCount / samplingFrequency` in seconds; NULL when frequency is 0. + durationSeconds Float? @map("duration_seconds") + /// Recording lead / pathway label when the source reports it (else NULL). + lead String? + /// Source-reported average heart rate (BPM) for the strip, when present. + averageHeartRate Int? @map("average_heart_rate") + /// Snapshot of the device AFib verdict for the paired EVENT row. + rhythmClassification RhythmClassification? @map("rhythm_classification") + /// FK to the IRREGULAR_RHYTHM_NOTIFICATION Measurement EVENT row, when known. + measurementId String? @map("measurement_id") + createdAt DateTime @default(now()) @map("created_at") + updatedAt DateTime @updatedAt @map("updated_at") + + user User @relation(fields: [userId], references: [id], onDelete: Cascade) + measurement Measurement? @relation(fields: [measurementId], references: [id], onDelete: SetNull) + + // Re-sync overwrites in place rather than inserting a duplicate. + @@unique([userId, source, externalRecordingId]) + @@index([userId, recordedAt(sort: Desc)]) + @@map("ecg_recordings") +} + // ─── Measurement rollups (v1.5.0) ────────────────────────── // // Persistent rollup table — second cache tier behind the 60-s LRU. diff --git a/src/lib/withings/__tests__/ecg-waveform-codec.test.ts b/src/lib/withings/__tests__/ecg-waveform-codec.test.ts new file mode 100644 index 000000000..29e4abbb6 --- /dev/null +++ b/src/lib/withings/__tests__/ecg-waveform-codec.test.ts @@ -0,0 +1,60 @@ +/** + * v1.19.0 — ECG waveform Bytes-codec round-trip tests. + * + * The raw micro-volt sample array is stored AES-256-GCM encrypted in the + * `ecg_recordings.waveform_encrypted` Bytes column. These tests assert the + * encrypt → Bytes → decrypt round-trip preserves the array exactly, that the + * stored bytes are NOT plaintext (no cleartext sample shows through), and that + * a tampered / non-array payload fails closed. + */ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +import { _resetCryptoCacheForTests, encrypt } from "@/lib/crypto"; +import { + decryptWaveformFromBytes, + encryptWaveformToBytes, +} from "../ecg-waveform-codec"; + +const KEY = "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef"; + +beforeEach(() => { + vi.unstubAllEnvs(); + vi.stubEnv("ENCRYPTION_KEYS", ""); + vi.stubEnv("ENCRYPTION_ACTIVE_KEY_ID", ""); + vi.stubEnv("ENCRYPTION_KEY", KEY); + _resetCryptoCacheForTests(); +}); + +afterEach(() => { + vi.unstubAllEnvs(); + _resetCryptoCacheForTests(); +}); + +describe("ecg-waveform-codec", () => { + it("round-trips a sample array through encrypt → Bytes → decrypt", () => { + const samples = [0, 12, -47, 1024, -2048, 999999, -1000000, 1]; + const bytes = encryptWaveformToBytes(samples); + expect(decryptWaveformFromBytes(bytes)).toEqual(samples); + }); + + it("round-trips an empty array", () => { + const bytes = encryptWaveformToBytes([]); + expect(decryptWaveformFromBytes(bytes)).toEqual([]); + }); + + it("stores the waveform encrypted, not as plaintext", () => { + const samples = [424242, -313131]; + const bytes = encryptWaveformToBytes(samples); + const asText = Buffer.from(bytes).toString("utf8"); + // The ciphertext is `.`; no raw sample value leaks. + expect(asText).not.toContain("424242"); + expect(asText).not.toContain("-313131"); + expect(asText.startsWith("v1.")).toBe(true); + }); + + it("fails closed on a payload that does not decode to a number array", () => { + // Hand-encrypt a non-array JSON value through the same active key. + const tampered = Buffer.from(encrypt(JSON.stringify({ not: "an array" }))); + expect(() => decryptWaveformFromBytes(tampered)).toThrow(/number array/); + }); +}); diff --git a/src/lib/withings/__tests__/sync-ecg.test.ts b/src/lib/withings/__tests__/sync-ecg.test.ts index 57d4ead02..ba4d306af 100644 --- a/src/lib/withings/__tests__/sync-ecg.test.ts +++ b/src/lib/withings/__tests__/sync-ecg.test.ts @@ -13,9 +13,18 @@ vi.mock("@/lib/db", () => ({ measurement: { upsert: vi.fn(), }, + ecgRecording: { + upsert: vi.fn(), + }, }, })); +vi.mock("../ecg-waveform-codec", () => ({ + encryptWaveformToBytes: vi.fn((samples: number[]) => + Uint8Array.from([samples.length & 0xff]), + ), +})); + vi.mock("@/lib/integrations/status", () => ({ isReauthRequired: vi.fn().mockResolvedValue(false), recordSyncFailure: vi.fn(), @@ -36,6 +45,7 @@ vi.mock("@/lib/logging/context", () => ({ addExternalCall: vi.fn(), addWarning: vi.fn(), })), + annotate: vi.fn(), })); vi.mock("@/lib/rollups/measurement-rollups", async () => { @@ -61,16 +71,40 @@ import { import { fetchWithingsHeartList, + fetchWithingsHeartSignal, 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 } }), - })); +/** A default signal `get` body returned for any `action=get` request. */ +const DEFAULT_SIGNAL = { + signal: [1, 2, 3, 4], + sampling_frequency: 300, + heart_rate: 72, +}; + +/** + * Install a fetch mock that routes by request body: `action=list` returns the + * heart-list series, `action=get` returns a signal body. The signal body can + * be overridden (or set to null-signal) per test. + */ +function installFetchMock( + entries: WithingsHeartEntry[], + signal: Record | null = DEFAULT_SIGNAL, +) { + const fetchMock = vi.fn(async (_url: string, init: { body: string }) => { + if (init.body.includes("action=get")) { + return { + status: 200, + json: async () => ({ status: 0, body: signal ?? {} }), + }; + } + return { + status: 200, + json: async () => ({ status: 0, body: { series: entries } }), + }; + }); vi.stubGlobal("fetch", fetchMock); return fetchMock; } @@ -78,6 +112,10 @@ function installFetchMock(entries: WithingsHeartEntry[]) { beforeEach(() => { vi.resetAllMocks(); vi.mocked(isReauthRequired).mockResolvedValue(false); + vi.mocked(prisma.measurement.upsert).mockResolvedValue({ + id: "m-1", + } as never); + vi.mocked(prisma.ecgRecording.upsert).mockResolvedValue({} as never); }); afterEach(() => { @@ -122,13 +160,36 @@ describe("fetchWithingsHeartList", () => { }); }); +describe("fetchWithingsHeartSignal", () => { + it("POSTs action=get with the signalid and returns the signal body", async () => { + const fetchMock = installFetchMock([]); + const sig = await fetchWithingsHeartSignal("token", 999); + + const init = fetchMock.mock.calls[0][1] as { body: string }; + expect(init.body).toContain("action=get"); + expect(init.body).toContain("signalid=999"); + expect(sig).toEqual({ + signal: [1, 2, 3, 4], + sampling_frequency: 300, + heart_rate: 72, + }); + }); + + it("returns null when the response carries no usable signal array", async () => { + installFetchMock([], { signal: [], sampling_frequency: 300 }); + expect(await fetchWithingsHeartSignal("token", 999)).toBeNull(); + }); +}); + 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); + vi.mocked(prisma.measurement.upsert).mockResolvedValue({ + id: "m-1", + } as never); const imported = await syncUserEcg("user-1"); @@ -152,7 +213,9 @@ describe("syncUserEcg", () => { { timestamp: 1715003600 }, { timestamp: 1715007200, ecg: { signalid: 333, afib: 0 } }, ]); - vi.mocked(prisma.measurement.upsert).mockResolvedValue({} as never); + vi.mocked(prisma.measurement.upsert).mockResolvedValue({ + id: "m-1", + } as never); const imported = await syncUserEcg("user-1"); @@ -162,7 +225,9 @@ describe("syncUserEcg", () => { 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); + vi.mocked(prisma.measurement.upsert).mockResolvedValue({ + id: "m-1", + } as never); await syncUserEcg("user-1"); @@ -194,4 +259,94 @@ describe("syncUserEcg", () => { expect.objectContaining({ kind: "reauth_required", errorCode: "403" }), ); }); + + it("captures the full waveform for each recording with a numeric signalid", async () => { + installFetchMock([ + { timestamp: 1715000000, ecg: { signalid: 111, afib: 1 } }, + ]); + + await syncUserEcg("user-1"); + + // The AFib verdict EVENT row is still written (v1.18.11 behaviour). + expect(prisma.measurement.upsert).toHaveBeenCalledTimes(1); + // The waveform row is written, keyed for idempotent re-sync and carrying + // the descriptors + the link back to the EVENT row. + expect(prisma.ecgRecording.upsert).toHaveBeenCalledTimes(1); + const call = vi.mocked(prisma.ecgRecording.upsert).mock.calls[0][0]; + expect(call.where).toEqual({ + userId_source_externalRecordingId: { + userId: "user-1", + source: "WITHINGS", + externalRecordingId: "111", + }, + }); + expect(call.create).toMatchObject({ + userId: "user-1", + source: "WITHINGS", + externalRecordingId: "111", + samplingFrequency: 300, + sampleCount: 4, + durationSeconds: 4 / 300, + averageHeartRate: 72, + rhythmClassification: "IRREGULAR", + measurementId: "m-1", + }); + expect(call.create.waveformEncrypted).toBeInstanceOf(Uint8Array); + }); + + it("does not fetch a waveform for a recording without a numeric signalid", async () => { + installFetchMock([{ timestamp: 1715000000, ecg: { afib: 1 } }]); + + await syncUserEcg("user-1"); + + // Verdict still captured, but no signal GET / no waveform row. + expect(prisma.measurement.upsert).toHaveBeenCalledTimes(1); + expect(prisma.ecgRecording.upsert).not.toHaveBeenCalled(); + }); + + it("upserts the waveform idempotently so a re-sync overwrites in place", async () => { + installFetchMock([ + { timestamp: 1715000000, ecg: { signalid: 111, afib: 1 } }, + ]); + + await syncUserEcg("user-1"); + await syncUserEcg("user-1"); + + // Two syncs → two upserts, both keyed on the same composite, never a plain + // create that would duplicate. + expect(prisma.ecgRecording.upsert).toHaveBeenCalledTimes(2); + const keys = vi + .mocked(prisma.ecgRecording.upsert) + .mock.calls.map((c) => c[0].where); + expect(keys[0]).toEqual(keys[1]); + }); + + 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 }) => { + if (init.body.includes("action=get")) { + return { status: 200, json: async () => ({ status: 503 }) }; + } + return { + status: 200, + json: async () => ({ + status: 0, + body: { + series: [ + { timestamp: 1715000000, ecg: { signalid: 111, afib: 1 } }, + ], + }, + }), + }; + }); + vi.stubGlobal("fetch", fetchMock); + + const imported = await syncUserEcg("user-1"); + + // The verdict EVENT row still landed; only the waveform was skipped. + expect(imported).toBe(1); + expect(prisma.measurement.upsert).toHaveBeenCalledTimes(1); + expect(prisma.ecgRecording.upsert).not.toHaveBeenCalled(); + expect(recordSyncSuccess).toHaveBeenCalledWith("user-1", "withings"); + }); }); diff --git a/src/lib/withings/ecg-waveform-codec.ts b/src/lib/withings/ecg-waveform-codec.ts new file mode 100644 index 000000000..0e42b678b --- /dev/null +++ b/src/lib/withings/ecg-waveform-codec.ts @@ -0,0 +1,40 @@ +/** + * AES-256-GCM ↔ `Bytes` codec for the ECG waveform column (v1.19.0). + * + * `ecg_recordings.waveform_encrypted` stores the JSON-encoded micro-volt + * sample array (`number[]`) as the `encrypt()` string format from + * `@/lib/crypto` (`"."`) encoded as UTF-8 bytes — the same + * Bytes-codec shape the Coach's encrypted columns use. The waveform is raw + * health data and is NEVER persisted as plaintext; crypto is fail-closed + * (a missing / malformed key throws rather than writing plaintext). + * + * Prisma's `Bytes` maps to `Uint8Array`, so writes allocate a + * fresh ArrayBuffer-backed `Uint8Array` to keep the structural type stable. + */ +import { Buffer } from "node:buffer"; + +import { decrypt, encrypt } from "@/lib/crypto"; + +/** Encrypt a waveform sample array into the `Bytes` payload the schema stores. */ +export function encryptWaveformToBytes( + samples: number[], +): Uint8Array { + const ciphertext = encrypt(JSON.stringify(samples)); + const encoded = Buffer.from(ciphertext, "utf8"); + const out = new Uint8Array(new ArrayBuffer(encoded.byteLength)); + out.set(encoded); + return out; +} + +/** + * Decrypt a `Bytes` waveform payload back to the sample array. Throws on a + * bad key id (fail-closed) or a non-array payload. + */ +export function decryptWaveformFromBytes(buf: Uint8Array): number[] { + const text = Buffer.from(buf).toString("utf8"); + const parsed: unknown = JSON.parse(decrypt(text)); + if (!Array.isArray(parsed) || !parsed.every((n) => typeof n === "number")) { + throw new Error("ECG waveform payload did not decode to a number array"); + } + return parsed as number[]; +} diff --git a/src/lib/withings/sync-ecg.ts b/src/lib/withings/sync-ecg.ts index 71dee7d05..636d73f1c 100644 --- a/src/lib/withings/sync-ecg.ts +++ b/src/lib/withings/sync-ecg.ts @@ -1,11 +1,23 @@ /** * v1.18.11 — Withings ECG / AFib capture. + * v1.19.0 — full ECG waveform capture (the raw signal samples). * * 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. * + * v1.19.0 additionally fetches each recording's full waveform via the Heart + * `get` endpoint (`POST /v2/heart?action=get&signalid=`), which returns the + * micro-volt sample array (`body.signal`) plus `body.sampling_frequency` + * (Hz) and an optional average `body.heart_rate`. The sample array is raw + * health data, so it is stored AES-256-GCM ENCRYPTED at rest in the new + * `EcgRecording.waveformEncrypted` Bytes column (fail-closed crypto, the + * `CoachMessage.encryptedContent` precedent). The descriptors (frequency, + * sample count, duration, lead, average HR, AFib verdict) stay plaintext so a + * future trace renderer reads them without a per-row decrypt. RENDERING UI IS + * OUT OF SCOPE for v1.19.0 — this remains ingest-only. + * * 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 @@ -39,8 +51,9 @@ import type { } from "@/generated/prisma/client"; import { prisma } from "@/lib/db"; -import { getEvent } from "@/lib/logging/context"; +import { annotate, getEvent } from "@/lib/logging/context"; import { safeFetch } from "@/lib/safe-fetch"; +import { encryptWaveformToBytes } from "./ecg-waveform-codec"; import { collapseToTypeDayKeys, recomputeBucketsForMeasurement, @@ -172,6 +185,160 @@ export async function fetchWithingsHeartList( return results; } +/** + * The Heart `get` response body for a single ECG signal. `signal` is the + * micro-volt sample array; `sampling_frequency` is the rate in Hz used to map + * a sample index to its time offset. `heart_rate` (average BPM for the strip) + * and `wavelets` are optional and source-version dependent — we capture the + * average HR when present and ignore the wavelet transform. + */ +export interface WithingsHeartSignal { + signal?: number[] | null; + sampling_frequency?: number | null; + heart_rate?: number | null; +} + +/** + * Fetch the full waveform for one ECG recording via `action=get`. Returns the + * raw signal body (no DB write) so the caller composes the encrypt + persist + * step. Returns `null` when the recording carries no usable signal array so + * the caller can skip waveform storage without failing the whole sync. + */ +export async function fetchWithingsHeartSignal( + accessToken: string, + signalId: number, +): Promise { + const params = new URLSearchParams({ + action: "get", + signalid: String(signalId), + }); + + const callStart = 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: "fetchWithingsHeartSignal", + duration_ms: Math.round(performance.now() - callStart), + 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 signal = Array.isArray(body.signal) + ? body.signal.filter((n: unknown): n is number => typeof n === "number") + : null; + if (!signal || signal.length === 0) return null; + + return { + signal, + sampling_frequency: + typeof body.sampling_frequency === "number" + ? body.sampling_frequency + : null, + heart_rate: typeof body.heart_rate === "number" ? body.heart_rate : null, + }; +} + +/** + * Fetch + encrypt + persist one ECG waveform. Idempotent: upserts on + * `(userId, source, externalRecordingId)` so a re-sync overwrites the same + * recording in place rather than inserting a duplicate. Returns `true` when a + * waveform row was written, `false` when the recording carried no usable + * signal (nothing to store). Throws on an upstream / persist error so the + * caller can warn without failing the verdict ingest. + */ +async function captureEcgWaveform(params: { + userId: string; + accessToken: string; + signalId: number; + externalRecordingId: string; + recordedAt: Date; + classification: RhythmClassification; + measurementId: string | null; +}): Promise { + const signal = await fetchWithingsHeartSignal( + params.accessToken, + params.signalId, + ); + if (!signal || !signal.signal || signal.signal.length === 0) { + return false; + } + + const samples = signal.signal; + const samplingFrequency = + typeof signal.sampling_frequency === "number" && + Number.isFinite(signal.sampling_frequency) && + signal.sampling_frequency > 0 + ? Math.round(signal.sampling_frequency) + : 0; + const sampleCount = samples.length; + const durationSeconds = + samplingFrequency > 0 ? sampleCount / samplingFrequency : null; + const averageHeartRate = + typeof signal.heart_rate === "number" && Number.isFinite(signal.heart_rate) + ? Math.round(signal.heart_rate) + : null; + + // Encrypt the raw sample array BEFORE it ever reaches the row. Fail-closed: + // a crypto error throws here and the waveform is never written as plaintext. + const waveformEncrypted = encryptWaveformToBytes(samples); + + // Build the `data` object field-by-field (no mass assignment); `userId` is + // the integration owner narrowed by the caller, never a client field. + await prisma.ecgRecording.upsert({ + where: { + userId_source_externalRecordingId: { + userId: params.userId, + source: "WITHINGS", + externalRecordingId: params.externalRecordingId, + }, + }, + create: { + userId: params.userId, + source: "WITHINGS", + externalRecordingId: params.externalRecordingId, + recordedAt: params.recordedAt, + waveformEncrypted, + samplingFrequency, + sampleCount, + durationSeconds, + averageHeartRate, + rhythmClassification: params.classification, + measurementId: params.measurementId, + }, + update: { + recordedAt: params.recordedAt, + waveformEncrypted, + samplingFrequency, + sampleCount, + durationSeconds, + averageHeartRate, + rhythmClassification: params.classification, + measurementId: params.measurementId, + }, + }); + + return true; +} + /** * Sync ECG / AFib recordings for a single user. Walks the trailing 30-day * window, writing one `IRREGULAR_RHYTHM_NOTIFICATION` row per ECG recording @@ -226,6 +393,7 @@ export async function syncUserEcg(userId: string): Promise { } let imported = 0; + let waveformsCaptured = 0; const touched: Array<{ type: MeasurementType; measuredAt: Date }> = []; for (const entry of entries) { @@ -235,11 +403,13 @@ export async function syncUserEcg(userId: string): Promise { continue; } const measuredAt = new Date(entry.timestamp * 1000); - const signalId = entry.ecg?.signalid ?? `ts-${entry.timestamp}`; + const numericSignalId = entry.ecg?.signalid; + const signalId = numericSignalId ?? `ts-${entry.timestamp}`; const externalId = `withings:ecg:${userId}:${signalId}`; + let measurementId: string | null = null; try { - await prisma.measurement.upsert({ + const row = await prisma.measurement.upsert({ where: { userId_type_source_externalId: { userId, @@ -265,16 +435,50 @@ export async function syncUserEcg(userId: string): Promise { measuredAt, syncVersion: { increment: 1 }, }, + select: { id: true }, }); + measurementId = row.id; touched.push({ type: ECG_TYPE, measuredAt }); imported++; } catch (err) { getEvent()?.addWarning( `withings ecg: failed to upsert recording (${classification}, ${measuredAt.toISOString()}): ${err}`, ); + continue; + } + + // v1.19.0 — capture the full waveform. Only a recording with a numeric + // `signalid` can be fetched (the `ts-` fallback has no signal to GET). A + // waveform-fetch / persist failure never fails the verdict ingest — the + // EVENT row already landed above. + if (typeof numericSignalId === "number") { + try { + const captured = await captureEcgWaveform({ + userId, + accessToken: tokenInfo.accessToken, + signalId: numericSignalId, + externalRecordingId: String(numericSignalId), + recordedAt: measuredAt, + classification, + measurementId, + }); + if (captured) waveformsCaptured++; + } catch (err) { + getEvent()?.addWarning( + `withings ecg: failed to capture waveform for signal ${numericSignalId}: ${err}`, + ); + } } } + annotate({ + action: { name: "withings.ecg.sync" }, + meta: { + withings_ecg_verdicts_imported: imported, + withings_ecg_waveforms_captured: waveformsCaptured, + }, + }); + // Refresh the persistent rollup tier for every distinct (type, day) touched, // then drop the per-metric assessment caches the new events dirty. try {