feat(01-19): persist Apple Health ECG recordings

- encrypt and upsert stable content-addressed ECG recordings transactionally
- keep XML imports successful while ECG members fail with scalar counters

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<PrismaClient, "$transaction">;
+}): Promise<AppleHealthEcgImportOutcome> {
+  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);
+  }
+}

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({

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,