mirror of
https://github.com/MBombeck/HealthLog.git
synced 2026-11-08 04:32:42 +00:00
16 files changed
+1761
-111
No files matched your search
@@ -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
|
||||
|
||||
+191
-1
@@ -1,7 +1,7 @@
|
||||
openapi: 3.1.0
|
||||
info:
|
||||
title: HealthLog API
|
||||
version: 1.35.2
|
||||
version: 1.35.3
|
||||
description: >-
|
||||
Self-hosted personal-health-tracking PWA — public API surface for the iOS native client and external ingest.
|
||||
|
||||
@@ -7993,6 +7993,44 @@ paths:
|
||||
"403": *a5
|
||||
"422": *a3
|
||||
"429": *a4
|
||||
post:
|
||||
tags:
|
||||
- Insights
|
||||
summary: Ingest one ECG recording
|
||||
description: "The live ECG ingest: one Apple Watch recording per request, for a client draining its HealthKit ECG
|
||||
observer. Before this existed, a watch ECG could only reach HealthLog inside a full `export.zip`. One recording
|
||||
per request because a 30 s / 512 Hz strip is ~15 360 samples. Samples are INTEGER MICRO-VOLTS (convert from
|
||||
HealthKit's Volts), stored AES-256-GCM encrypted; `sampleCount` and `durationSeconds` are derived server-side.
|
||||
The `classification` is the device's own verdict stored verbatim — HealthLog never reads the waveform to produce
|
||||
or revise one. `source` accepts `APPLE_HEALTH` only; `userId` comes from the session and is never a body field.
|
||||
Unknown body keys are rejected with a 422 naming them. No `Idempotency-Key` is needed: the recording carries its
|
||||
own identity, so a retry resolves to the same row by construction — see `status` for what a re-post reports.
|
||||
Limits: 32 768 samples, 2 MB body, 60 recordings per minute per user. Module-gated on `insights` and the
|
||||
operator `insightStatus` assistant surface; no LLM call. Auth via cookie or Bearer."
|
||||
requestBody:
|
||||
required: true
|
||||
content:
|
||||
application/json:
|
||||
schema:
|
||||
$ref: "#/components/schemas/EcgIngestRequest"
|
||||
responses:
|
||||
"200":
|
||||
description: "The recording was already stored: `updated` (same id, overwritten in place) or `duplicate` (already
|
||||
present under a different id, nothing written)."
|
||||
content:
|
||||
application/json:
|
||||
schema:
|
||||
$ref: "#/components/schemas/EcgIngestResponseEnvelope"
|
||||
"201":
|
||||
description: "A new recording was stored (`status: inserted`)."
|
||||
content:
|
||||
application/json:
|
||||
schema:
|
||||
$ref: "#/components/schemas/EcgIngestCreatedEnvelope"
|
||||
"401": *a1
|
||||
"403": *a5
|
||||
"422": *a3
|
||||
"429": *a4
|
||||
/api/insights/ecg/{id}:
|
||||
get:
|
||||
tags:
|
||||
@@ -14481,6 +14519,81 @@ components:
|
||||
properties: {}
|
||||
description: No body fields. The user is taken from the session / Bearer and the locale from the session; the warm
|
||||
covers every assessment for that user.
|
||||
EcgIngestRequest:
|
||||
type: object
|
||||
properties:
|
||||
externalRecordingId:
|
||||
type: string
|
||||
minLength: 1
|
||||
maxLength: 120
|
||||
description: "The recording's stable id AS THE CLIENT KNOWS IT — for an Apple Health client, the `HKSample.uuid`. It is
|
||||
the source's own identity, not a content hash: re-posting the same id overwrites that recording in place. It
|
||||
does not have to agree with the id the `export.zip` importer derives for the same strip; the server
|
||||
reconciles the two doors itself (see `status: duplicate`)."
|
||||
recordedAt:
|
||||
type: string
|
||||
format: date-time
|
||||
pattern: ^(?:(?:\d\d[2468][048]|\d\d[13579][26]|\d\d0[48]|[02468][048]00|[13579][26]00)-02-29|\d{4}-(?:(?:0[13578]|1[02])-(?:0[1-9]|[12]\d|3[01])|(?:0[469]|11)-(?:0[1-9]|[12]\d|30)|(?:02)-(?:0[1-9]|1\d|2[0-8])))T(?:(?:[01]\d|2[0-3]):[0-5]\d(?::[0-5]\d(?:\.\d+)?)?(?:Z|([+-](?:[01]\d|2[0-3]):[0-5]\d)))$
|
||||
description: "When the strip was recorded on-device, with offset. Part of the recording's identity — see `status:
|
||||
duplicate`."
|
||||
samplingFrequency:
|
||||
type: integer
|
||||
minimum: 1
|
||||
maximum: 10000
|
||||
description: Signal sampling rate in Hz (an Apple Watch ECG is 512). Part of the recording's identity, and the divisor
|
||||
for the server-derived `durationSeconds`.
|
||||
samples:
|
||||
minItems: 1
|
||||
maxItems: 32768
|
||||
type: array
|
||||
items:
|
||||
type: integer
|
||||
minimum: -1000000
|
||||
maximum: 1000000
|
||||
description: The waveform as INTEGER MICRO-VOLTS, in recording order — the same unit the `export.zip` CSV parser
|
||||
produces, so convert from HealthKit's Volts before sending. Capped at 32 768 samples (a clean factor of two
|
||||
above a 30 s / 512 Hz strip). Stored AES-256-GCM encrypted; never persisted as plaintext.
|
||||
lead:
|
||||
description: Recording lead label when the device reports one (e.g. `I`).
|
||||
anyOf:
|
||||
- type: string
|
||||
minLength: 1
|
||||
maxLength: 40
|
||||
- type: "null"
|
||||
averageHeartRate:
|
||||
description: The device's average heart rate (BPM) for the strip.
|
||||
anyOf:
|
||||
- type: integer
|
||||
minimum: 1
|
||||
maximum: 300
|
||||
- type: "null"
|
||||
classification:
|
||||
description: The RECORDING DEVICE's own verdict, stored verbatim. Only the three ECG values are accepted here.
|
||||
`RhythmClassification` in the database has six members because walking-steadiness and neutral event verdicts
|
||||
share it, and those appear on GET /api/insights/rhythm-events — a decoder generated from this schema must
|
||||
NOT be reused there. HealthLog never reads the waveform to produce or revise a verdict.
|
||||
anyOf:
|
||||
- type: string
|
||||
enum:
|
||||
- IRREGULAR
|
||||
- NOT_DETECTED
|
||||
- INCONCLUSIVE
|
||||
- type: "null"
|
||||
source:
|
||||
type: string
|
||||
enum:
|
||||
- APPLE_HEALTH
|
||||
description: "The client's own source. `APPLE_HEALTH` only: `WITHINGS` rows are minted by the OAuth sync and `COMPUTED`
|
||||
ones by the server, so neither is client-assertable — the same posture POST /api/measurements/batch takes."
|
||||
required:
|
||||
- externalRecordingId
|
||||
- recordedAt
|
||||
- samplingFrequency
|
||||
- samples
|
||||
- source
|
||||
description: One Apple Watch ECG recording. Unknown keys are REJECTED with a 422 naming them rather than silently
|
||||
dropped, because every field here is load-bearing metadata for a stored recording. `sampleCount` and
|
||||
`durationSeconds` are derived server-side and must not be sent.
|
||||
UpdateCorrelationPatternRequest:
|
||||
type: object
|
||||
properties:
|
||||
@@ -29971,6 +30084,83 @@ components:
|
||||
- source
|
||||
- hasWaveform
|
||||
additionalProperties: false
|
||||
EcgIngestResponseEnvelope:
|
||||
type: object
|
||||
properties:
|
||||
data:
|
||||
$ref: "#/components/schemas/EcgIngestResponse"
|
||||
error:
|
||||
type: "null"
|
||||
meta:
|
||||
type: object
|
||||
properties:
|
||||
requestId:
|
||||
type: string
|
||||
additionalProperties: false
|
||||
required:
|
||||
- data
|
||||
- error
|
||||
additionalProperties: false
|
||||
EcgIngestResponse:
|
||||
type: object
|
||||
properties:
|
||||
id:
|
||||
type: string
|
||||
description: The recording row's id (cuid).
|
||||
status:
|
||||
type: string
|
||||
enum:
|
||||
- inserted
|
||||
- updated
|
||||
- duplicate
|
||||
description: What the write did. `inserted` — a new recording landed (HTTP 201). `updated` — this `externalRecordingId`
|
||||
was already stored and the row was overwritten in place (HTTP 200); a retry is safe and creates nothing.
|
||||
`duplicate` — this exact recording is already stored under a DIFFERENT id, i.e. it arrived earlier through
|
||||
the `export.zip` importer (HTTP 200); nothing was written and `id` names the row that already holds it. A
|
||||
recording is the same recording when `(source, recordedAt, samplingFrequency)` match, which is enforced by a
|
||||
unique index rather than by convention. All three statuses mean the recording is stored and the client may
|
||||
advance its cursor.
|
||||
recordedAt:
|
||||
type: string
|
||||
format: date-time
|
||||
pattern: ^(?:(?:\d\d[2468][048]|\d\d[13579][26]|\d\d0[48]|[02468][048]00|[13579][26]00)-02-29|\d{4}-(?:(?:0[13578]|1[02])-(?:0[1-9]|[12]\d|3[01])|(?:0[469]|11)-(?:0[1-9]|[12]\d|30)|(?:02)-(?:0[1-9]|1\d|2[0-8])))T(?:(?:[01]\d|2[0-3]):[0-5]\d(?::[0-5]\d(?:\.\d+)?)?(?:Z|([+-](?:[01]\d|2[0-3]):[0-5]\d)))$
|
||||
description: The `recordedAt` that was submitted, echoed back.
|
||||
sampleCount:
|
||||
type: integer
|
||||
minimum: -9007199254740991
|
||||
maximum: 9007199254740991
|
||||
description: "Server-derived: the number of samples received."
|
||||
durationSeconds:
|
||||
anyOf:
|
||||
- type: number
|
||||
- type: "null"
|
||||
description: "Server-derived: `sampleCount / samplingFrequency`."
|
||||
required:
|
||||
- id
|
||||
- status
|
||||
- recordedAt
|
||||
- sampleCount
|
||||
- durationSeconds
|
||||
additionalProperties: false
|
||||
description: The outcome of one ECG ingest. Reported honestly rather than optimistically — a re-post never claims to
|
||||
have inserted.
|
||||
EcgIngestCreatedEnvelope:
|
||||
type: object
|
||||
properties:
|
||||
data:
|
||||
$ref: "#/components/schemas/EcgIngestResponse"
|
||||
error:
|
||||
type: "null"
|
||||
meta:
|
||||
type: object
|
||||
properties:
|
||||
requestId:
|
||||
type: string
|
||||
additionalProperties: false
|
||||
required:
|
||||
- data
|
||||
- error
|
||||
additionalProperties: false
|
||||
EcgDetailResponseEnvelope:
|
||||
type: object
|
||||
properties:
|
||||
|
||||
+1
-1
@@ -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",
|
||||
|
||||
@@ -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");
|
||||
@@ -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")
|
||||
}
|
||||
|
||||
+1
-1
@@ -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/*`
|
||||
|
||||
@@ -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<unknown>(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,
|
||||
);
|
||||
});
|
||||
@@ -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";
|
||||
}
|
||||
@@ -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<string, unknown> = {}) {
|
||||
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]);
|
||||
});
|
||||
});
|
||||
@@ -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<PrismaClient, "$transaction">;
|
||||
|
||||
export async function persistEcgRecording(
|
||||
input: EcgRecordingWrite,
|
||||
db: TransactionalClient,
|
||||
): Promise<EcgPersistResult> {
|
||||
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);
|
||||
}
|
||||
}
|
||||
@@ -37,6 +37,8 @@ import {
|
||||
insightsPregenerateResponse,
|
||||
dashboardSnapshotResponse,
|
||||
ecgListResponse,
|
||||
ecgIngestRequest,
|
||||
ecgIngestResponse,
|
||||
ecgDetailQuery,
|
||||
ecgDetailResponse,
|
||||
rhythmEventsResponse,
|
||||
@@ -457,6 +459,45 @@ export const insightsPaths: NonNullable<ZodOpenApiObject["paths"]> = {
|
||||
...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: {
|
||||
|
||||
@@ -1439,6 +1439,96 @@ export const ecgDetailResponse = z
|
||||
"One ECG recording's decrypted waveform plus metadata and the DEVICE's verbatim classification. Ownership is narrowed in the query where — a foreign / unknown id 404s. The waveform is AES-256-GCM at rest, decrypted through the fail-closed codec. HealthLog does not interpret the trace, measure intervals, annotate beats, or emit a verdict of its own. no-store.",
|
||||
});
|
||||
|
||||
// The live ECG ingest. One recording per request: a 30 s Apple Watch strip at
|
||||
// 512 Hz is ~15 360 samples, so a batch of them is not a shape anyone wants on
|
||||
// the wire. `userId` is never a body field — it comes from the session.
|
||||
export const ecgIngestRequest = z
|
||||
.object({
|
||||
externalRecordingId: z
|
||||
.string()
|
||||
.min(1)
|
||||
.max(120)
|
||||
.describe(
|
||||
"The recording's stable id AS THE CLIENT KNOWS IT — for an Apple Health client, the `HKSample.uuid`. It is the source's own identity, not a content hash: re-posting the same id overwrites that recording in place. It does not have to agree with the id the `export.zip` importer derives for the same strip; the server reconciles the two doors itself (see `status: duplicate`).",
|
||||
),
|
||||
recordedAt: z.iso
|
||||
.datetime({ offset: true })
|
||||
.describe(
|
||||
"When the strip was recorded on-device, with offset. Part of the recording's identity — see `status: duplicate`.",
|
||||
),
|
||||
samplingFrequency: z
|
||||
.number()
|
||||
.int()
|
||||
.min(1)
|
||||
.max(10_000)
|
||||
.describe(
|
||||
"Signal sampling rate in Hz (an Apple Watch ECG is 512). Part of the recording's identity, and the divisor for the server-derived `durationSeconds`.",
|
||||
),
|
||||
samples: z
|
||||
.array(z.number().int().min(-1_000_000).max(1_000_000))
|
||||
.min(1)
|
||||
.max(32_768)
|
||||
.describe(
|
||||
"The waveform as INTEGER MICRO-VOLTS, in recording order — the same unit the `export.zip` CSV parser produces, so convert from HealthKit's Volts before sending. Capped at 32 768 samples (a clean factor of two above a 30 s / 512 Hz strip). Stored AES-256-GCM encrypted; never persisted as plaintext.",
|
||||
),
|
||||
lead: z
|
||||
.string()
|
||||
.min(1)
|
||||
.max(40)
|
||||
.nullable()
|
||||
.optional()
|
||||
.describe("Recording lead label when the device reports one (e.g. `I`)."),
|
||||
averageHeartRate: z
|
||||
.number()
|
||||
.int()
|
||||
.min(1)
|
||||
.max(300)
|
||||
.nullable()
|
||||
.optional()
|
||||
.describe("The device's average heart rate (BPM) for the strip."),
|
||||
classification: ecgClassification
|
||||
.optional()
|
||||
.describe(
|
||||
"The RECORDING DEVICE's own verdict, stored verbatim. Only the three ECG values are accepted here. `RhythmClassification` in the database has six members because walking-steadiness and neutral event verdicts share it, and those appear on GET /api/insights/rhythm-events — a decoder generated from this schema must NOT be reused there. HealthLog never reads the waveform to produce or revise a verdict.",
|
||||
),
|
||||
source: z
|
||||
.enum(["APPLE_HEALTH"])
|
||||
.describe(
|
||||
"The client's own source. `APPLE_HEALTH` only: `WITHINGS` rows are minted by the OAuth sync and `COMPUTED` ones by the server, so neither is client-assertable — the same posture POST /api/measurements/batch takes.",
|
||||
),
|
||||
})
|
||||
.meta({
|
||||
id: "EcgIngestRequest",
|
||||
description:
|
||||
"One Apple Watch ECG recording. Unknown keys are REJECTED with a 422 naming them rather than silently dropped, because every field here is load-bearing metadata for a stored recording. `sampleCount` and `durationSeconds` are derived server-side and must not be sent.",
|
||||
});
|
||||
|
||||
export const ecgIngestResponse = z
|
||||
.object({
|
||||
id: z.string().describe("The recording row's id (cuid)."),
|
||||
status: z
|
||||
.enum(["inserted", "updated", "duplicate"])
|
||||
.describe(
|
||||
"What the write did. `inserted` — a new recording landed (HTTP 201). `updated` — this `externalRecordingId` was already stored and the row was overwritten in place (HTTP 200); a retry is safe and creates nothing. `duplicate` — this exact recording is already stored under a DIFFERENT id, i.e. it arrived earlier through the `export.zip` importer (HTTP 200); nothing was written and `id` names the row that already holds it. A recording is the same recording when `(source, recordedAt, samplingFrequency)` match, which is enforced by a unique index rather than by convention. All three statuses mean the recording is stored and the client may advance its cursor.",
|
||||
),
|
||||
recordedAt: z.iso
|
||||
.datetime({ offset: true })
|
||||
.describe("The `recordedAt` that was submitted, echoed back."),
|
||||
sampleCount: z
|
||||
.number()
|
||||
.int()
|
||||
.describe("Server-derived: the number of samples received."),
|
||||
durationSeconds: z
|
||||
.number()
|
||||
.nullable()
|
||||
.describe("Server-derived: `sampleCount / samplingFrequency`."),
|
||||
})
|
||||
.meta({
|
||||
id: "EcgIngestResponse",
|
||||
description:
|
||||
"The outcome of one ECG ingest. Reported honestly rather than optimistically — a re-post never claims to have inserted.",
|
||||
});
|
||||
|
||||
// v1.10.0 — device-flagged event timeline (rhythm-events, WX-B). Deliberately
|
||||
// a SEPARATE enum from `ecgClassification` above: an EVENT row's
|
||||
// `rhythmClassification` carries the FULL RhythmClassification set (the
|
||||
|
||||
@@ -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(() => {
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
@@ -0,0 +1,368 @@
|
||||
/**
|
||||
* `POST /api/insights/ecg` end to end against real Postgres.
|
||||
*
|
||||
* Everything here goes through the actual route handlers — the ingest, the
|
||||
* metadata list, and the waveform detail — because the thing worth proving is
|
||||
* the assembly between them, not that each end works in isolation. What a
|
||||
* client receives is read out of the real GET responses, never rebuilt from a
|
||||
* fixture.
|
||||
*
|
||||
* The cross-door case is the point of the second unique index: the same
|
||||
* physical recording reaching the export-archive importer and the live route
|
||||
* carries two different source ids, and must still be one row.
|
||||
*/
|
||||
import { NextRequest } from "next/server";
|
||||
import { beforeEach, describe, expect, it, vi } from "vitest";
|
||||
|
||||
import { cookieJar, headerJar } from "./mock-next-headers";
|
||||
import { getPrismaClient, truncateAllTables } from "./setup";
|
||||
|
||||
process.env.ENCRYPTION_KEY ??=
|
||||
"0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef";
|
||||
|
||||
vi.mock("next/headers", async () => {
|
||||
const { cookieJar, headerJar } = await import("./mock-next-headers");
|
||||
return {
|
||||
headers: vi.fn(async () => ({
|
||||
get: (name: string) => headerJar.get(name.toLowerCase()) ?? null,
|
||||
})),
|
||||
cookies: vi.fn(async () => ({
|
||||
get: (name: string) => {
|
||||
const value = cookieJar.get(name);
|
||||
return value ? { name, value } : undefined;
|
||||
},
|
||||
set: (name: string, value: string) => {
|
||||
cookieJar.set(name, value);
|
||||
},
|
||||
delete: (name: string) => {
|
||||
cookieJar.delete(name);
|
||||
},
|
||||
})),
|
||||
};
|
||||
});
|
||||
|
||||
vi.mock("@/lib/db-compat", () => ({
|
||||
ensureDbCompatibility: vi.fn().mockResolvedValue(undefined),
|
||||
}));
|
||||
|
||||
import {
|
||||
GET as getEcgList,
|
||||
POST as postEcg,
|
||||
} from "@/app/api/insights/ecg/route";
|
||||
import { GET as getEcgDetail } from "@/app/api/insights/ecg/[id]/route";
|
||||
import { importAppleHealthEcg } from "@/lib/apple-health/ecg-import";
|
||||
|
||||
const USER_ID = "ecg-ingest-roundtrip";
|
||||
const OTHER_USER_ID = "ecg-ingest-roundtrip-other";
|
||||
const RECORDED_AT = "2026-07-18T08:14:03+02:00";
|
||||
const SAMPLES = [12, -7, 3, 240, -180, 60];
|
||||
|
||||
type IngestBody = Record<string, unknown>;
|
||||
|
||||
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<Record<string, unknown>>;
|
||||
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<Response>;
|
||||
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<string, unknown> }> {
|
||||
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<void> {
|
||||
const session = await getPrismaClient().session.create({
|
||||
data: { userId, expiresAt: new Date(Date.now() + 60 * 60 * 1000) },
|
||||
});
|
||||
cookieJar.set("healthlog_session", session.id);
|
||||
}
|
||||
|
||||
beforeEach(async () => {
|
||||
const prisma = getPrismaClient();
|
||||
await truncateAllTables(prisma);
|
||||
cookieJar.clear();
|
||||
headerJar.clear();
|
||||
await prisma.user.createMany({
|
||||
data: [
|
||||
{
|
||||
id: USER_ID,
|
||||
username: "ecg-roundtrip",
|
||||
email: "ecg-roundtrip@example.test",
|
||||
timezone: "Europe/Berlin",
|
||||
},
|
||||
{
|
||||
id: OTHER_USER_ID,
|
||||
username: "ecg-roundtrip-other",
|
||||
email: "ecg-roundtrip-other@example.test",
|
||||
timezone: "Europe/Berlin",
|
||||
},
|
||||
],
|
||||
});
|
||||
await signIn(USER_ID);
|
||||
});
|
||||
|
||||
describe("POST /api/insights/ecg — round trip through the real routes", () => {
|
||||
it("stores a posted recording and serves it back through the list and the detail", async () => {
|
||||
const posted = await ingest(ingestPayload());
|
||||
expect(posted.status).toBe(201);
|
||||
expect(posted.data.status).toBe("inserted");
|
||||
expect(posted.data.sampleCount).toBe(SAMPLES.length);
|
||||
expect(posted.data.durationSeconds).toBeCloseTo(SAMPLES.length / 512, 10);
|
||||
|
||||
const list = await readList();
|
||||
expect(list.hasRecordings).toBe(true);
|
||||
expect(list.recordings).toHaveLength(1);
|
||||
expect(list.recordings[0]).toMatchObject({
|
||||
id: posted.data.id,
|
||||
samplingFrequency: 512,
|
||||
sampleCount: SAMPLES.length,
|
||||
averageHeartRate: 64,
|
||||
lead: "I",
|
||||
classification: "NOT_DETECTED",
|
||||
source: "APPLE_HEALTH",
|
||||
hasWaveform: true,
|
||||
});
|
||||
expect(
|
||||
new Date(list.recordings[0].recordedAt as string).toISOString(),
|
||||
).toBe(new Date(RECORDED_AT).toISOString());
|
||||
|
||||
const detail = await readDetail(posted.data.id, true);
|
||||
expect(detail.response.status).toBe(200);
|
||||
expect(detail.body.samples).toEqual(SAMPLES);
|
||||
expect(detail.body.decimated).toBe(false);
|
||||
expect(detail.body.classification).toBe("NOT_DETECTED");
|
||||
});
|
||||
|
||||
it("never writes the waveform as plaintext", async () => {
|
||||
const posted = await ingest(ingestPayload());
|
||||
const row = await getPrismaClient().ecgRecording.findUniqueOrThrow({
|
||||
where: { id: posted.data.id },
|
||||
});
|
||||
const stored = Buffer.from(row.waveformEncrypted).toString("utf8");
|
||||
// The column holds the `<keyId>.<base64>` envelope and nothing else. A
|
||||
// serialised sample array cannot satisfy that shape — it carries
|
||||
// brackets, commas and minus signs — so this rules plaintext out without
|
||||
// asking whether some digit happens to appear in a base64 blob, which it
|
||||
// routinely does.
|
||||
expect(stored).toMatch(/^[A-Za-z0-9_-]+\.[A-Za-z0-9+/=]+$/);
|
||||
expect(stored).not.toContain(JSON.stringify(SAMPLES));
|
||||
});
|
||||
|
||||
it("reports a re-post as updated and does not grow the table", async () => {
|
||||
const first = await ingest(ingestPayload());
|
||||
const second = await ingest(ingestPayload());
|
||||
|
||||
expect(second.status).toBe(200);
|
||||
expect(second.data.status).toBe("updated");
|
||||
expect(second.data.id).toBe(first.data.id);
|
||||
expect(
|
||||
await getPrismaClient().ecgRecording.count({
|
||||
where: { userId: USER_ID },
|
||||
}),
|
||||
).toBe(1);
|
||||
expect((await readList()).recordings).toHaveLength(1);
|
||||
});
|
||||
|
||||
it("overwrites the stored recording in place when a re-post revises it", async () => {
|
||||
const first = await ingest(ingestPayload());
|
||||
const revised = [1, 2, 3, 4];
|
||||
const second = await ingest(
|
||||
ingestPayload({ samples: revised, classification: "IRREGULAR" }),
|
||||
);
|
||||
|
||||
expect(second.data.status).toBe("updated");
|
||||
expect(second.data.id).toBe(first.data.id);
|
||||
const detail = await readDetail(first.data.id, true);
|
||||
expect(detail.body.samples).toEqual(revised);
|
||||
expect(detail.body.classification).toBe("IRREGULAR");
|
||||
});
|
||||
|
||||
it("keeps one row when the same recording also arrives through the export archive", async () => {
|
||||
const posted = await ingest(ingestPayload());
|
||||
|
||||
// The archive importer identifies the strip by a hash of its content, so
|
||||
// it arrives under a completely different externalRecordingId.
|
||||
const outcome = await importAppleHealthEcg({
|
||||
userId: USER_ID,
|
||||
ecg: {
|
||||
recordedAt: new Date(RECORDED_AT),
|
||||
samplingFrequency: 512,
|
||||
samples: [...SAMPLES],
|
||||
lead: "I",
|
||||
averageHeartRate: 64,
|
||||
rhythmClassification: "NOT_DETECTED",
|
||||
},
|
||||
prisma: getPrismaClient(),
|
||||
});
|
||||
|
||||
expect(outcome).toBe("skipped");
|
||||
const rows = await getPrismaClient().ecgRecording.findMany({
|
||||
where: { userId: USER_ID },
|
||||
});
|
||||
expect(rows).toHaveLength(1);
|
||||
expect(rows[0].id).toBe(posted.data.id);
|
||||
expect((await readList()).recordings).toHaveLength(1);
|
||||
});
|
||||
|
||||
it("reports duplicate when the export archive got there first", async () => {
|
||||
await importAppleHealthEcg({
|
||||
userId: USER_ID,
|
||||
ecg: {
|
||||
recordedAt: new Date(RECORDED_AT),
|
||||
samplingFrequency: 512,
|
||||
samples: [...SAMPLES],
|
||||
lead: "I",
|
||||
averageHeartRate: 64,
|
||||
rhythmClassification: "NOT_DETECTED",
|
||||
},
|
||||
prisma: getPrismaClient(),
|
||||
});
|
||||
const archived = await getPrismaClient().ecgRecording.findFirstOrThrow({
|
||||
where: { userId: USER_ID },
|
||||
});
|
||||
|
||||
const posted = await ingest(ingestPayload());
|
||||
expect(posted.status).toBe(200);
|
||||
expect(posted.data.status).toBe("duplicate");
|
||||
expect(posted.data.id).toBe(archived.id);
|
||||
expect(
|
||||
await getPrismaClient().ecgRecording.count({
|
||||
where: { userId: USER_ID },
|
||||
}),
|
||||
).toBe(1);
|
||||
// Nothing was written: the archive's own external id still owns the row.
|
||||
expect(
|
||||
(
|
||||
await getPrismaClient().ecgRecording.findUniqueOrThrow({
|
||||
where: { id: archived.id },
|
||||
})
|
||||
).externalRecordingId,
|
||||
).toBe(archived.externalRecordingId);
|
||||
});
|
||||
|
||||
it("stores distinct recordings distinctly", async () => {
|
||||
await ingest(ingestPayload());
|
||||
const other = await ingest(
|
||||
ingestPayload({
|
||||
externalRecordingId: "8E1F3B0C-0000-4000-8000-000000000002",
|
||||
recordedAt: "2026-07-19T08:14:03+02:00",
|
||||
}),
|
||||
);
|
||||
expect(other.data.status).toBe("inserted");
|
||||
expect((await readList()).recordings).toHaveLength(2);
|
||||
});
|
||||
|
||||
it("writes to the session's user and cannot reach another account's rows", async () => {
|
||||
const mine = await ingest(ingestPayload());
|
||||
|
||||
cookieJar.clear();
|
||||
await signIn(OTHER_USER_ID);
|
||||
// Same recording id, same instant — a different account entirely.
|
||||
const theirs = await ingest(ingestPayload());
|
||||
expect(theirs.status).toBe(201);
|
||||
expect(theirs.data.status).toBe("inserted");
|
||||
expect(theirs.data.id).not.toBe(mine.data.id);
|
||||
|
||||
expect((await readList()).recordings).toHaveLength(1);
|
||||
const foreign = await readDetail(mine.data.id, true);
|
||||
expect(foreign.response.status).toBe(404);
|
||||
|
||||
const owners = await getPrismaClient().ecgRecording.findMany({
|
||||
select: { userId: true },
|
||||
});
|
||||
expect(owners.map((r) => r.userId).sort()).toEqual(
|
||||
[USER_ID, OTHER_USER_ID].sort(),
|
||||
);
|
||||
});
|
||||
|
||||
it("refuses a source a client is not allowed to claim", async () => {
|
||||
const response = await postEcg(
|
||||
new NextRequest("http://localhost/api/insights/ecg", {
|
||||
method: "POST",
|
||||
headers: { "content-type": "application/json" },
|
||||
body: JSON.stringify(ingestPayload({ source: "WITHINGS" })),
|
||||
}),
|
||||
);
|
||||
expect(response.status).toBe(422);
|
||||
expect(await getPrismaClient().ecgRecording.count()).toBe(0);
|
||||
});
|
||||
|
||||
it("refuses a userId in the body rather than honouring it", async () => {
|
||||
const response = await postEcg(
|
||||
new NextRequest("http://localhost/api/insights/ecg", {
|
||||
method: "POST",
|
||||
headers: { "content-type": "application/json" },
|
||||
body: JSON.stringify(ingestPayload({ userId: OTHER_USER_ID })),
|
||||
}),
|
||||
);
|
||||
expect(response.status).toBe(422);
|
||||
expect(await getPrismaClient().ecgRecording.count()).toBe(0);
|
||||
});
|
||||
|
||||
it("refuses an event verdict that cannot occur on an ECG", async () => {
|
||||
const response = await postEcg(
|
||||
new NextRequest("http://localhost/api/insights/ecg", {
|
||||
method: "POST",
|
||||
headers: { "content-type": "application/json" },
|
||||
body: JSON.stringify(ingestPayload({ classification: "VERY_LOW" })),
|
||||
}),
|
||||
);
|
||||
expect(response.status).toBe(422);
|
||||
expect(await getPrismaClient().ecgRecording.count()).toBe(0);
|
||||
});
|
||||
});
|
||||
@@ -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<void> {
|
||||
await getPrismaClient().$executeRawUnsafe(
|
||||
`DROP INDEX IF EXISTS "${INDEX_NAME}"`,
|
||||
);
|
||||
}
|
||||
|
||||
async function indexExists(): Promise<boolean> {
|
||||
const rows = await getPrismaClient().$queryRaw<Array<{ indexname: string }>>`
|
||||
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<void> {
|
||||
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<string> {
|
||||
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(),
|
||||
);
|
||||
});
|
||||
});
|
||||
Reference in new issue
Block a user