Merge w11b/a3ecg into release/v1.19.0

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 `<keyId>.<base64>`; 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/);
+  });
+});

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<string, unknown> | 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");
+  });
 });

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` (`"<keyId>.<base64>"`) 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<ArrayBuffer>`, 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<ArrayBuffer> {
+  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[];
+}

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=<id>`), 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<WithingsHeartSignal | null> {
+  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<boolean> {
+  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<number> {
   }
 
   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<number> {
       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<number> {
           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 {