From 0e301b580cb4675715950490f5b79da194d2694a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Sat, 1 Aug 2026 19:19:01 +0200 Subject: [PATCH] Let an Apple Watch ECG arrive without a 1.5 GB export archive POST /api/insights/ecg takes one recording per request: the stable recording id the client already has, the on-device timestamp, the sampling rate, the waveform as integer micro-volts, the lead, the average heart rate, and the device's own verdict. Same gating as the reads (auth, the insights module, the insightStatus surface), userId from the session, APPLE_HEALTH the only source a client may claim. sampleCount and durationSeconds are derived here, and nothing about the waveform is ever interpreted. The write body now lives in one place, src/lib/ecg/persist-recording.ts, shared by the Withings sync, the export-archive importer and the new route. Three writers with three persistence bodies is how the two ends of one rule drift apart. Migration 0291 adds the second unique index on (user_id, source, recorded_at, sampling_frequency). The archive importer identifies a strip by a hash of its content and a live client by the device's own sample uuid, so the same physical recording arriving through both doors carried two different ids and would have landed twice. One device does not record two different strips at the same instant at the same sampling rate, so the constraint is sound and the convergence stops being a convention. The table already holds rows on deployed instances and two writers have been filling it since v1.19.0, so the index cannot just be created: a pre-existing collision would abort the migration. The migration resolves duplicates first, keeping the row with the most samples (a stored strip beats a verdict-only row), then the one linked to its event measurement, then the most recently written, then the lowest id so the choice is total. The losers' event link is salvaged onto the keeper first, so resolving a duplicate never costs the recording its place in the event timeline. If that resolution is wrong on some instance, the failure is visible at deploy time and not later: db:migrate:deploy runs on boot, so the container crash-loops with the migration named in the logs and the release never serves traffic. A deploy that finishes in under thirty seconds did not run it. The GET list and detail routes and the export-archive path are unchanged; this is a second door, not a replacement. --- docs/api/openapi.yaml | 190 ++++++++++++++++ .../0291_ecg_recording_identity/migration.sql | 90 ++++++++ prisma/schema.prisma | 8 + src/app/api/insights/ecg/route.ts | 173 +++++++++++++- src/lib/apple-health/ecg-import.ts | 87 +++---- src/lib/ecg/persist-recording.ts | 213 ++++++++++++++++++ src/lib/openapi/routes/insights/paths.ts | 41 ++++ src/lib/openapi/routes/insights/schemas.ts | 90 ++++++++ src/lib/withings/sync-ecg.ts | 46 +--- 9 files changed, 843 insertions(+), 95 deletions(-) create mode 100644 prisma/migrations/0291_ecg_recording_identity/migration.sql create mode 100644 src/lib/ecg/persist-recording.ts diff --git a/docs/api/openapi.yaml b/docs/api/openapi.yaml index 1634c89f7..a4e2ed30b 100644 --- a/docs/api/openapi.yaml +++ b/docs/api/openapi.yaml @@ -7888,6 +7888,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: @@ -14329,6 +14367,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: @@ -29594,6 +29707,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/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/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/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 bd7aae6f1..90f547846 100644 --- a/src/lib/openapi/routes/insights/schemas.ts +++ b/src/lib/openapi/routes/insights/schemas.ts @@ -1422,6 +1422,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/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; }