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 1/4] 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; } From 8f90fd31746c643bad3da4d8a46f1ea974827747 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Sat, 1 Aug 2026 19:26:47 +0200 Subject: [PATCH 2/4] Prove the ECG ingest, the dedupe and the migration on dirty data A round trip through the real handlers: post a recording, read it back through the list and the detail routes, and check what a client actually receives rather than a fixture rebuilt at each end. Post it twice and the second call says updated, not inserted, and the table does not grow. Feed the same strip to the export-archive importer afterwards and it says skipped; feed it to the importer first and the route says duplicate. Both directions leave one row. The migration test drops the index, manufactures the collisions the resolution step exists for, and runs the migration file through the same executor the deploy uses. It checks which row survived each tiebreak, that the salvaged event link followed the keeper, that rows differing only by user, source or sampling rate were left alone, and that the constraint rejects a new twin afterwards. Applying the migration to the empty testcontainer proves none of that. The Withings ECG tests now hand the transaction the same delegate the shared writer uses, so they still watch the real write instead of a stub standing in for it. --- .../ecg/__tests__/persist-recording.test.ts | 163 ++++++++ src/lib/withings/__tests__/sync-ecg.test.ts | 45 ++- .../integration/ecg-ingest-roundtrip.test.ts | 365 ++++++++++++++++++ .../ecg-recording-identity-migration.test.ts | 322 +++++++++++++++ 4 files changed, 882 insertions(+), 13 deletions(-) create mode 100644 src/lib/ecg/__tests__/persist-recording.test.ts create mode 100644 tests/integration/ecg-ingest-roundtrip.test.ts create mode 100644 tests/integration/ecg-recording-identity-migration.test.ts 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/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/tests/integration/ecg-ingest-roundtrip.test.ts b/tests/integration/ecg-ingest-roundtrip.test.ts new file mode 100644 index 000000000..bb2d6cb38 --- /dev/null +++ b/tests/integration/ecg-ingest-roundtrip.test.ts @@ -0,0 +1,365 @@ +/** + * `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"); + for (const sample of SAMPLES) { + expect(stored).not.toContain(String(sample)); + } + expect(stored).not.toContain("[12,-7,3"); + }); + + 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(), + ); + }); +}); From 42e16a6e120b6e5d49a6b0d77c0df924b527ba8c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Sat, 1 Aug 2026 19:35:16 +0200 Subject: [PATCH 3/4] Stop the plaintext-waveform check passing on a coincidence It asserted the stored ciphertext does not contain each sample rendered as a string. Base64 is full of digits, so a short sample value like 3 matches by accident: the check failed on a mutation that had nothing to do with encryption, and would equally have passed while leaking. It now asserts the column holds the keyId.base64 envelope and nothing else, which a serialised sample array cannot satisfy. --- tests/integration/ecg-ingest-roundtrip.test.ts | 11 +++++++---- 1 file changed, 7 insertions(+), 4 deletions(-) diff --git a/tests/integration/ecg-ingest-roundtrip.test.ts b/tests/integration/ecg-ingest-roundtrip.test.ts index bb2d6cb38..94a104947 100644 --- a/tests/integration/ecg-ingest-roundtrip.test.ts +++ b/tests/integration/ecg-ingest-roundtrip.test.ts @@ -194,10 +194,13 @@ describe("POST /api/insights/ecg — round trip through the real routes", () => where: { id: posted.data.id }, }); const stored = Buffer.from(row.waveformEncrypted).toString("utf8"); - for (const sample of SAMPLES) { - expect(stored).not.toContain(String(sample)); - } - expect(stored).not.toContain("[12,-7,3"); + // 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 () => { From ffe5545032735a1b9d15154cb56adde1d82c03bc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Sat, 1 Aug 2026 19:54:08 +0200 Subject: [PATCH 4/4] =?UTF-8?q?chore(release):=20v1.35.3=20=E2=80=94=20an?= =?UTF-8?q?=20ECG=20arrives=20on=20its=20own,=20and=20only=20once?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Apple Watch ECG recordings reach the record straight from the app instead of only through an export archive. One recording per request, gated by the same switches that already govern reading them. A recording is now identified by what it is rather than by what a source calls it, so the same strip arriving through both doors converges on one row. The migration that adds the constraint resolves the duplicates already stored: it keeps the row holding the waveform over one holding only a verdict, and moves a removed row's event link onto the row that is kept. The losing rows are deleted. --- CHANGELOG.md | 30 ++++++++++++++++++++++++++++++ docs/api/openapi.yaml | 2 +- package.json | 2 +- public/sw.js | 2 +- 4 files changed, 33 insertions(+), 3 deletions(-) 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 17a7f0cc2..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. 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/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/*`