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; }