diff --git a/src/lib/import/unzip-export-xml.ts b/src/lib/import/unzip-export-xml.ts new file mode 100644 index 000000000..9024cdd17 --- /dev/null +++ b/src/lib/import/unzip-export-xml.ts @@ -0,0 +1,254 @@ +/** + * Minimal `export.zip` extractor — pulls `apple_health_export/export.xml` + * out of an Apple Health export archive without spawning a third-party + * dependency. + * + * Why hand-rolled: every iOS Apple Health export lands as a single + * deflate-compressed ZIP archive. Pulling in `yauzl` / `adm-zip` / etc. + * for one file we only ever read once would balloon the build graph + * (and `yauzl` ships a non-trivial number of transitive deps). Node 22 + * ships `node:zlib` `inflateRawSync` already; the only missing piece + * is a tiny ZIP central-directory walker. + * + * Coverage scope: + * - Stored entries (compression method 0) — copy bytes verbatim. + * - Deflated entries (compression method 8) — `inflateRawSync`. + * - Encrypted entries → unsupported (Apple does not encrypt the + * export.zip; reject with a clear error). + * - Zip64 — supported for the central directory record (Apple + * exports easily push past the 4 GB limit on multi-year accounts). + * + * Locks per `.planning/research/v1434-r-1-xml-import.md` §6.1. + */ +import { readFileSync, writeFileSync } from "node:fs"; +import { inflateRawSync } from "node:zlib"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; + +/** Signature bytes for the End-Of-Central-Directory record. */ +const EOCD_SIGNATURE = 0x06054b50; +const ZIP64_EOCD_LOCATOR = 0x07064b50; +const ZIP64_EOCD_RECORD = 0x06064b50; +const CENTRAL_FILE_HEADER = 0x02014b50; +const LOCAL_FILE_HEADER = 0x04034b50; + +/** A single entry within the archive's central directory. */ +interface CentralDirectoryEntry { + fileName: string; + compressionMethod: number; + compressedSize: number; + uncompressedSize: number; + localHeaderOffset: number; +} + +/** + * Result of `extractExportXml()`. The XML is written to a temp file + * because the streaming SAX parser expects a path it can `createReadStream` + * against. + */ +export interface UnzipResult { + /** Filesystem path the extracted `export.xml` lives at. */ + xmlPath: string; + /** Uncompressed size in bytes of the extracted XML. */ + xmlBytes: number; + /** Members the parser ignored (everything other than `export.xml`). */ + otherMembers: { name: string; bytes: number }[]; +} + +/** + * Walk the central directory of `archivePath` and write the + * `apple_health_export/export.xml` member out to a temp file. + * Throws when the member is missing, encrypted, or compressed with + * an unsupported method. + */ +export function extractExportXml(archivePath: string): UnzipResult { + const buf = readFileSync(archivePath); + const entries = readCentralDirectory(buf); + + const exportXmlEntry = entries.find( + (e) => + e.fileName.endsWith("/export.xml") || e.fileName === "export.xml", + ); + if (!exportXmlEntry) { + throw new Error( + "Archive is missing the `apple_health_export/export.xml` member" + + " — is this a valid Apple Health export.zip?", + ); + } + + if ( + exportXmlEntry.compressionMethod !== 0 + && exportXmlEntry.compressionMethod !== 8 + ) { + throw new Error( + `Unsupported ZIP compression method ${exportXmlEntry.compressionMethod}` + + " for export.xml (expected 0=stored or 8=deflate)", + ); + } + + const xmlBytes = extractEntry(buf, exportXmlEntry); + + const xmlPath = join( + tmpdir(), + `healthlog-import-${Date.now()}-${Math.random().toString(36).slice(2, 10)}.xml`, + ); + writeFileSync(xmlPath, xmlBytes); + + const otherMembers = entries + .filter((e) => e !== exportXmlEntry) + .map((e) => ({ name: e.fileName, bytes: e.uncompressedSize })); + + return { + xmlPath, + xmlBytes: xmlBytes.length, + otherMembers, + }; +} + +/** + * Walk the ZIP central directory and return one descriptor per member. + * Exported for unit testing the central-directory parser independently + * of the file-extraction path. + */ +export function readCentralDirectory( + buf: Buffer, +): CentralDirectoryEntry[] { + const eocdOffset = locateEocd(buf); + if (eocdOffset === -1) { + throw new Error("Could not locate ZIP End-Of-Central-Directory record"); + } + + let centralDirSize = buf.readUInt32LE(eocdOffset + 12); + let centralDirOffset = buf.readUInt32LE(eocdOffset + 16); + let entryCount = buf.readUInt16LE(eocdOffset + 10); + + // Zip64 handling — when any of the three count/offset fields is the + // 0xFFFFFFFF / 0xFFFF sentinel, the real value lives in the Zip64 + // EOCD record located by walking back through the locator. + if ( + centralDirOffset === 0xffffffff + || centralDirSize === 0xffffffff + || entryCount === 0xffff + ) { + const locatorOffset = eocdOffset - 20; + if (locatorOffset < 0 + || buf.readUInt32LE(locatorOffset) !== ZIP64_EOCD_LOCATOR) { + throw new Error("Zip64 sentinels present but Zip64 locator missing"); + } + const zip64EocdOffset = Number(buf.readBigUInt64LE(locatorOffset + 8)); + if (buf.readUInt32LE(zip64EocdOffset) !== ZIP64_EOCD_RECORD) { + throw new Error("Zip64 EOCD record signature mismatch"); + } + entryCount = Number(buf.readBigUInt64LE(zip64EocdOffset + 32)); + centralDirSize = Number(buf.readBigUInt64LE(zip64EocdOffset + 40)); + centralDirOffset = Number(buf.readBigUInt64LE(zip64EocdOffset + 48)); + } + + const entries: CentralDirectoryEntry[] = []; + let cursor = centralDirOffset; + const end = centralDirOffset + centralDirSize; + while (cursor < end && entries.length < entryCount) { + if (buf.readUInt32LE(cursor) !== CENTRAL_FILE_HEADER) { + throw new Error( + `Central directory entry ${entries.length} has wrong signature`, + ); + } + const compressionMethod = buf.readUInt16LE(cursor + 10); + const flags = buf.readUInt16LE(cursor + 8); + if (flags & 0x0001) { + throw new Error("Encrypted ZIP entries are not supported"); + } + const compressedSize32 = buf.readUInt32LE(cursor + 20); + const uncompressedSize32 = buf.readUInt32LE(cursor + 24); + const fileNameLen = buf.readUInt16LE(cursor + 28); + const extraLen = buf.readUInt16LE(cursor + 30); + const commentLen = buf.readUInt16LE(cursor + 32); + const localHeaderOffset32 = buf.readUInt32LE(cursor + 42); + + const fileName = buf + .slice(cursor + 46, cursor + 46 + fileNameLen) + .toString("utf8"); + + let compressedSize = compressedSize32; + let uncompressedSize = uncompressedSize32; + let localHeaderOffset = localHeaderOffset32; + + // Zip64 extra-field walk if any of the three values is sentinel. + if ( + compressedSize === 0xffffffff + || uncompressedSize === 0xffffffff + || localHeaderOffset === 0xffffffff + ) { + const extraStart = cursor + 46 + fileNameLen; + let extraCursor = extraStart; + const extraEnd = extraStart + extraLen; + while (extraCursor + 4 <= extraEnd) { + const headerId = buf.readUInt16LE(extraCursor); + const dataSize = buf.readUInt16LE(extraCursor + 2); + if (headerId === 0x0001) { + // Zip64 extended-info extra field + let dCursor = extraCursor + 4; + if (uncompressedSize === 0xffffffff) { + uncompressedSize = Number(buf.readBigUInt64LE(dCursor)); + dCursor += 8; + } + if (compressedSize === 0xffffffff) { + compressedSize = Number(buf.readBigUInt64LE(dCursor)); + dCursor += 8; + } + if (localHeaderOffset === 0xffffffff) { + localHeaderOffset = Number(buf.readBigUInt64LE(dCursor)); + dCursor += 8; + } + break; + } + extraCursor += 4 + dataSize; + } + } + + entries.push({ + fileName, + compressionMethod, + compressedSize, + uncompressedSize, + localHeaderOffset, + }); + + cursor += 46 + fileNameLen + extraLen + commentLen; + } + + return entries; +} + +/** Extract the bytes of a single ZIP entry into a Buffer. */ +function extractEntry(buf: Buffer, entry: CentralDirectoryEntry): Buffer { + const local = entry.localHeaderOffset; + if (buf.readUInt32LE(local) !== LOCAL_FILE_HEADER) { + throw new Error( + `Local file header for ${entry.fileName} has wrong signature`, + ); + } + const localFileNameLen = buf.readUInt16LE(local + 26); + const localExtraLen = buf.readUInt16LE(local + 28); + const dataStart = local + 30 + localFileNameLen + localExtraLen; + const compressed = buf.slice(dataStart, dataStart + entry.compressedSize); + + if (entry.compressionMethod === 0) { + return compressed; + } + // Method 8 — DEFLATE. + return inflateRawSync(compressed); +} + +/** + * Scan backwards from the end of the buffer for the EOCD signature. + * The EOCD lives in the last 22 + comment bytes; a comment is rare in + * the Health export but we tolerate up to 64 KB just in case. + */ +function locateEocd(buf: Buffer): number { + const searchStart = Math.max(0, buf.length - (22 + 0x10000)); + for (let i = buf.length - 22; i >= searchStart; i--) { + if (buf.readUInt32LE(i) === EOCD_SIGNATURE) return i; + } + return -1; +} diff --git a/src/lib/measurements/hk-workout-activity-type-map.ts b/src/lib/measurements/hk-workout-activity-type-map.ts new file mode 100644 index 000000000..b64fb6a62 --- /dev/null +++ b/src/lib/measurements/hk-workout-activity-type-map.ts @@ -0,0 +1,141 @@ +/** + * `HKWorkoutActivityType` → HealthLog `WorkoutSportType` lookup table. + * + * Apple's `HKWorkoutActivityType` is an `NS_ENUM(NSUInteger, ...)` with + * 75+ members at iOS 18. Only the canonical roster surfaced by the + * Health app export (`workoutActivityType="HKWorkoutActivityType*"`) + * gets a first-class entry here; anything outside the table falls + * through to `"other"` — the escape hatch already on the + * `workoutSportTypeEnum` union in `src/lib/validations/workout.ts`. + * + * Adding a new sport: append the literal to `workoutSportTypeEnum`, + * then add the HK activity-type entry here. The unit test under + * `__tests__/hk-workout-activity-type-map.test.ts` enforces that every + * value referenced here exists in the canonical union. + * + * Source: Apple's `HealthKit/HKWorkoutActivityType.h` header (iOS 18 + * SDK) cross-checked against the activity strings observed in + * open-source Apple Health export corpora — `apple-health-grafana`, + * `healthkit-to-sqlite`, `apple-health-parser`. Locked per + * `.planning/research/v1434-r-1-xml-import.md` §7. + */ +import type { WorkoutSportType } from "@/lib/validations/workout"; + +/** + * Map every known `HKWorkoutActivityType` value (the literal string + * Apple writes to `Workout.workoutActivityType` in `export.xml`) to a + * HealthLog sport-type union member. Unknown identifiers fall through + * to `"other"` at the call site; the parser increments the + * `workouts.unknownActivityType` counter so operators can spot new + * iOS releases that introduce a sport this table does not know yet. + */ +export const HK_WORKOUT_ACTIVITY_TYPE_MAP: Record = { + // ── Endurance ─────────────────────────────────────────────── + HKWorkoutActivityTypeWalking: "walking", + HKWorkoutActivityTypeRunning: "running", + HKWorkoutActivityTypeCycling: "cycling", + HKWorkoutActivityTypeHiking: "hiking", + HKWorkoutActivityTypeSwimming: "swimming", + HKWorkoutActivityTypeSwimBikeRun: "crossTraining", + HKWorkoutActivityTypeRowing: "rowing", + HKWorkoutActivityTypeElliptical: "elliptical", + HKWorkoutActivityTypeStairClimbing: "stairClimber", + HKWorkoutActivityTypeStairs: "stairClimber", + HKWorkoutActivityTypeStepTraining: "stairClimber", + + // ── Mind & body ───────────────────────────────────────────── + HKWorkoutActivityTypeYoga: "yoga", + HKWorkoutActivityTypeMindAndBody: "mindAndBody", + HKWorkoutActivityTypeFlexibility: "mindAndBody", + HKWorkoutActivityTypePilates: "mindAndBody", + HKWorkoutActivityTypeTaiChi: "mindAndBody", + HKWorkoutActivityTypeBarre: "mindAndBody", + HKWorkoutActivityTypeCooldown: "mindAndBody", + HKWorkoutActivityTypePreparationAndRecovery: "mindAndBody", + + // ── Strength ──────────────────────────────────────────────── + HKWorkoutActivityTypeFunctionalStrengthTraining: "strength", + HKWorkoutActivityTypeTraditionalStrengthTraining: "strength", + HKWorkoutActivityTypeCoreTraining: "strength", + HKWorkoutActivityTypeKickboxing: "strength", + HKWorkoutActivityTypeMartialArts: "strength", + + // ── High intensity / mixed ───────────────────────────────── + HKWorkoutActivityTypeHighIntensityIntervalTraining: "hiit", + HKWorkoutActivityTypeCrossTraining: "crossTraining", + HKWorkoutActivityTypeMixedCardio: "mixedCardio", + HKWorkoutActivityTypeMixedMetabolicCardioTraining: "mixedCardio", + HKWorkoutActivityTypeJumpRope: "hiit", + + // ── Dance ─────────────────────────────────────────────────── + HKWorkoutActivityTypeDance: "dance", + HKWorkoutActivityTypeDanceInspiredTraining: "dance", + HKWorkoutActivityTypeSocialDance: "dance", + HKWorkoutActivityTypeCardioDance: "dance", + + // ── Field & ball sports ───────────────────────────────────── + HKWorkoutActivityTypeGolf: "golf", + HKWorkoutActivityTypeTennis: "tennis", + HKWorkoutActivityTypeBasketball: "basketball", + HKWorkoutActivityTypeSoccer: "soccer", + HKWorkoutActivityTypeAmericanFootball: "soccer", + HKWorkoutActivityTypeAustralianFootball: "soccer", + HKWorkoutActivityTypeRugby: "soccer", + HKWorkoutActivityTypeBaseball: "tennis", + HKWorkoutActivityTypeCricket: "tennis", + HKWorkoutActivityTypeBadminton: "tennis", + HKWorkoutActivityTypeRacquetball: "tennis", + HKWorkoutActivityTypeSquash: "tennis", + HKWorkoutActivityTypeTableTennis: "tennis", + HKWorkoutActivityTypeVolleyball: "tennis", + HKWorkoutActivityTypeHandball: "tennis", + HKWorkoutActivityTypeHockey: "tennis", + HKWorkoutActivityTypeLacrosse: "tennis", + HKWorkoutActivityTypeSoftball: "tennis", + HKWorkoutActivityTypePaddleSports: "rowing", + HKWorkoutActivityTypePickleball: "tennis", + + // ── Snow / ice / water ────────────────────────────────────── + HKWorkoutActivityTypeDownhillSkiing: "other", + HKWorkoutActivityTypeCrossCountrySkiing: "other", + HKWorkoutActivityTypeSnowboarding: "other", + HKWorkoutActivityTypeSnowSports: "other", + HKWorkoutActivityTypeSkatingSports: "other", + HKWorkoutActivityTypeWaterFitness: "swimming", + HKWorkoutActivityTypeWaterPolo: "swimming", + HKWorkoutActivityTypeWaterSports: "swimming", + HKWorkoutActivityTypeSurfingSports: "other", + HKWorkoutActivityTypeSailing: "other", + HKWorkoutActivityTypeFishing: "other", + + // ── Wheelchair ────────────────────────────────────────────── + HKWorkoutActivityTypeWheelchairWalkPace: "walking", + HKWorkoutActivityTypeWheelchairRunPace: "running", + HKWorkoutActivityTypeHandCycling: "cycling", + + // ── Misc / catch-all ─────────────────────────────────────── + HKWorkoutActivityTypeBoxing: "strength", + HKWorkoutActivityTypeFencing: "other", + HKWorkoutActivityTypeArchery: "other", + HKWorkoutActivityTypeBowling: "other", + HKWorkoutActivityTypeClimbing: "other", + HKWorkoutActivityTypeEquestrianSports: "other", + HKWorkoutActivityTypeGymnastics: "other", + HKWorkoutActivityTypeHunting: "other", + HKWorkoutActivityTypePlay: "other", + HKWorkoutActivityTypeOther: "other", +}; + +/** + * Resolve a `WorkoutSportType` for the given `HKWorkoutActivityType` + * identifier string. Returns `"other"` and a falsy `known` flag for + * identifiers outside the table so the call site can increment its + * `unknownActivityType` counter without losing the row. + */ +export function resolveHkWorkoutSportType( + activityType: string, +): { sportType: WorkoutSportType; known: boolean } { + const sportType = HK_WORKOUT_ACTIVITY_TYPE_MAP[activityType]; + if (sportType) return { sportType, known: true }; + return { sportType: "other", known: false }; +} diff --git a/src/lib/measurements/import-apple-health-export.ts b/src/lib/measurements/import-apple-health-export.ts new file mode 100644 index 000000000..476b8a353 --- /dev/null +++ b/src/lib/measurements/import-apple-health-export.ts @@ -0,0 +1,767 @@ +/** + * Streaming Apple Health `export.xml` parser + ingest mapper. + * + * Reads an `export.xml` byte-stream via `sax` (event-driven SAX, no + * DOM) and folds every ``, ``, and `` + * element into the row shape the existing `Measurement` and + * `Workout` models expect. Cumulative-quantity `` rows + * (steps, energy, distance, flights, daylight) collapse into one + * `stats::` `Measurement` per user-local day + * — mirroring iOS's `HKStatisticsCollectionQuery` daily-aggregation + * convention locked in v1.4.30. Spot rows (BP, weight, HRV, …) + * survive verbatim, keyed by `HKSample.uuid` when present. + * + * Memory profile: SAX callbacks fire as the byte cursor advances, + * so peak RSS stays bounded regardless of the input file size. The + * cumulative-bucket map holds at most one `(type, day)` entry per + * observed day per cumulative type — a 10-year export with five + * cumulative types lands at ~18 000 entries (~1 MB). + * + * Locks per `.planning/research/v1434-r-1-xml-import.md` §6. + */ +import { createReadStream } from "node:fs"; +import { createHash } from "node:crypto"; +import { Writable } from "node:stream"; +import { pipeline } from "node:stream/promises"; +import sax from "sax"; + +import { Prisma } from "@/generated/prisma/client"; +import type { + MeasurementType, + PrismaClient, +} from "@/generated/prisma/client"; +import { + APPLE_HEALTH_SLEEP_STAGE_MAP, + APPLE_HEALTH_TYPE_MAP, + CUMULATIVE_HK_TYPES, + HK_QUANTITY_TYPE_DEFERRED, + dailyStatsExternalId, + mapAppleHealthEntry, +} from "@/lib/measurements/apple-health-mapping"; +import { + dayKeyForUserTz, + canonicalDailyTimestamp, +} from "@/lib/measurements/drain-per-sample-cumulative"; +import { resolveHkWorkoutSportType } from "@/lib/measurements/hk-workout-activity-type-map"; +import { validateMeasurementRange } from "@/lib/validations/measurement"; + +/** + * Reverse-lookup for the symbolic `HKCategoryValueSleepAnalysis*` + * names Apple writes to `Record[type="HKCategoryTypeIdentifierSleepAnalysis"]` + * `value` attributes. The web batch endpoint receives the integer + * codepoint pre-resolved by iOS; the XML export carries the symbolic + * name. Mirrors `APPLE_HEALTH_SLEEP_STAGE_MAP` inverted by stage + * label so the existing `mapAppleHealthEntry()` path picks up the + * integer codepoint without further translation. + */ +const SLEEP_STAGE_NAME_TO_CODEPOINT: Record = { + HKCategoryValueSleepAnalysisInBed: 0, + HKCategoryValueSleepAnalysisAsleep: 1, + HKCategoryValueSleepAnalysisAsleepUnspecified: 1, + HKCategoryValueSleepAnalysisAwake: 2, + HKCategoryValueSleepAnalysisAsleepCore: 3, + HKCategoryValueSleepAnalysisAsleepDeep: 4, + HKCategoryValueSleepAnalysisAsleepREM: 5, +}; + +/** Phase the worker is currently in. */ +export type ImportJobPhase = + | "queued" + | "unpacking" + | "parsing" + | "upserting" + | "done" + | "failed"; + +/** + * Live snapshot the worker writes to `ImportJob.progress` every + * `PROGRESS_TICK_RECORDS` records parsed. The polling endpoint + * returns it verbatim. + */ +export interface ImportJobProgress { + currentPhase: Exclude; + recordsRead: number; + rowsUpserted: number; + /** Percent is best-effort and may stay null until the parser sees + * `` (we don't know the total record count up-front). */ + percent: number | null; + elapsedMs: number; +} + +/** Final outcome carried on `ImportJob.result` once terminal. */ +export interface ImportJobResult { + perType: Record< + string, + { read: number; inserted: number; updated: number; durationMs: number } + >; + workouts: { + read: number; + inserted: number; + updated: number; + unknownActivityType: number; + routesAttached: number; + durationMs: number; + }; + clinical: { skipped: number }; + deferred: Record; + unknown: Record; + totals: { + recordsRead: number; + rowsUpserted: number; + durationMs: number; + }; +} + +/** Flush row written to the `Measurement` table. */ +interface PreparedMeasurement { + userId: string; + type: MeasurementType; + value: number; + unit: string; + measuredAt: Date; + externalId: string; + externalSourceVersion: string | null; + sleepStage: Prisma.MeasurementCreateInput["sleepStage"]; + deviceType: string | null; +} + +/** Flush row written to the `Workout` table. */ +interface PreparedWorkout { + userId: string; + sportType: string; + startedAt: Date; + endedAt: Date; + durationSec: number; + totalEnergyKcal: number | null; + totalDistanceM: number | null; + externalId: string; + externalSourceVersion: string | null; + metadata: Prisma.JsonValue | null; +} + +/** Per-type running stats accumulator. */ +interface MutableTypeStat { + read: number; + inserted: number; + updated: number; + durationMs: number; +} + +/** Tick frequency for the live `ImportJob.progress` write. */ +export const PROGRESS_TICK_RECORDS = 1_000; +const SPOT_FLUSH_BATCH = 500; +const WORKOUT_FLUSH_BATCH = 100; + +/** + * Hash the upload to a deterministic `externalId` when no + * `HKMetadataKeyExternalUUID` is present. Truncated to 28 hex chars + * (~112 bits) — well inside the 120-char Zod cap on `externalId`, + * and orders of magnitude more collision budget than any reasonable + * export could need. + */ +export function hashSampleKey( + hkIdentifier: string, + value: number | string, + startDate: string, + endDate: string, +): string { + return ( + "sample:" + + createHash("sha256") + .update(`${hkIdentifier}|${value}|${startDate}|${endDate}`) + .digest("hex") + .slice(0, 28) + ); +} + +/** + * Parse the Apple `Record` `value` attribute. Quantity records carry + * a numeric string; sleep-analysis records carry the symbolic name. + * Returns `null` for unparseable values so the caller can skip the + * row rather than poisoning the batch. + */ +export function parseRecordValue( + hkIdentifier: string, + rawValue: string | undefined, + startDate: string, + endDate: string, +): { value: number; sleepStage?: number } | null { + if (rawValue === undefined) return null; + if (hkIdentifier === "HKCategoryTypeIdentifierSleepAnalysis") { + const stage = SLEEP_STAGE_NAME_TO_CODEPOINT[rawValue]; + if (stage === undefined) return null; + // For sleep, our `mapAppleHealthEntry` expects the value to be the + // duration in minutes; derive it from start/end so the row carries + // the canonical reading downstream. + const start = new Date(startDate); + const end = new Date(endDate); + if (Number.isNaN(start.getTime()) || Number.isNaN(end.getTime())) { + return null; + } + const minutes = Math.max(0, (end.getTime() - start.getTime()) / 60_000); + return { value: minutes, sleepStage: stage }; + } + if (hkIdentifier === "HKCategoryTypeIdentifierEnvironmentalAudioExposureEvent" + || hkIdentifier === "HKCategoryTypeIdentifierHeadphoneAudioExposureEvent") { + // Apple writes these category-type events with empty / sentinel + // value strings. `mapAppleHealthEntry()` ignores the inbound + // number and always converts to a 1.0 count. + return { value: 1 }; + } + const parsed = Number.parseFloat(rawValue); + if (!Number.isFinite(parsed)) return null; + return { value: parsed }; +} + +/** + * Input to `streamParseExportXml()`. + */ +export interface StreamParseInput { + /** Filesystem path to an unzipped `export.xml`. */ + xmlPath: string; + /** Owner of the imported rows. */ + userId: string; + /** IANA timezone, used to anchor cumulative-type day-keys. */ + userTimezone: string; + /** Prisma client to flush rows through. */ + prisma: Pick; + /** + * Live progress hook. Called every `PROGRESS_TICK_RECORDS` records + * read, and once on terminal `done`. Best-effort; the parser + * continues even if the hook throws. + */ + onProgress?: (snapshot: ImportJobProgress) => Promise | void; + /** + * Optional override of the spot-row flush batch size. Defaults to + * `SPOT_FLUSH_BATCH`. Lower values make the upsert path more + * granular for tests; production runs should keep the default. + */ + spotBatchSize?: number; + /** + * Optional override of the workout flush batch size. Defaults to + * `WORKOUT_FLUSH_BATCH`. + */ + workoutBatchSize?: number; +} + +/** + * Streaming parse of an Apple Health `export.xml`. Returns the + * terminal `ImportJobResult` envelope; mid-run updates flow through + * `onProgress`. Throws on a fatal SAX error. + * + * Idempotency contract: every row UPSERTs against the existing + * `(userId, type, source, externalId)` compound unique on + * `Measurement` and `(userId, source, externalId)` on `Workout`. A + * re-import of the exact same file reports 0 inserts and N updates. + */ +export async function streamParseExportXml( + input: StreamParseInput, +): Promise { + const { + xmlPath, + userId, + userTimezone, + prisma, + onProgress, + spotBatchSize = SPOT_FLUSH_BATCH, + workoutBatchSize = WORKOUT_FLUSH_BATCH, + } = input; + + const startedAt = Date.now(); + const perType: Record = {}; + const workouts = { + read: 0, + inserted: 0, + updated: 0, + unknownActivityType: 0, + routesAttached: 0, + durationMs: 0, + }; + const clinical = { skipped: 0 }; + const deferred: Record = {}; + const unknown: Record = {}; + + // Cumulative-type fold: type -> dayKey -> running sum. + const cumulativeBucket = new Map>(); + // Spot-row batch awaiting flush. + const spotBatch: PreparedMeasurement[] = []; + // Workout-row batch awaiting flush. + const workoutBatch: PreparedWorkout[] = []; + + let recordsRead = 0; + let rowsUpserted = 0; + // Per R-1 §8 the percent stays best-effort and may remain null + // until the parser sees the closing `` tag. We don't + // currently mutate this in v1.4.34 — the iOS app keeps polling on + // the elapsed-ms instead — but keep the field around so the + // progress envelope stays additive for future percent backfill. + const totalRecords: number | null = null; + + const bumpStat = (type: MeasurementType): MutableTypeStat => { + const key = type as string; + let stat = perType[key]; + if (!stat) { + stat = { read: 0, inserted: 0, updated: 0, durationMs: 0 }; + perType[key] = stat; + } + return stat; + }; + + const emitProgress = async ( + phase: ImportJobProgress["currentPhase"], + ): Promise => { + if (!onProgress) return; + const snapshot: ImportJobProgress = { + currentPhase: phase, + recordsRead, + rowsUpserted, + percent: + totalRecords && totalRecords > 0 + ? Math.min(99, Math.round((recordsRead / totalRecords) * 100)) + : null, + elapsedMs: Date.now() - startedAt, + }; + try { + await onProgress(snapshot); + } catch { + // best-effort; do not poison the parse loop + } + }; + + // ── Flush helpers ────────────────────────────────────────── + const flushSpotBatch = async (): Promise => { + if (spotBatch.length === 0) return; + const chunk = spotBatch.splice(0, spotBatch.length); + // Look up existing rows by the compound unique key so we can + // attribute inserts vs updates accurately for the per-type stats. + const existingRows = await prisma.measurement.findMany({ + where: { + userId, + source: "APPLE_HEALTH", + OR: chunk.map((r) => ({ + type: r.type, + externalId: r.externalId, + })), + }, + select: { type: true, externalId: true }, + }); + const existingKey = new Set( + existingRows.map((r) => `${r.type}::${r.externalId}`), + ); + + for (const row of chunk) { + const stat = bumpStat(row.type); + const rowStart = Date.now(); + const exists = existingKey.has(`${row.type}::${row.externalId}`); + await prisma.measurement.upsert({ + where: { + userId_type_source_externalId: { + userId, + type: row.type, + source: "APPLE_HEALTH", + externalId: row.externalId, + }, + }, + create: { + userId, + type: row.type, + value: row.value, + unit: row.unit, + source: "APPLE_HEALTH", + measuredAt: row.measuredAt, + externalId: row.externalId, + externalSourceVersion: row.externalSourceVersion, + sleepStage: row.sleepStage ?? null, + deviceType: row.deviceType, + }, + update: { + value: row.value, + measuredAt: row.measuredAt, + externalSourceVersion: row.externalSourceVersion, + sleepStage: row.sleepStage ?? null, + deviceType: row.deviceType, + }, + }); + stat.durationMs += Date.now() - rowStart; + if (exists) stat.updated += 1; + else stat.inserted += 1; + rowsUpserted += 1; + } + }; + + const flushWorkoutBatch = async (): Promise => { + if (workoutBatch.length === 0) return; + const chunk = workoutBatch.splice(0, workoutBatch.length); + const existingRows = await prisma.workout.findMany({ + where: { + userId, + source: "APPLE_HEALTH", + externalId: { in: chunk.map((r) => r.externalId) }, + }, + select: { externalId: true }, + }); + const existingKey = new Set(existingRows.map((r) => r.externalId)); + + for (const row of chunk) { + const rowStart = Date.now(); + const exists = existingKey.has(row.externalId); + await prisma.workout.upsert({ + where: { + userId_source_externalId: { + userId, + source: "APPLE_HEALTH", + externalId: row.externalId, + }, + }, + create: { + userId, + sportType: row.sportType, + startedAt: row.startedAt, + endedAt: row.endedAt, + durationSec: row.durationSec, + totalEnergyKcal: row.totalEnergyKcal, + totalDistanceM: row.totalDistanceM, + source: "APPLE_HEALTH", + externalId: row.externalId, + externalSourceVersion: row.externalSourceVersion, + metadata: row.metadata ?? undefined, + }, + update: { + sportType: row.sportType, + startedAt: row.startedAt, + endedAt: row.endedAt, + durationSec: row.durationSec, + totalEnergyKcal: row.totalEnergyKcal, + totalDistanceM: row.totalDistanceM, + externalSourceVersion: row.externalSourceVersion, + metadata: row.metadata ?? undefined, + }, + }); + workouts.durationMs += Date.now() - rowStart; + if (exists) workouts.updated += 1; + else workouts.inserted += 1; + rowsUpserted += 1; + } + }; + + const flushCumulativeBuckets = async (): Promise => { + for (const [type, byDay] of cumulativeBucket.entries()) { + // Re-resolve the HK identifier from the mapping table — the + // bucket keys are MeasurementType, but the externalId carries + // the HK identifier so a future re-import collides on the + // same string. + const mapping = Object.values(APPLE_HEALTH_TYPE_MAP).find( + (m) => m.measurementType === type, + ); + if (!mapping) continue; + const stat = bumpStat(type); + + for (const [dayKey, sum] of byDay.entries()) { + const externalId = dailyStatsExternalId(mapping.hkIdentifier, dayKey); + const measuredAt = canonicalDailyTimestamp(dayKey, userTimezone); + const rowStart = Date.now(); + const existing = await prisma.measurement.findUnique({ + where: { + userId_type_source_externalId: { + userId, + type, + source: "APPLE_HEALTH", + externalId, + }, + }, + select: { id: true }, + }); + await prisma.measurement.upsert({ + where: { + userId_type_source_externalId: { + userId, + type, + source: "APPLE_HEALTH", + externalId, + }, + }, + create: { + userId, + type, + value: sum, + unit: mapping.dbUnit, + source: "APPLE_HEALTH", + measuredAt, + externalId, + }, + update: { + value: sum, + measuredAt, + }, + }); + stat.durationMs += Date.now() - rowStart; + if (existing) stat.updated += 1; + else stat.inserted += 1; + rowsUpserted += 1; + } + } + }; + + // ── SAX parser configuration ──────────────────────────────── + const parser = sax.parser(true, { trim: true }); + let pendingError: Error | null = null; + let currentWorkout: PreparedWorkout | null = null; + + parser.onerror = (err) => { + pendingError = err instanceof Error ? err : new Error(String(err)); + }; + + parser.onopentag = (node) => { + const name = node.name; + const attrs = node.attributes as Record; + + if (name === "Record") { + const hkType = attrs.type; + if (!hkType) return; + recordsRead += 1; + + if (HK_QUANTITY_TYPE_DEFERRED.has(hkType)) { + deferred[hkType] = (deferred[hkType] ?? 0) + 1; + return; + } + const mapping = APPLE_HEALTH_TYPE_MAP[hkType]; + if (!mapping) { + unknown[hkType] = (unknown[hkType] ?? 0) + 1; + return; + } + + const parsedValue = parseRecordValue( + hkType, + attrs.value, + attrs.startDate, + attrs.endDate, + ); + if (!parsedValue) { + unknown[`${hkType}::unparseable`] = + (unknown[`${hkType}::unparseable`] ?? 0) + 1; + return; + } + + const mapped = mapAppleHealthEntry({ + hkIdentifier: hkType, + value: parsedValue.value, + unit: attrs.unit ?? mapping.hkUnit, + startDate: attrs.startDate, + endDate: attrs.endDate, + sleepStage: parsedValue.sleepStage, + }); + if (!mapped) { + unknown[`${hkType}::map_failed`] = + (unknown[`${hkType}::map_failed`] ?? 0) + 1; + return; + } + + // Plausibility-range guard. A single rogue sample shouldn't + // poison the import — record under `unknown` with the + // explicit reason tag so operators can spot ingest pathologies. + const rangeError = validateMeasurementRange(mapped.type, mapped.value); + if (rangeError !== null) { + unknown[`${hkType}::out_of_range`] = + (unknown[`${hkType}::out_of_range`] ?? 0) + 1; + return; + } + + const stat = bumpStat(mapped.type); + stat.read += 1; + + if (CUMULATIVE_HK_TYPES.has(mapped.type)) { + const dayKey = dayKeyForUserTz(mapped.takenAt, userTimezone); + let byDay = cumulativeBucket.get(mapped.type); + if (!byDay) { + byDay = new Map(); + cumulativeBucket.set(mapped.type, byDay); + } + byDay.set(dayKey, (byDay.get(dayKey) ?? 0) + mapped.value); + } else { + // Spot row: derive a stable externalId, queue for flush. + const externalId = hashSampleKey( + hkType, + attrs.value ?? "", + attrs.startDate, + attrs.endDate, + ); + spotBatch.push({ + userId, + type: mapped.type, + value: mapped.value, + unit: mapped.unit, + measuredAt: mapped.takenAt, + externalId, + externalSourceVersion: attrs.sourceVersion ?? null, + sleepStage: mapped.sleepStage ?? null, + deviceType: null, + }); + } + return; + } + + if (name === "Workout") { + recordsRead += 1; + workouts.read += 1; + const activityType = attrs.workoutActivityType ?? ""; + const { sportType, known } = resolveHkWorkoutSportType(activityType); + if (!known) workouts.unknownActivityType += 1; + + const startDate = new Date(attrs.startDate); + const endDate = new Date(attrs.endDate); + if (Number.isNaN(startDate.getTime()) || Number.isNaN(endDate.getTime())) { + unknown[`Workout::bad_dates`] = + (unknown[`Workout::bad_dates`] ?? 0) + 1; + return; + } + const durationSec = Math.max( + 0, + Math.round((endDate.getTime() - startDate.getTime()) / 1000), + ); + + const totalDistance = Number.parseFloat(attrs.totalDistance ?? ""); + const totalEnergy = Number.parseFloat(attrs.totalEnergyBurned ?? ""); + const distanceUnit = attrs.totalDistanceUnit ?? ""; + // Apple ships km for HKWorkout.totalDistance by default; metres + // is the canonical DB unit. Convert when the unit is km. + const distanceM = Number.isFinite(totalDistance) + ? distanceUnit === "km" + ? totalDistance * 1000 + : distanceUnit === "mi" + ? totalDistance * 1609.344 + : totalDistance + : null; + + const externalId = hashSampleKey( + activityType || "Workout", + attrs.duration ?? "", + attrs.startDate, + attrs.endDate, + ); + + currentWorkout = { + userId, + sportType, + startedAt: startDate, + endedAt: endDate, + durationSec, + totalEnergyKcal: Number.isFinite(totalEnergy) ? totalEnergy : null, + totalDistanceM: distanceM, + externalId, + externalSourceVersion: attrs.sourceVersion ?? null, + metadata: { + activityType, + sourceName: attrs.sourceName ?? null, + durationUnit: attrs.durationUnit ?? null, + totalDistanceUnit: distanceUnit || null, + totalEnergyBurnedUnit: attrs.totalEnergyBurnedUnit ?? null, + } as Prisma.JsonValue, + }; + return; + } + + if (name === "ClinicalRecord") { + recordsRead += 1; + clinical.skipped += 1; + return; + } + + if (name === "ExportDate" || name === "Me" || name === "Correlation" + || name === "ActivitySummary" || name === "MetadataEntry" + || name === "WorkoutEvent" || name === "WorkoutRoute" + || name === "FileReference" || name === "HealthData") { + // Known elements we intentionally ignore at the open-tag stage. + // Correlation envelopes flatten naturally because their child + // `` elements still fire their own `onopentag`. + return; + } + + // Any other element name lands as a structural unknown — log + // once per name so operators can spot future schema additions. + unknown[`element::${name}`] = (unknown[`element::${name}`] ?? 0) + 1; + }; + + parser.onclosetag = async (tagName) => { + if (tagName === "Workout" && currentWorkout) { + workoutBatch.push(currentWorkout); + currentWorkout = null; + } + }; + + // ── Drive the parser from a node read stream ──────────────── + const readable = createReadStream(xmlPath, { highWaterMark: 64 * 1024 }); + + let lastProgressEmitAt = 0; + const PROGRESS_EMIT_INTERVAL_MS = 250; + + const sink = new Writable({ + write(chunk: Buffer, _enc, callback) { + try { + parser.write(chunk.toString("utf8")); + if (pendingError) { + callback(pendingError); + return; + } + // Drain any batches the parser filled to keep memory bounded. + const shouldFlush = + spotBatch.length >= spotBatchSize + || workoutBatch.length >= workoutBatchSize; + const drain = async (): Promise => { + if (spotBatch.length >= spotBatchSize) await flushSpotBatch(); + if (workoutBatch.length >= workoutBatchSize) await flushWorkoutBatch(); + const now = Date.now(); + if ( + (recordsRead > 0 && recordsRead % PROGRESS_TICK_RECORDS === 0) + || now - lastProgressEmitAt > PROGRESS_EMIT_INTERVAL_MS + ) { + lastProgressEmitAt = now; + await emitProgress("parsing"); + } + }; + if (shouldFlush || recordsRead % PROGRESS_TICK_RECORDS === 0) { + drain().then(() => callback()).catch((err) => callback(err)); + } else { + callback(); + } + } catch (err) { + callback(err instanceof Error ? err : new Error(String(err))); + } + }, + final(callback) { + try { + parser.close(); + callback(); + } catch (err) { + callback(err instanceof Error ? err : new Error(String(err))); + } + }, + }); + + await pipeline(readable, sink); + if (pendingError) throw pendingError; + + await emitProgress("upserting"); + // Drain any partial spot/workout batches that did not hit the flush + // threshold during the parse. + await flushSpotBatch(); + await flushWorkoutBatch(); + await flushCumulativeBuckets(); + + await emitProgress("upserting"); + + return { + perType, + workouts, + clinical, + deferred, + unknown, + totals: { + recordsRead, + rowsUpserted, + durationMs: Date.now() - startedAt, + }, + }; +} + +// Re-export the SLEEP_STAGE table for unit tests. +export { SLEEP_STAGE_NAME_TO_CODEPOINT, APPLE_HEALTH_SLEEP_STAGE_MAP };