diff --git a/CHANGELOG.md b/CHANGELOG.md index 50fdb12c5..981f8f156 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,36 @@ ## [Unreleased] +## [1.35.3] — 2026-08-01 + +### Added + +- Apple Watch ECG recordings can now be sent directly by the iOS app, through a + new endpoint at `POST /api/insights/ecg`, instead of only arriving inside an + export archive. One recording travels per request, and the endpoint is gated + by the same module and status switches that already govern reading ECGs, so an + account with the surface switched off is unaffected. + +### Changed + +- A recording carries its own identity, so sending the same one again resolves + to the same record. A strip that arrives once inside an export archive and + again from the app ends up as a single entry rather than a duplicate or an + error. + +### Migration + +- This release removes duplicate ECG recordings. An existing database can hold + more than one row for the same recording, because the export archive and a + live client name the same strip differently. A uniqueness rule now prevents + that, and the migration resolves what is already stored before applying it. + Where two rows describe the same recording, it keeps the one holding the actual + waveform over one holding only a verdict, and it carries a removed row's link + to its rhythm event onto the row that is kept, so resolving a duplicate never + costs the recording its place in the timeline. Rows that lose are deleted, and + they cannot be recovered. Back up the database before deploying if that matters + for your instance. + ## [1.35.2] — 2026-08-01 ### Added diff --git a/docs/api/openapi.yaml b/docs/api/openapi.yaml index 4ae490966..01bb4b8d6 100644 --- a/docs/api/openapi.yaml +++ b/docs/api/openapi.yaml @@ -1,7 +1,7 @@ openapi: 3.1.0 info: title: HealthLog API - version: 1.35.2 + version: 1.35.3 description: >- Self-hosted personal-health-tracking PWA — public API surface for the iOS native client and external ingest. @@ -7993,6 +7993,44 @@ paths: "403": *a5 "422": *a3 "429": *a4 + post: + tags: + - Insights + summary: Ingest one ECG recording + description: "The live ECG ingest: one Apple Watch recording per request, for a client draining its HealthKit ECG + observer. Before this existed, a watch ECG could only reach HealthLog inside a full `export.zip`. One recording + per request because a 30 s / 512 Hz strip is ~15 360 samples. Samples are INTEGER MICRO-VOLTS (convert from + HealthKit's Volts), stored AES-256-GCM encrypted; `sampleCount` and `durationSeconds` are derived server-side. + The `classification` is the device's own verdict stored verbatim — HealthLog never reads the waveform to produce + or revise one. `source` accepts `APPLE_HEALTH` only; `userId` comes from the session and is never a body field. + Unknown body keys are rejected with a 422 naming them. No `Idempotency-Key` is needed: the recording carries its + own identity, so a retry resolves to the same row by construction — see `status` for what a re-post reports. + Limits: 32 768 samples, 2 MB body, 60 recordings per minute per user. Module-gated on `insights` and the + operator `insightStatus` assistant surface; no LLM call. Auth via cookie or Bearer." + requestBody: + required: true + content: + application/json: + schema: + $ref: "#/components/schemas/EcgIngestRequest" + responses: + "200": + description: "The recording was already stored: `updated` (same id, overwritten in place) or `duplicate` (already + present under a different id, nothing written)." + content: + application/json: + schema: + $ref: "#/components/schemas/EcgIngestResponseEnvelope" + "201": + description: "A new recording was stored (`status: inserted`)." + content: + application/json: + schema: + $ref: "#/components/schemas/EcgIngestCreatedEnvelope" + "401": *a1 + "403": *a5 + "422": *a3 + "429": *a4 /api/insights/ecg/{id}: get: tags: @@ -14481,6 +14519,81 @@ components: properties: {} description: No body fields. The user is taken from the session / Bearer and the locale from the session; the warm covers every assessment for that user. + EcgIngestRequest: + type: object + properties: + externalRecordingId: + type: string + minLength: 1 + maxLength: 120 + description: "The recording's stable id AS THE CLIENT KNOWS IT — for an Apple Health client, the `HKSample.uuid`. It is + the source's own identity, not a content hash: re-posting the same id overwrites that recording in place. It + does not have to agree with the id the `export.zip` importer derives for the same strip; the server + reconciles the two doors itself (see `status: duplicate`)." + recordedAt: + type: string + format: date-time + pattern: ^(?:(?:\d\d[2468][048]|\d\d[13579][26]|\d\d0[48]|[02468][048]00|[13579][26]00)-02-29|\d{4}-(?:(?:0[13578]|1[02])-(?:0[1-9]|[12]\d|3[01])|(?:0[469]|11)-(?:0[1-9]|[12]\d|30)|(?:02)-(?:0[1-9]|1\d|2[0-8])))T(?:(?:[01]\d|2[0-3]):[0-5]\d(?::[0-5]\d(?:\.\d+)?)?(?:Z|([+-](?:[01]\d|2[0-3]):[0-5]\d)))$ + description: "When the strip was recorded on-device, with offset. Part of the recording's identity — see `status: + duplicate`." + samplingFrequency: + type: integer + minimum: 1 + maximum: 10000 + description: Signal sampling rate in Hz (an Apple Watch ECG is 512). Part of the recording's identity, and the divisor + for the server-derived `durationSeconds`. + samples: + minItems: 1 + maxItems: 32768 + type: array + items: + type: integer + minimum: -1000000 + maximum: 1000000 + description: The waveform as INTEGER MICRO-VOLTS, in recording order — the same unit the `export.zip` CSV parser + produces, so convert from HealthKit's Volts before sending. Capped at 32 768 samples (a clean factor of two + above a 30 s / 512 Hz strip). Stored AES-256-GCM encrypted; never persisted as plaintext. + lead: + description: Recording lead label when the device reports one (e.g. `I`). + anyOf: + - type: string + minLength: 1 + maxLength: 40 + - type: "null" + averageHeartRate: + description: The device's average heart rate (BPM) for the strip. + anyOf: + - type: integer + minimum: 1 + maximum: 300 + - type: "null" + classification: + description: The RECORDING DEVICE's own verdict, stored verbatim. Only the three ECG values are accepted here. + `RhythmClassification` in the database has six members because walking-steadiness and neutral event verdicts + share it, and those appear on GET /api/insights/rhythm-events — a decoder generated from this schema must + NOT be reused there. HealthLog never reads the waveform to produce or revise a verdict. + anyOf: + - type: string + enum: + - IRREGULAR + - NOT_DETECTED + - INCONCLUSIVE + - type: "null" + source: + type: string + enum: + - APPLE_HEALTH + description: "The client's own source. `APPLE_HEALTH` only: `WITHINGS` rows are minted by the OAuth sync and `COMPUTED` + ones by the server, so neither is client-assertable — the same posture POST /api/measurements/batch takes." + required: + - externalRecordingId + - recordedAt + - samplingFrequency + - samples + - source + description: One Apple Watch ECG recording. Unknown keys are REJECTED with a 422 naming them rather than silently + dropped, because every field here is load-bearing metadata for a stored recording. `sampleCount` and + `durationSeconds` are derived server-side and must not be sent. UpdateCorrelationPatternRequest: type: object properties: @@ -29971,6 +30084,83 @@ components: - source - hasWaveform additionalProperties: false + EcgIngestResponseEnvelope: + type: object + properties: + data: + $ref: "#/components/schemas/EcgIngestResponse" + error: + type: "null" + meta: + type: object + properties: + requestId: + type: string + additionalProperties: false + required: + - data + - error + additionalProperties: false + EcgIngestResponse: + type: object + properties: + id: + type: string + description: The recording row's id (cuid). + status: + type: string + enum: + - inserted + - updated + - duplicate + description: What the write did. `inserted` — a new recording landed (HTTP 201). `updated` — this `externalRecordingId` + was already stored and the row was overwritten in place (HTTP 200); a retry is safe and creates nothing. + `duplicate` — this exact recording is already stored under a DIFFERENT id, i.e. it arrived earlier through + the `export.zip` importer (HTTP 200); nothing was written and `id` names the row that already holds it. A + recording is the same recording when `(source, recordedAt, samplingFrequency)` match, which is enforced by a + unique index rather than by convention. All three statuses mean the recording is stored and the client may + advance its cursor. + recordedAt: + type: string + format: date-time + pattern: ^(?:(?:\d\d[2468][048]|\d\d[13579][26]|\d\d0[48]|[02468][048]00|[13579][26]00)-02-29|\d{4}-(?:(?:0[13578]|1[02])-(?:0[1-9]|[12]\d|3[01])|(?:0[469]|11)-(?:0[1-9]|[12]\d|30)|(?:02)-(?:0[1-9]|1\d|2[0-8])))T(?:(?:[01]\d|2[0-3]):[0-5]\d(?::[0-5]\d(?:\.\d+)?)?(?:Z|([+-](?:[01]\d|2[0-3]):[0-5]\d)))$ + description: The `recordedAt` that was submitted, echoed back. + sampleCount: + type: integer + minimum: -9007199254740991 + maximum: 9007199254740991 + description: "Server-derived: the number of samples received." + durationSeconds: + anyOf: + - type: number + - type: "null" + description: "Server-derived: `sampleCount / samplingFrequency`." + required: + - id + - status + - recordedAt + - sampleCount + - durationSeconds + additionalProperties: false + description: The outcome of one ECG ingest. Reported honestly rather than optimistically — a re-post never claims to + have inserted. + EcgIngestCreatedEnvelope: + type: object + properties: + data: + $ref: "#/components/schemas/EcgIngestResponse" + error: + type: "null" + meta: + type: object + properties: + requestId: + type: string + additionalProperties: false + required: + - data + - error + additionalProperties: false EcgDetailResponseEnvelope: type: object properties: diff --git a/package.json b/package.json index 328052eaf..96e4eeb1f 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "healthlog", - "version": "1.35.2", + "version": "1.35.3", "description": "Self-hosted personal-health-tracking PWA with Withings integration, AI insights, and doctor-report PDF export.", "license": "PolyForm-Noncommercial-1.0.0", "homepage": "https://healthlog.dev", diff --git a/prisma/migrations/0291_ecg_recording_identity/migration.sql b/prisma/migrations/0291_ecg_recording_identity/migration.sql new file mode 100644 index 000000000..6d7f1abc5 --- /dev/null +++ b/prisma/migrations/0291_ecg_recording_identity/migration.sql @@ -0,0 +1,90 @@ +-- ECG recordings gain a second identity: what the recording IS, not what a +-- source calls it. +-- +-- `(user_id, source, external_recording_id)` is the source's own id. The +-- `export.zip` importer derives it from a hash of the recording's content, +-- the live ingest sends the device's stable sample uuid, Withings sends its +-- `signalid`. One Apple Watch strip arriving through both doors therefore +-- carries two different ids and would land as two rows. This index closes +-- that: one device does not record two different strips at the same instant +-- at the same sampling rate. +-- +-- The table already holds rows on every deployed instance, and two writers +-- have been creating them since v1.19.0, so the index cannot simply be +-- created — a pre-existing collision would abort the migration at boot. The +-- duplicates are resolved first, deterministically, and the rule is: +-- +-- 1. the row with the MOST SAMPLES wins. A stored strip beats a +-- verdict-only row with no signal to draw. This is the row a person +-- opens and sees their ECG in. +-- 2. then the row LINKED to its paired event measurement. That link is +-- how the recording is reached from the rhythm-event timeline. +-- 3. then the most RECENTLY WRITTEN row (`updated_at`), i.e. the freshest +-- version of the same recording. +-- 4. then the lowest `id`, so the choice is total and the same on every +-- instance rather than left to whatever order the planner returns. +-- +-- Before the losers go, their event link is salvaged onto the keeper when the +-- keeper has none, so resolving a duplicate never costs the recording its +-- place in the event timeline. + +-- Salvage the event link. +WITH ranked AS ( + SELECT + id, + user_id, + source, + recorded_at, + sampling_frequency, + measurement_id, + ROW_NUMBER() OVER ( + PARTITION BY user_id, source, recorded_at, sampling_frequency + ORDER BY + sample_count DESC, + (measurement_id IS NOT NULL) DESC, + updated_at DESC, + id ASC + ) AS rn + FROM "ecg_recordings" +), +donors AS ( + SELECT keeper.id AS keeper_id, MIN(loser.measurement_id) AS measurement_id + FROM ranked AS keeper + JOIN ranked AS loser + ON loser.user_id = keeper.user_id + AND loser.source = keeper.source + AND loser.recorded_at = keeper.recorded_at + AND loser.sampling_frequency = keeper.sampling_frequency + AND loser.rn > 1 + WHERE keeper.rn = 1 + AND keeper.measurement_id IS NULL + AND loser.measurement_id IS NOT NULL + GROUP BY keeper.id +) +UPDATE "ecg_recordings" AS e +SET "measurement_id" = donors.measurement_id +FROM donors +WHERE e."id" = donors.keeper_id; + +-- Drop the resolved duplicates. +WITH ranked AS ( + SELECT + id, + ROW_NUMBER() OVER ( + PARTITION BY user_id, source, recorded_at, sampling_frequency + ORDER BY + sample_count DESC, + (measurement_id IS NOT NULL) DESC, + updated_at DESC, + id ASC + ) AS rn + FROM "ecg_recordings" +) +DELETE FROM "ecg_recordings" AS e +USING ranked +WHERE e."id" = ranked.id + AND ranked.rn > 1; + +-- The guarantee itself. +CREATE UNIQUE INDEX "ecg_recordings_user_source_recorded_at_freq_key" + ON "ecg_recordings" ("user_id", "source", "recorded_at", "sampling_frequency"); diff --git a/prisma/schema.prisma b/prisma/schema.prisma index bb5745087..bc3a14196 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -1907,6 +1907,14 @@ model EcgRecording { // Re-sync overwrites in place rather than inserting a duplicate. @@unique([userId, source, externalRecordingId]) + // The same physical recording can reach us through two doors with two + // different source ids: the `export.zip` importer identifies an Apple + // Watch strip by a hash of its content, the live `POST /api/insights/ecg` + // ingest by the device's own sample uuid. One device does not record two + // different strips at the same instant at the same sampling rate, so this + // is what the recording IS rather than what a source calls it, and it + // makes the convergence a guarantee instead of a convention. + @@unique([userId, source, recordedAt, samplingFrequency], map: "ecg_recordings_user_source_recorded_at_freq_key") @@index([userId, recordedAt(sort: Desc)]) @@map("ecg_recordings") } diff --git a/public/sw.js b/public/sw.js index 7bc9df343..23a1b22c7 100644 --- a/public/sw.js +++ b/public/sw.js @@ -36,7 +36,7 @@ try { // v1.4.38.4 → v1.4.42. Do not hand-edit; bump `package.json` and rebuild. const CACHE_VERSION = (typeof self !== "undefined" && self.__APP_VERSION__) || - /* @sw-version-fallback */ "v1.35.2"; + /* @sw-version-fallback */ "v1.35.3"; const STATIC_CACHE = `healthlog-static-${CACHE_VERSION}`; const PAGE_CACHE = `healthlog-pages-${CACHE_VERSION}`; // v1.18.6 — read-only data cache for a curated allowlist of safe GET `/api/*` diff --git a/src/app/api/insights/ecg/route.ts b/src/app/api/insights/ecg/route.ts index f997721b1..76b10a488 100644 --- a/src/app/api/insights/ecg/route.ts +++ b/src/app/api/insights/ecg/route.ts @@ -1,5 +1,5 @@ /** - * v1.28.50 — ECG recording list route (metadata only, no waveform). + * ECG recording list route (metadata only, no waveform) + the live ingest. * * `GET /api/insights/ecg` returns the authenticated user's ECG recordings * as a cheap, index-covered metadata list — recorded time, duration, @@ -18,17 +18,35 @@ * gets `{ recordings: [], hasRecordings: false }`, and the client un-mounts * the whole surface rather than painting an empty card. * - * Mirrors the `rhythm-events` route gating exactly: `apiHandler` wrapper, - * cookie OR Bearer auth, `userId` narrowed from the session (never a query - * field), the `insights` module gate, and the `insightStatus` - * assistant-surface gate. No AI provider call — this is a pure DB read. + * `POST /api/insights/ecg` is the live ingest: one Apple Watch recording per + * request, sent by the native client's HealthKit reader as it observes new + * strips. Before it existed, a watch ECG could only reach HealthLog inside a + * full `export.zip`. It stores what the device reported and derives nothing + * clinical: `sampleCount` and `durationSeconds` come from the payload, the + * classification is the device's own verdict copied verbatim, and the + * waveform is never read to form an opinion. + * + * Both verbs mirror the `rhythm-events` route gating exactly: `apiHandler` + * wrapper, cookie OR Bearer auth, `userId` narrowed from the session (never a + * body or query field), the `insights` module gate, and the `insightStatus` + * assistant-surface gate. No AI provider call on either path. */ -import { apiSuccess } from "@/lib/api-response"; +import { NextRequest } from "next/server"; +import { z } from "zod/v4"; + +import { + apiError, + apiSuccess, + returnAllZodIssues, + safeJson, +} from "@/lib/api-response"; import { apiHandler, requireAuth } from "@/lib/api-handler"; import { annotate } from "@/lib/logging/context"; import { requireAssistantSurface } from "@/lib/feature-flags"; import { requireModuleEnabled } from "@/lib/modules/gate"; +import { checkRateLimit } from "@/lib/rate-limit"; import { prisma } from "@/lib/db"; +import { persistEcgRecording } from "@/lib/ecg/persist-recording"; export const dynamic = "force-dynamic"; @@ -36,6 +54,57 @@ export const dynamic = "force-dynamic"; // over time; this is a bound, not an expected ceiling. const MAX_RECORDINGS = 200; +// A 30 s Apple Watch strip at 512 Hz is ~15 360 samples. This leaves a clean +// factor of two above that, which covers a 60 s recording or a 1024 Hz one +// without inviting an unbounded array. +const MAX_SAMPLES = 32_768; + +// Backstop for the sample cap. 32 768 integer micro-volt samples serialise to +// roughly 250 KB with separators; 2 MB rejects a hostile payload before +// `JSON.parse` sees it while leaving an order of magnitude of headroom above +// any real recording. A DoS ceiling, not a tight bound. +const MAX_BODY_BYTES = 2 * 1024 * 1024; + +// One recording per request, so the limit is per recording. A person records +// a handful of strips a week; 60/min still drains a full watch history in +// minutes and stops a leaked wildcard token from saturating the write path. +const INGEST_RATE_LIMIT_MAX = 60; +const INGEST_RATE_LIMIT_WINDOW_MS = 60 * 1000; + +// Only the three ECG verdicts. `RhythmClassification` in the database has six +// members because walking-steadiness and neutral event verdicts share the +// enum; those cannot appear on an ECG row and are not accepted here. +const ecgClassificationEnum = z.enum([ + "IRREGULAR", + "NOT_DETECTED", + "INCONCLUSIVE", +]); + +// A client may only claim the source it actually is. `WITHINGS` rows are +// minted by the OAuth sync and `COMPUTED` ones by the server, so letting an +// authenticated client assert either would let it forge rows an integration +// never produced — the same posture the measurement batch takes. +const ingestSourceEnum = z.enum(["APPLE_HEALTH"]); + +// `.strict()` on purpose: every field here is load-bearing, and a silently +// dropped unknown key would mean a recording stored with wrong metadata and +// nobody told. An unknown key gets a 422 that names it. +const ecgIngestSchema = z + .object({ + externalRecordingId: z.string().min(1).max(120), + recordedAt: z.iso.datetime({ offset: true }), + samplingFrequency: z.number().int().min(1).max(10_000), + samples: z + .array(z.number().int().min(-1_000_000).max(1_000_000)) + .min(1) + .max(MAX_SAMPLES), + lead: z.string().min(1).max(40).nullable().optional(), + averageHeartRate: z.number().int().min(1).max(300).nullable().optional(), + classification: ecgClassificationEnum.nullable().optional(), + source: ingestSourceEnum, + }) + .strict(); + export const GET = apiHandler(async () => { const { user } = await requireAuth(); const m = await requireModuleEnabled(user.id, "insights"); @@ -85,3 +154,95 @@ export const GET = apiHandler(async () => { hasRecordings: recordings.length > 0, }); }); + +/** + * `POST /api/insights/ecg` — one recording per request. + * + * The outcome is reported honestly rather than optimistically, because the + * client uses it to advance its sync cursor: + * + * - `inserted` — a new recording landed. + * - `updated` — this recording id was already stored and the row was + * overwritten in place. A re-post is safe and creates nothing. + * - `duplicate` — this exact recording is already stored under a different + * id, i.e. it already came in through the `export.zip` importer. Nothing + * was written; `id` names the row that holds it. A re-sync after an + * archive import is harmless, not an error. + * + * There is no `Idempotency-Key` on this route and none is needed: the + * recording carries its own identity, so a retried or replayed POST resolves + * to the same row by construction rather than by a cached response. + */ +export const POST = apiHandler(async (request: NextRequest) => { + const { user } = await requireAuth(); + const m = await requireModuleEnabled(user.id, "insights"); + if (!m.enabled) return m.response; + await requireAssistantSurface("insightStatus"); + + const rl = await checkRateLimit( + `insights:ecg:ingest:${user.id}`, + INGEST_RATE_LIMIT_MAX, + INGEST_RATE_LIMIT_WINDOW_MS, + ); + if (!rl.allowed) { + annotate({ + action: { name: "insights.ecg.ingest" }, + meta: { outcome: "rate_limited" }, + }); + return apiError("Too many ECG submissions, try again later", 429); + } + + const { data: rawBody, error: jsonError } = await safeJson(request, { + maxBytes: MAX_BODY_BYTES, + }); + if (jsonError) return jsonError; + + const parsed = ecgIngestSchema.safeParse(rawBody); + if (!parsed.success) { + annotate({ + action: { name: "insights.ecg.ingest" }, + meta: { outcome: "invalid" }, + }); + return returnAllZodIssues(parsed.error); + } + + const body = parsed.data; + const result = await persistEcgRecording( + { + // Narrowed from the session. The body has no `userId` field and a + // caller cannot name one. + userId: user.id, + source: body.source, + externalRecordingId: body.externalRecordingId, + recordedAt: new Date(body.recordedAt), + samples: body.samples, + samplingFrequency: body.samplingFrequency, + lead: body.lead ?? null, + averageHeartRate: body.averageHeartRate ?? null, + // The device's own verdict, stored verbatim. Never derived here. + rhythmClassification: body.classification ?? null, + }, + prisma, + ); + + annotate({ + action: { name: "insights.ecg.ingest" }, + meta: { + outcome: result.outcome, + source: body.source, + sampleCount: result.sampleCount, + samplingFrequency: body.samplingFrequency, + }, + }); + + return apiSuccess( + { + id: result.id, + status: result.outcome, + recordedAt: body.recordedAt, + sampleCount: result.sampleCount, + durationSeconds: result.durationSeconds, + }, + result.outcome === "inserted" ? 201 : 200, + ); +}); diff --git a/src/lib/apple-health/ecg-import.ts b/src/lib/apple-health/ecg-import.ts index c48096758..4e512b404 100644 --- a/src/lib/apple-health/ecg-import.ts +++ b/src/lib/apple-health/ecg-import.ts @@ -1,15 +1,24 @@ /** - * Transactional persistence for one normalized Apple Health ECG recording. + * Persistence for one normalized Apple Health ECG recording out of an + * `export.zip`. * * The stable identity is derived from canonical recording content, never the - * archive filename. The database uniqueness constraint adds the user boundary, - * so identical recordings deduplicate for one user while remaining independent + * archive filename, so the same strip re-imported under a different filename + * deduplicates. The database uniqueness constraint adds the user boundary, so + * identical recordings deduplicate for one user while remaining independent * across tenants. + * + * The write itself lives in `@/lib/ecg/persist-recording`, shared with the + * live `POST /api/insights/ecg` ingest and the Withings sync. That is also + * where the second identity — `(userId, source, recordedAt, + * samplingFrequency)` — is honoured: a strip already stored from the live + * sync under its device sample uuid reports `skipped` here rather than + * landing a second time under its content hash. */ import { createHash } from "node:crypto"; import type { PrismaClient } from "@/generated/prisma/client"; import type { NormalizedAppleHealthEcg } from "@/lib/apple-health/ecg-csv"; -import { encryptWaveformToBytes } from "@/lib/withings/ecg-waveform-codec"; +import { persistEcgRecording } from "@/lib/ecg/persist-recording"; export type AppleHealthEcgImportOutcome = "imported" | "updated" | "skipped"; @@ -38,56 +47,24 @@ export async function importAppleHealthEcg(input: { const { userId, ecg, prisma } = input; if (ecg.samples.length === 0) return "skipped"; - const externalRecordingId = recordingIdentity(ecg); - // Encryption is deliberately completed before entering the transaction. - // Crypto failure therefore fails closed without opening a write transaction. - const waveformEncrypted = encryptWaveformToBytes(ecg.samples); - const sampleCount = ecg.samples.length; - const durationSeconds = sampleCount / ecg.samplingFrequency; + const result = await persistEcgRecording( + { + userId, + source: "APPLE_HEALTH", + externalRecordingId: recordingIdentity(ecg), + recordedAt: ecg.recordedAt, + samples: ecg.samples, + samplingFrequency: ecg.samplingFrequency, + lead: ecg.lead, + averageHeartRate: ecg.averageHeartRate, + rhythmClassification: ecg.rhythmClassification, + }, + prisma, + ); - try { - return await prisma.$transaction(async (tx) => { - const where = { - userId_source_externalRecordingId: { - userId, - source: "APPLE_HEALTH" as const, - externalRecordingId, - }, - }; - const existing = await tx.ecgRecording.findUnique({ - where, - select: { id: true }, - }); - await tx.ecgRecording.upsert({ - where, - create: { - userId, - source: "APPLE_HEALTH", - externalRecordingId, - recordedAt: ecg.recordedAt, - waveformEncrypted, - samplingFrequency: ecg.samplingFrequency, - sampleCount, - durationSeconds, - lead: ecg.lead, - averageHeartRate: ecg.averageHeartRate, - rhythmClassification: ecg.rhythmClassification, - }, - update: { - recordedAt: ecg.recordedAt, - waveformEncrypted, - samplingFrequency: ecg.samplingFrequency, - sampleCount, - durationSeconds, - lead: ecg.lead, - averageHeartRate: ecg.averageHeartRate, - rhythmClassification: ecg.rhythmClassification, - }, - }); - return existing ? "updated" : "imported"; - }); - } finally { - waveformEncrypted.fill(0); - ecg.samples.fill(0); - } + if (result.outcome === "inserted") return "imported"; + // The live ingest already holds this strip under its device sample uuid. + // Nothing was written and nothing was lost. + if (result.outcome === "duplicate") return "skipped"; + return "updated"; } diff --git a/src/lib/ecg/__tests__/persist-recording.test.ts b/src/lib/ecg/__tests__/persist-recording.test.ts new file mode 100644 index 000000000..9a0319a19 --- /dev/null +++ b/src/lib/ecg/__tests__/persist-recording.test.ts @@ -0,0 +1,163 @@ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +vi.mock("@/lib/withings/ecg-waveform-codec", () => ({ + encryptWaveformToBytes: vi.fn(), +})); + +import { encryptWaveformToBytes } from "@/lib/withings/ecg-waveform-codec"; +import { + ecgDurationSeconds, + persistEcgRecording, +} from "@/lib/ecg/persist-recording"; + +function fakeDb() { + const findUnique = vi.fn(); + const upsert = vi.fn(); + const $transaction = vi.fn( + async (fn: (tx: unknown) => unknown) => + await fn({ ecgRecording: { findUnique, upsert } }), + ); + return { db: { $transaction }, findUnique, upsert, $transaction }; +} + +function input(overrides: Record = {}) { + return { + userId: "user-1", + source: "APPLE_HEALTH" as const, + externalRecordingId: "hk-uuid-1", + recordedAt: new Date("2026-07-18T06:14:03.000Z"), + samples: [12, -7, 3], + samplingFrequency: 512, + ...overrides, + }; +} + +beforeEach(() => { + vi.clearAllMocks(); + vi.mocked(encryptWaveformToBytes).mockImplementation( + () => new Uint8Array(new ArrayBuffer(4)), + ); +}); + +describe("ecgDurationSeconds", () => { + it("divides the sample count by the sampling rate", () => { + expect(ecgDurationSeconds(15_360, 512)).toBe(30); + }); + + it("is null when the source reported no sampling rate", () => { + expect(ecgDurationSeconds(15_360, 0)).toBeNull(); + }); +}); + +describe("persistEcgRecording", () => { + it("fails closed on a crypto error without ever opening a write transaction", async () => { + vi.mocked(encryptWaveformToBytes).mockImplementation(() => { + throw new Error("no encryption key"); + }); + const { db, $transaction } = fakeDb(); + + await expect(persistEcgRecording(input(), db as never)).rejects.toThrow( + "no encryption key", + ); + expect($transaction).not.toHaveBeenCalled(); + }); + + it("inserts when neither identity matches an existing row", async () => { + const { db, findUnique, upsert } = fakeDb(); + findUnique.mockResolvedValue(null); + upsert.mockResolvedValue({ id: "row-new" }); + + const result = await persistEcgRecording(input(), db as never); + + expect(result).toMatchObject({ + outcome: "inserted", + id: "row-new", + sampleCount: 3, + }); + expect(upsert).toHaveBeenCalledTimes(1); + }); + + it("updates in place when the source id is already stored", async () => { + const { db, findUnique, upsert } = fakeDb(); + findUnique.mockResolvedValueOnce({ id: "row-existing" }); + upsert.mockResolvedValue({ id: "row-existing" }); + + const result = await persistEcgRecording(input(), db as never); + + expect(result.outcome).toBe("updated"); + expect(result.id).toBe("row-existing"); + // The source-id lookup answered; the recording-identity lookup is not + // needed and must not run. + expect(findUnique).toHaveBeenCalledTimes(1); + }); + + it("writes nothing when the recording is already stored under another id", async () => { + const { db, findUnique, upsert } = fakeDb(); + findUnique + .mockResolvedValueOnce(null) + .mockResolvedValueOnce({ id: "row-from-the-archive" }); + + const result = await persistEcgRecording(input(), db as never); + + expect(result.outcome).toBe("duplicate"); + expect(result.id).toBe("row-from-the-archive"); + expect(upsert).not.toHaveBeenCalled(); + }); + + it("looks the twin up by the recording's own identity, not by the source id", async () => { + const { db, findUnique, upsert } = fakeDb(); + findUnique.mockResolvedValue(null); + upsert.mockResolvedValue({ id: "row-new" }); + + await persistEcgRecording(input(), db as never); + + expect(findUnique.mock.calls[1][0].where).toEqual({ + userId_source_recordedAt_samplingFrequency: { + userId: "user-1", + source: "APPLE_HEALTH", + recordedAt: new Date("2026-07-18T06:14:03.000Z"), + samplingFrequency: 512, + }, + }); + }); + + it("builds the row field by field and narrows the owner to the caller's user", async () => { + const { db, findUnique, upsert } = fakeDb(); + findUnique.mockResolvedValue(null); + upsert.mockResolvedValue({ id: "row-new" }); + + await persistEcgRecording( + input({ + lead: "I", + averageHeartRate: 64, + rhythmClassification: "IRREGULAR", + }), + db as never, + ); + + const created = upsert.mock.calls[0][0].create; + expect(created).toMatchObject({ + userId: "user-1", + source: "APPLE_HEALTH", + externalRecordingId: "hk-uuid-1", + samplingFrequency: 512, + sampleCount: 3, + lead: "I", + averageHeartRate: 64, + rhythmClassification: "IRREGULAR", + measurementId: null, + }); + expect(created.durationSeconds).toBeCloseTo(3 / 512, 10); + }); + + it("leaves no plaintext samples behind in the caller's array", async () => { + const { db, findUnique, upsert } = fakeDb(); + findUnique.mockResolvedValue(null); + upsert.mockResolvedValue({ id: "row-new" }); + + const payload = input(); + await persistEcgRecording(payload, db as never); + + expect(payload.samples).toEqual([0, 0, 0]); + }); +}); diff --git a/src/lib/ecg/persist-recording.ts b/src/lib/ecg/persist-recording.ts new file mode 100644 index 000000000..6c26a0eff --- /dev/null +++ b/src/lib/ecg/persist-recording.ts @@ -0,0 +1,213 @@ +/** + * The one persistence body for an ECG recording, whichever door it came + * through. + * + * Three writers put rows in `ecg_recordings`: the Withings OAuth sync, the + * Apple Health `export.zip` importer, and the live `POST /api/insights/ecg` + * ingest. They disagree about where a recording comes from and about how it + * is identified, and about nothing else — so encryption, the derived + * `sampleCount` / `durationSeconds`, the upsert, and the duplicate rule live + * here and only here. + * + * Two identities, one row: + * + * - `(userId, source, externalRecordingId)` is the SOURCE's own idea of a + * recording. The archive importer derives it from the recording's + * content (a hash), the live ingest sends the device's stable sample + * uuid, Withings sends its `signalid`. A re-post under the same id + * overwrites in place. + * - `(userId, source, recordedAt, samplingFrequency)` is what the recording + * actually IS. One device does not produce two different recordings for + * the same instant at the same sampling rate, so this catches the same + * physical strip arriving under two different source ids — the archive's + * content hash and the live sync's sample uuid for one Apple Watch ECG. + * The second arrival is reported as `duplicate` and writes nothing. + * + * The duplicate is resolved by an explicit lookup rather than by reading the + * shape of a Prisma unique-violation error: the database index is the + * structural backstop for a concurrent race, not the thing the outcome is + * derived from. A race that slips past the lookup still cannot create a twin + * — it raises `P2002`, and the handler below re-resolves it to the same + * `duplicate` verdict. + * + * The waveform is encrypted BEFORE the transaction opens, so a crypto failure + * fails closed without a write transaction ever existing. Both the ciphertext + * buffer and the caller's plaintext sample array are zero-filled afterwards; + * callers must not reuse `samples` after the call. + */ +import { Prisma } from "@/generated/prisma/client"; +import type { + MeasurementSource, + PrismaClient, + RhythmClassification, +} from "@/generated/prisma/client"; +import { encryptWaveformToBytes } from "@/lib/withings/ecg-waveform-codec"; + +/** + * What the write did. + * + * - `inserted` — a new row. + * - `updated` — a row with this source id already existed and was overwritten. + * - `duplicate` — this exact recording is already stored under a DIFFERENT + * source id (the other door). Nothing was written; `id` names the row that + * already holds it. + */ +export type EcgPersistOutcome = "inserted" | "updated" | "duplicate"; + +export interface EcgRecordingWrite { + userId: string; + source: MeasurementSource; + /** The source's own stable id for this recording. */ + externalRecordingId: string; + recordedAt: Date; + /** Integer micro-volt samples. Zero-filled by this function. */ + samples: number[]; + /** Signal sampling rate in Hz; 0 when the source omitted it. */ + samplingFrequency: number; + lead?: string | null; + averageHeartRate?: number | null; + /** The DEVICE's own verdict, verbatim. Never derived from the waveform. */ + rhythmClassification?: RhythmClassification | null; + /** FK to the paired EVENT measurement row, when the writer knows it. */ + measurementId?: string | null; +} + +export interface EcgPersistResult { + outcome: EcgPersistOutcome; + /** The row holding this recording — the written one, or the twin. */ + id: string; + sampleCount: number; + durationSeconds: number | null; +} + +/** + * `sampleCount / samplingFrequency`, or null when the source omitted the + * frequency — a strip with no sampling rate has no knowable duration, and + * dividing by zero would store Infinity. + */ +export function ecgDurationSeconds( + sampleCount: number, + samplingFrequency: number, +): number | null { + return samplingFrequency > 0 ? sampleCount / samplingFrequency : null; +} + +type TransactionalClient = Pick; + +export async function persistEcgRecording( + input: EcgRecordingWrite, + db: TransactionalClient, +): Promise { + const { + userId, + source, + externalRecordingId, + recordedAt, + samples, + samplingFrequency, + } = input; + + const sampleCount = samples.length; + const durationSeconds = ecgDurationSeconds(sampleCount, samplingFrequency); + + // Encryption happens before the transaction opens. A crypto failure + // therefore fails closed without a write transaction ever existing, and + // plaintext never reaches the row. + const waveformEncrypted = encryptWaveformToBytes(samples); + + const sourceKey = { + userId_source_externalRecordingId: { + userId, + source, + externalRecordingId, + }, + }; + const recordingKey = { + userId_source_recordedAt_samplingFrequency: { + userId, + source, + recordedAt, + samplingFrequency, + }, + }; + + // Built field-by-field from the typed input; the parsed request object is + // never spread into `data`. + const columns = { + recordedAt, + waveformEncrypted, + samplingFrequency, + sampleCount, + durationSeconds, + lead: input.lead ?? null, + averageHeartRate: input.averageHeartRate ?? null, + rhythmClassification: input.rhythmClassification ?? null, + measurementId: input.measurementId ?? null, + }; + + try { + return await db.$transaction(async (tx) => { + const existing = await tx.ecgRecording.findUnique({ + where: sourceKey, + select: { id: true }, + }); + + if (!existing) { + const twin = await tx.ecgRecording.findUnique({ + where: recordingKey, + select: { id: true }, + }); + if (twin) { + return { + outcome: "duplicate" as const, + id: twin.id, + sampleCount, + durationSeconds, + }; + } + } + + const row = await tx.ecgRecording.upsert({ + where: sourceKey, + create: { userId, source, externalRecordingId, ...columns }, + update: columns, + select: { id: true }, + }); + + return { + outcome: existing ? ("updated" as const) : ("inserted" as const), + id: row.id, + sampleCount, + durationSeconds, + }; + }); + } catch (err) { + // A concurrent writer claimed the recording between the lookup and the + // write. The index held; re-resolve to the same honest verdict rather + // than surfacing a constraint violation to a client that did nothing + // wrong. + if ( + err instanceof Prisma.PrismaClientKnownRequestError && + err.code === "P2002" + ) { + const twin = await db.$transaction(async (tx) => + tx.ecgRecording.findUnique({ + where: recordingKey, + select: { id: true }, + }), + ); + if (twin) { + return { + outcome: "duplicate", + id: twin.id, + sampleCount, + durationSeconds, + }; + } + } + throw err; + } finally { + waveformEncrypted.fill(0); + samples.fill(0); + } +} diff --git a/src/lib/openapi/routes/insights/paths.ts b/src/lib/openapi/routes/insights/paths.ts index 07f5185b2..e749c10e3 100644 --- a/src/lib/openapi/routes/insights/paths.ts +++ b/src/lib/openapi/routes/insights/paths.ts @@ -37,6 +37,8 @@ import { insightsPregenerateResponse, dashboardSnapshotResponse, ecgListResponse, + ecgIngestRequest, + ecgIngestResponse, ecgDetailQuery, ecgDetailResponse, rhythmEventsResponse, @@ -457,6 +459,45 @@ export const insightsPaths: NonNullable = { ...stdResponses, }, }, + post: { + tags: ["Insights"], + summary: "Ingest one ECG recording", + description: + "The live ECG ingest: one Apple Watch recording per request, for a client draining its HealthKit ECG observer. Before this existed, a watch ECG could only reach HealthLog inside a full `export.zip`. One recording per request because a 30 s / 512 Hz strip is ~15 360 samples. Samples are INTEGER MICRO-VOLTS (convert from HealthKit's Volts), stored AES-256-GCM encrypted; `sampleCount` and `durationSeconds` are derived server-side. The `classification` is the device's own verdict stored verbatim — HealthLog never reads the waveform to produce or revise one. `source` accepts `APPLE_HEALTH` only; `userId` comes from the session and is never a body field. Unknown body keys are rejected with a 422 naming them. No `Idempotency-Key` is needed: the recording carries its own identity, so a retry resolves to the same row by construction — see `status` for what a re-post reports. Limits: 32 768 samples, 2 MB body, 60 recordings per minute per user. Module-gated on `insights` and the operator `insightStatus` assistant surface; no LLM call. Auth via cookie or Bearer.", + requestBody: { + required: true, + content: { + "application/json": { schema: ecgIngestRequest }, + }, + }, + responses: { + "200": { + description: + "The recording was already stored: `updated` (same id, overwritten in place) or `duplicate` (already present under a different id, nothing written).", + content: { + "application/json": { + schema: dataEnvelope( + ecgIngestResponse, + "EcgIngestResponseEnvelope", + ), + }, + }, + }, + "201": { + description: "A new recording was stored (`status: inserted`).", + content: { + "application/json": { + schema: dataEnvelope( + ecgIngestResponse, + "EcgIngestCreatedEnvelope", + ), + }, + }, + }, + ...moduleDisabledResponse, + ...stdResponses, + }, + }, }, "/api/insights/ecg/{id}": { get: { diff --git a/src/lib/openapi/routes/insights/schemas.ts b/src/lib/openapi/routes/insights/schemas.ts index a36af7e3d..a40ba2a24 100644 --- a/src/lib/openapi/routes/insights/schemas.ts +++ b/src/lib/openapi/routes/insights/schemas.ts @@ -1439,6 +1439,96 @@ export const ecgDetailResponse = z "One ECG recording's decrypted waveform plus metadata and the DEVICE's verbatim classification. Ownership is narrowed in the query where — a foreign / unknown id 404s. The waveform is AES-256-GCM at rest, decrypted through the fail-closed codec. HealthLog does not interpret the trace, measure intervals, annotate beats, or emit a verdict of its own. no-store.", }); +// The live ECG ingest. One recording per request: a 30 s Apple Watch strip at +// 512 Hz is ~15 360 samples, so a batch of them is not a shape anyone wants on +// the wire. `userId` is never a body field — it comes from the session. +export const ecgIngestRequest = z + .object({ + externalRecordingId: z + .string() + .min(1) + .max(120) + .describe( + "The recording's stable id AS THE CLIENT KNOWS IT — for an Apple Health client, the `HKSample.uuid`. It is the source's own identity, not a content hash: re-posting the same id overwrites that recording in place. It does not have to agree with the id the `export.zip` importer derives for the same strip; the server reconciles the two doors itself (see `status: duplicate`).", + ), + recordedAt: z.iso + .datetime({ offset: true }) + .describe( + "When the strip was recorded on-device, with offset. Part of the recording's identity — see `status: duplicate`.", + ), + samplingFrequency: z + .number() + .int() + .min(1) + .max(10_000) + .describe( + "Signal sampling rate in Hz (an Apple Watch ECG is 512). Part of the recording's identity, and the divisor for the server-derived `durationSeconds`.", + ), + samples: z + .array(z.number().int().min(-1_000_000).max(1_000_000)) + .min(1) + .max(32_768) + .describe( + "The waveform as INTEGER MICRO-VOLTS, in recording order — the same unit the `export.zip` CSV parser produces, so convert from HealthKit's Volts before sending. Capped at 32 768 samples (a clean factor of two above a 30 s / 512 Hz strip). Stored AES-256-GCM encrypted; never persisted as plaintext.", + ), + lead: z + .string() + .min(1) + .max(40) + .nullable() + .optional() + .describe("Recording lead label when the device reports one (e.g. `I`)."), + averageHeartRate: z + .number() + .int() + .min(1) + .max(300) + .nullable() + .optional() + .describe("The device's average heart rate (BPM) for the strip."), + classification: ecgClassification + .optional() + .describe( + "The RECORDING DEVICE's own verdict, stored verbatim. Only the three ECG values are accepted here. `RhythmClassification` in the database has six members because walking-steadiness and neutral event verdicts share it, and those appear on GET /api/insights/rhythm-events — a decoder generated from this schema must NOT be reused there. HealthLog never reads the waveform to produce or revise a verdict.", + ), + source: z + .enum(["APPLE_HEALTH"]) + .describe( + "The client's own source. `APPLE_HEALTH` only: `WITHINGS` rows are minted by the OAuth sync and `COMPUTED` ones by the server, so neither is client-assertable — the same posture POST /api/measurements/batch takes.", + ), + }) + .meta({ + id: "EcgIngestRequest", + description: + "One Apple Watch ECG recording. Unknown keys are REJECTED with a 422 naming them rather than silently dropped, because every field here is load-bearing metadata for a stored recording. `sampleCount` and `durationSeconds` are derived server-side and must not be sent.", + }); + +export const ecgIngestResponse = z + .object({ + id: z.string().describe("The recording row's id (cuid)."), + status: z + .enum(["inserted", "updated", "duplicate"]) + .describe( + "What the write did. `inserted` — a new recording landed (HTTP 201). `updated` — this `externalRecordingId` was already stored and the row was overwritten in place (HTTP 200); a retry is safe and creates nothing. `duplicate` — this exact recording is already stored under a DIFFERENT id, i.e. it arrived earlier through the `export.zip` importer (HTTP 200); nothing was written and `id` names the row that already holds it. A recording is the same recording when `(source, recordedAt, samplingFrequency)` match, which is enforced by a unique index rather than by convention. All three statuses mean the recording is stored and the client may advance its cursor.", + ), + recordedAt: z.iso + .datetime({ offset: true }) + .describe("The `recordedAt` that was submitted, echoed back."), + sampleCount: z + .number() + .int() + .describe("Server-derived: the number of samples received."), + durationSeconds: z + .number() + .nullable() + .describe("Server-derived: `sampleCount / samplingFrequency`."), + }) + .meta({ + id: "EcgIngestResponse", + description: + "The outcome of one ECG ingest. Reported honestly rather than optimistically — a re-post never claims to have inserted.", + }); + // v1.10.0 — device-flagged event timeline (rhythm-events, WX-B). Deliberately // a SEPARATE enum from `ecgClassification` above: an EVENT row's // `rhythmClassification` carries the FULL RhythmClassification set (the diff --git a/src/lib/withings/__tests__/sync-ecg.test.ts b/src/lib/withings/__tests__/sync-ecg.test.ts index 5baca7a92..c916e0a15 100644 --- a/src/lib/withings/__tests__/sync-ecg.test.ts +++ b/src/lib/withings/__tests__/sync-ecg.test.ts @@ -8,19 +8,31 @@ */ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; -vi.mock("@/lib/db", () => ({ - prisma: { - measurement: { - upsert: vi.fn(), +// The waveform write goes through the shared writer +// (`@/lib/ecg/persist-recording`), which opens a transaction and looks the +// recording up under both identities before upserting. The mock hands the +// transaction the same `ecgRecording` delegate, so these tests still watch the +// real assembly rather than a stub standing in for it. +vi.mock("@/lib/db", () => { + const ecgRecording = { + upsert: vi.fn(), + findUnique: vi.fn(), + }; + return { + prisma: { + measurement: { + upsert: vi.fn(), + }, + ecgRecording, + withingsConnection: { + findUnique: vi.fn(), + }, + $transaction: vi.fn( + async (fn: (tx: unknown) => unknown) => await fn({ ecgRecording }), + ), }, - ecgRecording: { - upsert: vi.fn(), - }, - withingsConnection: { - findUnique: vi.fn(), - }, - }, -})); + }; +}); vi.mock("../ecg-waveform-codec", () => ({ encryptWaveformToBytes: vi.fn((samples: number[]) => @@ -126,7 +138,14 @@ beforeEach(() => { vi.mocked(prisma.measurement.upsert).mockResolvedValue({ id: "m-1", } as never); - vi.mocked(prisma.ecgRecording.upsert).mockResolvedValue({} as never); + vi.mocked(prisma.ecgRecording.findUnique).mockResolvedValue(null as never); + vi.mocked(prisma.ecgRecording.upsert).mockResolvedValue({ + id: "ecg-1", + } as never); + vi.mocked(prisma.$transaction).mockImplementation( + (async (fn: (tx: unknown) => unknown) => + await fn({ ecgRecording: prisma.ecgRecording })) as never, + ); }); afterEach(() => { diff --git a/src/lib/withings/sync-ecg.ts b/src/lib/withings/sync-ecg.ts index 7f226913f..5213bbb83 100644 --- a/src/lib/withings/sync-ecg.ts +++ b/src/lib/withings/sync-ecg.ts @@ -53,7 +53,7 @@ import type { import { prisma } from "@/lib/db"; import { annotate, getEvent } from "@/lib/logging/context"; import { safeFetch } from "@/lib/safe-fetch"; -import { encryptWaveformToBytes } from "./ecg-waveform-codec"; +import { persistEcgRecording } from "@/lib/ecg/persist-recording"; import { collapseToTypeDayKeys, recomputeBucketsForMeasurement, @@ -262,7 +262,9 @@ export async function fetchWithingsHeartSignal( } /** - * Fetch + encrypt + persist one ECG waveform. Idempotent: upserts on + * Fetch one ECG waveform and hand it to the shared writer + * (`@/lib/ecg/persist-recording`, also used by the Apple Health archive + * importer and the live ingest). 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 @@ -293,52 +295,28 @@ async function captureEcgWaveform(params: { 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: { + // Encryption, the derived counts, the upsert and the duplicate rule are the + // shared writer's job; `userId` is the integration owner narrowed by the + // caller, never a client field. + await persistEcgRecording( + { userId: params.userId, source: "WITHINGS", externalRecordingId: params.externalRecordingId, recordedAt: params.recordedAt, - waveformEncrypted, + samples, 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, - }, - }); + prisma, + ); return true; } diff --git a/tests/integration/ecg-ingest-roundtrip.test.ts b/tests/integration/ecg-ingest-roundtrip.test.ts new file mode 100644 index 000000000..94a104947 --- /dev/null +++ b/tests/integration/ecg-ingest-roundtrip.test.ts @@ -0,0 +1,368 @@ +/** + * `POST /api/insights/ecg` end to end against real Postgres. + * + * Everything here goes through the actual route handlers — the ingest, the + * metadata list, and the waveform detail — because the thing worth proving is + * the assembly between them, not that each end works in isolation. What a + * client receives is read out of the real GET responses, never rebuilt from a + * fixture. + * + * The cross-door case is the point of the second unique index: the same + * physical recording reaching the export-archive importer and the live route + * carries two different source ids, and must still be one row. + */ +import { NextRequest } from "next/server"; +import { beforeEach, describe, expect, it, vi } from "vitest"; + +import { cookieJar, headerJar } from "./mock-next-headers"; +import { getPrismaClient, truncateAllTables } from "./setup"; + +process.env.ENCRYPTION_KEY ??= + "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef"; + +vi.mock("next/headers", async () => { + const { cookieJar, headerJar } = await import("./mock-next-headers"); + return { + headers: vi.fn(async () => ({ + get: (name: string) => headerJar.get(name.toLowerCase()) ?? null, + })), + cookies: vi.fn(async () => ({ + get: (name: string) => { + const value = cookieJar.get(name); + return value ? { name, value } : undefined; + }, + set: (name: string, value: string) => { + cookieJar.set(name, value); + }, + delete: (name: string) => { + cookieJar.delete(name); + }, + })), + }; +}); + +vi.mock("@/lib/db-compat", () => ({ + ensureDbCompatibility: vi.fn().mockResolvedValue(undefined), +})); + +import { + GET as getEcgList, + POST as postEcg, +} from "@/app/api/insights/ecg/route"; +import { GET as getEcgDetail } from "@/app/api/insights/ecg/[id]/route"; +import { importAppleHealthEcg } from "@/lib/apple-health/ecg-import"; + +const USER_ID = "ecg-ingest-roundtrip"; +const OTHER_USER_ID = "ecg-ingest-roundtrip-other"; +const RECORDED_AT = "2026-07-18T08:14:03+02:00"; +const SAMPLES = [12, -7, 3, 240, -180, 60]; + +type IngestBody = Record; + +function ingestPayload(overrides: IngestBody = {}): IngestBody { + return { + externalRecordingId: "8E1F3B0C-0000-4000-8000-000000000001", + recordedAt: RECORDED_AT, + samplingFrequency: 512, + samples: [...SAMPLES], + lead: "I", + averageHeartRate: 64, + classification: "NOT_DETECTED", + source: "APPLE_HEALTH", + ...overrides, + }; +} + +async function ingest(body: IngestBody): Promise<{ + status: number; + data: { + id: string; + status: string; + recordedAt: string; + sampleCount: number; + durationSeconds: number | null; + }; +}> { + const response = await postEcg( + new NextRequest("http://localhost/api/insights/ecg", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify(body), + }), + ); + return { + status: response.status, + data: (await response.json()).data, + }; +} + +async function readList(): Promise<{ + recordings: Array>; + hasRecordings: boolean; +}> { + // The list handler declares no parameters; the wrapper still receives the + // request, so hand it one the way the route's own unit test does. + const callList = getEcgList as unknown as ( + request: NextRequest, + ) => Promise; + const response = await callList( + new NextRequest("http://localhost/api/insights/ecg"), + ); + expect(response.status).toBe(200); + return (await response.json()).data; +} + +async function readDetail( + id: string, + full = false, +): Promise<{ response: Response; body: Record }> { + const url = full + ? `http://localhost/api/insights/ecg/${id}?full=1` + : `http://localhost/api/insights/ecg/${id}`; + const response = await getEcgDetail(new NextRequest(url), { + params: Promise.resolve({ id }), + }); + return { response, body: (await response.json()).data }; +} + +async function signIn(userId: string): Promise { + const session = await getPrismaClient().session.create({ + data: { userId, expiresAt: new Date(Date.now() + 60 * 60 * 1000) }, + }); + cookieJar.set("healthlog_session", session.id); +} + +beforeEach(async () => { + const prisma = getPrismaClient(); + await truncateAllTables(prisma); + cookieJar.clear(); + headerJar.clear(); + await prisma.user.createMany({ + data: [ + { + id: USER_ID, + username: "ecg-roundtrip", + email: "ecg-roundtrip@example.test", + timezone: "Europe/Berlin", + }, + { + id: OTHER_USER_ID, + username: "ecg-roundtrip-other", + email: "ecg-roundtrip-other@example.test", + timezone: "Europe/Berlin", + }, + ], + }); + await signIn(USER_ID); +}); + +describe("POST /api/insights/ecg — round trip through the real routes", () => { + it("stores a posted recording and serves it back through the list and the detail", async () => { + const posted = await ingest(ingestPayload()); + expect(posted.status).toBe(201); + expect(posted.data.status).toBe("inserted"); + expect(posted.data.sampleCount).toBe(SAMPLES.length); + expect(posted.data.durationSeconds).toBeCloseTo(SAMPLES.length / 512, 10); + + const list = await readList(); + expect(list.hasRecordings).toBe(true); + expect(list.recordings).toHaveLength(1); + expect(list.recordings[0]).toMatchObject({ + id: posted.data.id, + samplingFrequency: 512, + sampleCount: SAMPLES.length, + averageHeartRate: 64, + lead: "I", + classification: "NOT_DETECTED", + source: "APPLE_HEALTH", + hasWaveform: true, + }); + expect( + new Date(list.recordings[0].recordedAt as string).toISOString(), + ).toBe(new Date(RECORDED_AT).toISOString()); + + const detail = await readDetail(posted.data.id, true); + expect(detail.response.status).toBe(200); + expect(detail.body.samples).toEqual(SAMPLES); + expect(detail.body.decimated).toBe(false); + expect(detail.body.classification).toBe("NOT_DETECTED"); + }); + + it("never writes the waveform as plaintext", async () => { + const posted = await ingest(ingestPayload()); + const row = await getPrismaClient().ecgRecording.findUniqueOrThrow({ + where: { id: posted.data.id }, + }); + const stored = Buffer.from(row.waveformEncrypted).toString("utf8"); + // The column holds the `.` envelope and nothing else. A + // serialised sample array cannot satisfy that shape — it carries + // brackets, commas and minus signs — so this rules plaintext out without + // asking whether some digit happens to appear in a base64 blob, which it + // routinely does. + expect(stored).toMatch(/^[A-Za-z0-9_-]+\.[A-Za-z0-9+/=]+$/); + expect(stored).not.toContain(JSON.stringify(SAMPLES)); + }); + + it("reports a re-post as updated and does not grow the table", async () => { + const first = await ingest(ingestPayload()); + const second = await ingest(ingestPayload()); + + expect(second.status).toBe(200); + expect(second.data.status).toBe("updated"); + expect(second.data.id).toBe(first.data.id); + expect( + await getPrismaClient().ecgRecording.count({ + where: { userId: USER_ID }, + }), + ).toBe(1); + expect((await readList()).recordings).toHaveLength(1); + }); + + it("overwrites the stored recording in place when a re-post revises it", async () => { + const first = await ingest(ingestPayload()); + const revised = [1, 2, 3, 4]; + const second = await ingest( + ingestPayload({ samples: revised, classification: "IRREGULAR" }), + ); + + expect(second.data.status).toBe("updated"); + expect(second.data.id).toBe(first.data.id); + const detail = await readDetail(first.data.id, true); + expect(detail.body.samples).toEqual(revised); + expect(detail.body.classification).toBe("IRREGULAR"); + }); + + it("keeps one row when the same recording also arrives through the export archive", async () => { + const posted = await ingest(ingestPayload()); + + // The archive importer identifies the strip by a hash of its content, so + // it arrives under a completely different externalRecordingId. + const outcome = await importAppleHealthEcg({ + userId: USER_ID, + ecg: { + recordedAt: new Date(RECORDED_AT), + samplingFrequency: 512, + samples: [...SAMPLES], + lead: "I", + averageHeartRate: 64, + rhythmClassification: "NOT_DETECTED", + }, + prisma: getPrismaClient(), + }); + + expect(outcome).toBe("skipped"); + const rows = await getPrismaClient().ecgRecording.findMany({ + where: { userId: USER_ID }, + }); + expect(rows).toHaveLength(1); + expect(rows[0].id).toBe(posted.data.id); + expect((await readList()).recordings).toHaveLength(1); + }); + + it("reports duplicate when the export archive got there first", async () => { + await importAppleHealthEcg({ + userId: USER_ID, + ecg: { + recordedAt: new Date(RECORDED_AT), + samplingFrequency: 512, + samples: [...SAMPLES], + lead: "I", + averageHeartRate: 64, + rhythmClassification: "NOT_DETECTED", + }, + prisma: getPrismaClient(), + }); + const archived = await getPrismaClient().ecgRecording.findFirstOrThrow({ + where: { userId: USER_ID }, + }); + + const posted = await ingest(ingestPayload()); + expect(posted.status).toBe(200); + expect(posted.data.status).toBe("duplicate"); + expect(posted.data.id).toBe(archived.id); + expect( + await getPrismaClient().ecgRecording.count({ + where: { userId: USER_ID }, + }), + ).toBe(1); + // Nothing was written: the archive's own external id still owns the row. + expect( + ( + await getPrismaClient().ecgRecording.findUniqueOrThrow({ + where: { id: archived.id }, + }) + ).externalRecordingId, + ).toBe(archived.externalRecordingId); + }); + + it("stores distinct recordings distinctly", async () => { + await ingest(ingestPayload()); + const other = await ingest( + ingestPayload({ + externalRecordingId: "8E1F3B0C-0000-4000-8000-000000000002", + recordedAt: "2026-07-19T08:14:03+02:00", + }), + ); + expect(other.data.status).toBe("inserted"); + expect((await readList()).recordings).toHaveLength(2); + }); + + it("writes to the session's user and cannot reach another account's rows", async () => { + const mine = await ingest(ingestPayload()); + + cookieJar.clear(); + await signIn(OTHER_USER_ID); + // Same recording id, same instant — a different account entirely. + const theirs = await ingest(ingestPayload()); + expect(theirs.status).toBe(201); + expect(theirs.data.status).toBe("inserted"); + expect(theirs.data.id).not.toBe(mine.data.id); + + expect((await readList()).recordings).toHaveLength(1); + const foreign = await readDetail(mine.data.id, true); + expect(foreign.response.status).toBe(404); + + const owners = await getPrismaClient().ecgRecording.findMany({ + select: { userId: true }, + }); + expect(owners.map((r) => r.userId).sort()).toEqual( + [USER_ID, OTHER_USER_ID].sort(), + ); + }); + + it("refuses a source a client is not allowed to claim", async () => { + const response = await postEcg( + new NextRequest("http://localhost/api/insights/ecg", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify(ingestPayload({ source: "WITHINGS" })), + }), + ); + expect(response.status).toBe(422); + expect(await getPrismaClient().ecgRecording.count()).toBe(0); + }); + + it("refuses a userId in the body rather than honouring it", async () => { + const response = await postEcg( + new NextRequest("http://localhost/api/insights/ecg", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify(ingestPayload({ userId: OTHER_USER_ID })), + }), + ); + expect(response.status).toBe(422); + expect(await getPrismaClient().ecgRecording.count()).toBe(0); + }); + + it("refuses an event verdict that cannot occur on an ECG", async () => { + const response = await postEcg( + new NextRequest("http://localhost/api/insights/ecg", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify(ingestPayload({ classification: "VERY_LOW" })), + }), + ); + expect(response.status).toBe(422); + expect(await getPrismaClient().ecgRecording.count()).toBe(0); + }); +}); diff --git a/tests/integration/ecg-recording-identity-migration.test.ts b/tests/integration/ecg-recording-identity-migration.test.ts new file mode 100644 index 000000000..f57fa4e96 --- /dev/null +++ b/tests/integration/ecg-recording-identity-migration.test.ts @@ -0,0 +1,322 @@ +/** + * Migration `0291_ecg_recording_identity` against a table that ALREADY holds + * colliding rows. + * + * The testcontainer starts empty, so applying this migration during global + * setup proves nothing about the deployed instances it will actually run on: + * two writers have been filling `ecg_recordings` since v1.19.0, and a unique + * index that meets a pre-existing duplicate aborts the migration on boot. So + * this test drops the index, manufactures the collisions the resolution step + * exists for, and runs the REAL migration file through the Prisma CLI — the + * same executor `db:migrate:deploy` uses — then checks which row survived and + * that the constraint is live afterwards. + */ +import { execFileSync } from "node:child_process"; +import { dirname, join, resolve } from "node:path"; +import { fileURLToPath } from "node:url"; + +import { afterEach, beforeEach, describe, expect, it } from "vitest"; + +import { getPrismaClient, truncateAllTables } from "./setup"; + +const PROJECT_ROOT = resolve( + dirname(fileURLToPath(import.meta.url)), + "..", + "..", +); +const MIGRATION_FILE = join( + PROJECT_ROOT, + "prisma", + "migrations", + "0291_ecg_recording_identity", + "migration.sql", +); +const INDEX_NAME = "ecg_recordings_user_source_recorded_at_freq_key"; + +const USER_ID = "ecg-identity-migration"; + +function applyMigration(): void { + // Same executor and same connection `db:migrate:deploy` uses; the URL comes + // from the child environment exactly as it does for the real deploy. + execFileSync( + "pnpm", + ["exec", "prisma", "db", "execute", "--file", MIGRATION_FILE], + { + cwd: PROJECT_ROOT, + stdio: "pipe", + env: { ...process.env, DATABASE_URL: process.env.DATABASE_URL }, + }, + ); +} + +async function dropIndex(): Promise { + await getPrismaClient().$executeRawUnsafe( + `DROP INDEX IF EXISTS "${INDEX_NAME}"`, + ); +} + +async function indexExists(): Promise { + const rows = await getPrismaClient().$queryRaw>` + SELECT indexname FROM pg_indexes + WHERE tablename = 'ecg_recordings' AND indexname = ${INDEX_NAME} + `; + return rows.length === 1; +} + +/** + * Insert a row directly. `waveform_encrypted` is opaque to the migration, so + * a marker byte string is enough and keeps the fixture honest about what the + * migration does and does not read. + */ +async function insertRecording(input: { + id: string; + externalRecordingId: string; + recordedAt: string; + samplingFrequency: number; + sampleCount: number; + measurementId?: string | null; + updatedAt: string; +}): Promise { + await getPrismaClient().$executeRawUnsafe( + `INSERT INTO ecg_recordings + (id, user_id, source, external_recording_id, recorded_at, + waveform_encrypted, sampling_frequency, sample_count, duration_seconds, + measurement_id, created_at, updated_at) + VALUES ($1, $2, 'APPLE_HEALTH', $3, $4::timestamptz, $5, $6, $7, $8, $9, + $10::timestamptz, $10::timestamptz)`, + input.id, + USER_ID, + input.externalRecordingId, + input.recordedAt, + Buffer.from(`ciphertext-${input.id}`, "utf8"), + input.samplingFrequency, + input.sampleCount, + input.sampleCount / input.samplingFrequency, + input.measurementId ?? null, + input.updatedAt, + ); +} + +async function createMeasurement( + id: string, + measuredAt: string, +): Promise { + const row = await getPrismaClient().measurement.create({ + data: { + id, + userId: USER_ID, + type: "IRREGULAR_RHYTHM_NOTIFICATION", + value: 1, + unit: "event", + measuredAt: new Date(measuredAt), + source: "APPLE_HEALTH", + }, + select: { id: true }, + }); + return row.id; +} + +beforeEach(async () => { + const prisma = getPrismaClient(); + await truncateAllTables(prisma); + await prisma.user.create({ + data: { + id: USER_ID, + username: "ecg-identity-migration", + email: "ecg-identity-migration@example.test", + timezone: "UTC", + }, + }); +}); + +afterEach(async () => { + // Whatever this test did to the schema, the next file gets the migrated one. + if (!(await indexExists())) applyMigration(); +}); + +describe("0291_ecg_recording_identity against a table that already has duplicates", () => { + it("resolves the duplicates, keeps the canonical row, and leaves the index enforcing", async () => { + const prisma = getPrismaClient(); + await dropIndex(); + expect(await indexExists()).toBe(false); + + const eventId = await createMeasurement( + "ecg-identity-event-1", + "2026-07-18T06:14:03.000Z", + ); + + // Partition 1 — the same recording under two source ids. The richer strip + // has no event link; the thin one does, and its link must survive. + await insertRecording({ + id: "p1-keeper-more-samples", + externalRecordingId: "apple-health:ecg:content-hash", + recordedAt: "2026-07-18T06:14:03Z", + samplingFrequency: 512, + sampleCount: 15_360, + updatedAt: "2026-07-18T07:00:00Z", + }); + await insertRecording({ + id: "p1-loser-fewer-samples", + externalRecordingId: "8E1F3B0C-0000-4000-8000-000000000001", + recordedAt: "2026-07-18T06:14:03Z", + samplingFrequency: 512, + sampleCount: 12, + measurementId: eventId, + updatedAt: "2026-07-18T09:00:00Z", + }); + + // Partition 2 — equal strips, one linked to its event row. + await insertRecording({ + id: "p2-unlinked", + externalRecordingId: "p2-a", + recordedAt: "2026-07-19T06:14:03Z", + samplingFrequency: 512, + sampleCount: 512, + updatedAt: "2026-07-19T09:00:00Z", + }); + const secondEventId = await createMeasurement( + "ecg-identity-event-2", + "2026-07-19T06:14:03.000Z", + ); + await insertRecording({ + id: "p2-linked", + externalRecordingId: "p2-b", + recordedAt: "2026-07-19T06:14:03Z", + samplingFrequency: 512, + sampleCount: 512, + measurementId: secondEventId, + updatedAt: "2026-07-19T07:00:00Z", + }); + + // Partition 3 — indistinguishable but for when they were last written. + await insertRecording({ + id: "p3-stale", + externalRecordingId: "p3-a", + recordedAt: "2026-07-20T06:14:03Z", + samplingFrequency: 512, + sampleCount: 512, + updatedAt: "2026-07-20T07:00:00Z", + }); + await insertRecording({ + id: "p3-fresh", + externalRecordingId: "p3-b", + recordedAt: "2026-07-20T06:14:03Z", + samplingFrequency: 512, + sampleCount: 512, + updatedAt: "2026-07-20T09:00:00Z", + }); + + // A recording that collides with nothing must be left exactly as it is. + await insertRecording({ + id: "p4-untouched", + externalRecordingId: "p4-a", + recordedAt: "2026-07-21T06:14:03Z", + samplingFrequency: 512, + sampleCount: 512, + updatedAt: "2026-07-21T07:00:00Z", + }); + + expect(await prisma.ecgRecording.count()).toBe(7); + + applyMigration(); + + expect(await indexExists()).toBe(true); + const survivors = await prisma.ecgRecording.findMany({ + orderBy: { recordedAt: "asc" }, + select: { id: true, measurementId: true, externalRecordingId: true }, + }); + expect(survivors.map((row) => row.id)).toEqual([ + "p1-keeper-more-samples", + "p2-linked", + "p3-fresh", + "p4-untouched", + ]); + // The loser's event link was salvaged onto the keeper. + expect(survivors[0].measurementId).toBe(eventId); + expect(survivors[1].measurementId).toBe(secondEventId); + expect(survivors[3].externalRecordingId).toBe("p4-a"); + + // The event rows themselves are untouched — resolving a duplicate + // recording never deletes a measurement. + expect(await prisma.measurement.count({ where: { userId: USER_ID } })).toBe( + 2, + ); + + // And the constraint is live: the collision cannot come back. + await expect( + insertRecording({ + id: "post-migration-twin", + externalRecordingId: "another-id-entirely", + recordedAt: "2026-07-21T06:14:03Z", + samplingFrequency: 512, + sampleCount: 512, + updatedAt: "2026-07-21T10:00:00Z", + }), + ).rejects.toThrow(); + expect(await prisma.ecgRecording.count()).toBe(4); + }); + + it("leaves rows alone that only look alike across users, sources or sampling rates", async () => { + const prisma = getPrismaClient(); + await dropIndex(); + const otherUser = await prisma.user.create({ + data: { + username: "ecg-identity-migration-other", + email: "ecg-identity-migration-other@example.test", + }, + }); + + await insertRecording({ + id: "same-instant-apple", + externalRecordingId: "a", + recordedAt: "2026-07-18T06:14:03Z", + samplingFrequency: 512, + sampleCount: 512, + updatedAt: "2026-07-18T07:00:00Z", + }); + // Same instant, different sampling rate — a different recording. + await insertRecording({ + id: "same-instant-other-rate", + externalRecordingId: "b", + recordedAt: "2026-07-18T06:14:03Z", + samplingFrequency: 300, + sampleCount: 300, + updatedAt: "2026-07-18T07:00:00Z", + }); + // Same instant and rate, but a different source. + await prisma.$executeRawUnsafe( + `INSERT INTO ecg_recordings + (id, user_id, source, external_recording_id, recorded_at, + waveform_encrypted, sampling_frequency, sample_count, created_at, updated_at) + VALUES ('same-instant-withings', $1, 'WITHINGS', 'c', + '2026-07-18T06:14:03Z'::timestamptz, $2, 512, 512, NOW(), NOW())`, + USER_ID, + Buffer.from("ciphertext-withings", "utf8"), + ); + // Same instant and rate, but somebody else's account. + await prisma.$executeRawUnsafe( + `INSERT INTO ecg_recordings + (id, user_id, source, external_recording_id, recorded_at, + waveform_encrypted, sampling_frequency, sample_count, created_at, updated_at) + VALUES ('same-instant-other-user', $1, 'APPLE_HEALTH', 'd', + '2026-07-18T06:14:03Z'::timestamptz, $2, 512, 512, NOW(), NOW())`, + otherUser.id, + Buffer.from("ciphertext-other-user", "utf8"), + ); + + applyMigration(); + + expect(await indexExists()).toBe(true); + const ids = ( + await prisma.ecgRecording.findMany({ select: { id: true } }) + ).map((row) => row.id); + expect(ids.sort()).toEqual( + [ + "same-instant-apple", + "same-instant-other-rate", + "same-instant-other-user", + "same-instant-withings", + ].sort(), + ); + }); +});