diff --git a/src/lib/apple-health/ecg-import.ts b/src/lib/apple-health/ecg-import.ts new file mode 100644 index 000000000..c48096758 --- /dev/null +++ b/src/lib/apple-health/ecg-import.ts @@ -0,0 +1,93 @@ +/** + * Transactional persistence for one normalized Apple Health ECG recording. + * + * 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 + * across tenants. + */ +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"; + +export type AppleHealthEcgImportOutcome = "imported" | "updated" | "skipped"; + +function recordingIdentity(ecg: NormalizedAppleHealthEcg): string { + const hash = createHash("sha256"); + hash.update("healthlog:apple-health-ecg:v1\0"); + hash.update(ecg.recordedAt.toISOString()); + hash.update(`\0${ecg.samplingFrequency}\0${ecg.lead ?? ""}`); + hash.update( + `\0${ecg.averageHeartRate ?? ""}\0${ecg.rhythmClassification ?? ""}\0`, + ); + const sample = Buffer.allocUnsafe(8); + for (const value of ecg.samples) { + sample.writeDoubleLE(value, 0); + hash.update(sample); + } + sample.fill(0); + return `apple-health:ecg:${hash.digest("hex")}`; +} + +export async function importAppleHealthEcg(input: { + userId: string; + ecg: NormalizedAppleHealthEcg; + prisma: Pick; +}): Promise { + 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; + + 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); + } +} diff --git a/src/lib/jobs/apple-health-import-worker.ts b/src/lib/jobs/apple-health-import-worker.ts index b71dfa0ab..4a492a9aa 100644 --- a/src/lib/jobs/apple-health-import-worker.ts +++ b/src/lib/jobs/apple-health-import-worker.ts @@ -29,6 +29,9 @@ import type { PrismaClient } from "@/generated/prisma/client"; import { prisma, toJson } from "@/lib/db"; import type { Job } from "pg-boss"; +import { streamAppleHealthEcgMembers } from "@/lib/apple-health/archive-stream"; +import { parseAppleHealthEcgCsv } from "@/lib/apple-health/ecg-csv"; +import { importAppleHealthEcg } from "@/lib/apple-health/ecg-import"; import { extractExportXml } from "@/lib/import/unzip-export-xml"; import { getGlobalBoss } from "@/lib/jobs/boss-instance"; import { jobDone, jobFailed, type JobOutcome } from "@/lib/jobs/job-outcome"; @@ -98,7 +101,16 @@ export const APPLE_HEALTH_IMPORT_V2_QUEUE = "apple-health-import-v2"; export const APPLE_HEALTH_IMPORT_LEGACY_QUEUE = "apple-health-import"; /** Parser semantics carried by every newly-created ImportJob. */ -export const APPLE_HEALTH_IMPORT_PARSER_REVISION = 2; +export const APPLE_HEALTH_IMPORT_PARSER_REVISION = 3; + +const APPLE_HEALTH_ECG_ARCHIVE_LIMITS = { + maxMembers: 20_000, + maxEcgMembers: 2_000, + maxMemberBytes: 16 * 1024 * 1024, + maxTotalEcgBytes: 512 * 1024 * 1024, + maxCompressionRatio: 200, +} as const; +const APPLE_HEALTH_ECG_MAX_SAMPLES = 2_000_000; /** Concurrency cap per worker host. */ export const APPLE_HEALTH_IMPORT_CONCURRENCY = 1; @@ -326,23 +338,66 @@ export async function handleAppleHealthImport( elapsedMs: 0, }); - const result: ImportJobResult = await streamParseExportXml({ - xmlPath: unzip.xmlPath, - userId, - userTimezone, - prisma, - onProgress: async (snapshot) => { - // The phase label here is what the polling endpoint surfaces; - // map "parsing" / "upserting" through verbatim. - await prisma.importJob.update({ - where: { id: importJobId }, - data: { - status: snapshot.currentPhase, - progress: toJson(snapshot), - }, - }); + const result: ImportJobResult = { + ...(await streamParseExportXml({ + xmlPath: unzip.xmlPath, + userId, + userTimezone, + prisma, + onProgress: async (snapshot) => { + // The phase label here is what the polling endpoint surfaces; + // map "parsing" / "upserting" through verbatim. + await prisma.importJob.update({ + where: { id: importJobId }, + data: { + status: snapshot.currentPhase, + progress: toJson(snapshot), + }, + }); + }, + })), + ecg: { + discovered: 0, + imported: 0, + updated: 0, + skipped: 0, + failed: 0, }, - }); + }; + + // Auxiliary ECGs are intentionally fail-soft. The canonical export.xml + // transaction has already completed; malformed, oversized, or unwritable + // ECG members affect only bounded scalar counters in the terminal result. + const ecg = result.ecg; + ecg.discovered = unzip.otherMembers.filter((member) => { + const lower = member.name.toLowerCase(); + return lower.includes("electrocardiograms/") && lower.endsWith(".csv"); + }).length; + try { + for await (const member of streamAppleHealthEcgMembers({ + archivePath: uploadPath, + limits: APPLE_HEALTH_ECG_ARCHIVE_LIMITS, + })) { + try { + const parsed = await parseAppleHealthEcgCsv({ + memberName: member.name, + stream: member.stream, + maxSamples: APPLE_HEALTH_ECG_MAX_SAMPLES, + }); + const outcome = await importAppleHealthEcg({ + userId, + ecg: parsed, + prisma, + }); + ecg[outcome] += 1; + } catch { + ecg.failed += 1; + } + } + } catch { + const completed = ecg.imported + ecg.updated + ecg.skipped + ecg.failed; + ecg.failed += Math.max(1, ecg.discovered - completed); + } // Done. Persist the terminal envelope. await prisma.importJob.update({ diff --git a/src/lib/measurements/import-apple-health-export.ts b/src/lib/measurements/import-apple-health-export.ts index 320e9c46f..fc1e9f838 100644 --- a/src/lib/measurements/import-apple-health-export.ts +++ b/src/lib/measurements/import-apple-health-export.ts @@ -120,6 +120,14 @@ export interface ImportJobResult { /** Estimated `(measurement type, local day)` aggregate rows considered. */ rows: number; }; + /** Bounded auxiliary ECG outcomes. Never contains filenames or waveform data. */ + ecg: { + discovered: number; + imported: number; + updated: number; + skipped: number; + failed: number; + }; totals: { recordsRead: number; rowsUpserted: number; @@ -1086,6 +1094,13 @@ export async function streamParseExportXml( days: cumulativeEstimatedDays.size, rows: cumulativeEstimatedRows, }, + ecg: { + discovered: 0, + imported: 0, + updated: 0, + skipped: 0, + failed: 0, + }, totals: { recordsRead, rowsUpserted,