diff --git a/prisma/schema.prisma b/prisma/schema.prisma index 311b0c061..77a61c6cf 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -4968,8 +4968,16 @@ model ImportJob { /// `APPLE_HEALTH_IMPORT_PARSER_REVISION`, so a job parsed under an older /// revision never short-circuits a fresh upload of the same bytes. parserRevision Int @default(1) @map("parser_revision") - /// Apple's `ExportDate` attribute string-formatted, parsed during - /// the unpack phase. NULL until the parser sees it. + /// The instant the Health app stamped on the archive itself — the + /// `value` attribute of the `` element that opens + /// `export.xml`. Written mid-run by the parse phase the moment the + /// element goes past (`onExportDate` in + /// `src/lib/measurements/import-apple-health-export.ts`, persisted by + /// `src/lib/jobs/apple-health-import-worker.ts`), so a job that later + /// fails still records which export it was working from. Consumer: the + /// status endpoint (`src/app/api/import/apple-health-export/[jobId]/ + /// status/route.ts`). NULL until the parser reaches the element, and + /// for an archive that omits it or writes an unreadable value. exportedAt DateTime? @map("exported_at") startedAt DateTime @default(now()) @map("started_at") completedAt DateTime? @map("completed_at") diff --git a/src/lib/jobs/__tests__/apple-health-import-export-date.test.ts b/src/lib/jobs/__tests__/apple-health-import-export-date.test.ts new file mode 100644 index 000000000..4b9b55f37 --- /dev/null +++ b/src/lib/jobs/__tests__/apple-health-import-export-date.test.ts @@ -0,0 +1,69 @@ +/** + * `ImportJob.exportedAt` — the worker end of the wiring. + * + * The parse side is proved behaviourally in + * `src/lib/measurements/__tests__/import-apple-health-export.test.ts`: + * the parser hands the archive's own stamp to `onExportDate`. What that + * cannot show is whether anything persists it. The column, the status + * endpoint that returns it and the published contract that requires it + * all existed for a year while the one line that would produce the value + * was a `return`, so the half of the pipe worth pinning is this one. + * + * Source-shaped rather than run-shaped: driving the worker needs a real + * `export.zip`, a queue and a database, which the ECG contract test + * beside this one already declines to do for the same reason. + */ +import { readFileSync } from "node:fs"; +import { join } from "node:path"; +import { describe, expect, it } from "vitest"; + +const workerSource = readFileSync( + join(process.cwd(), "src/lib/jobs/apple-health-import-worker.ts"), + "utf8", +); +const parserSource = readFileSync( + join(process.cwd(), "src/lib/measurements/import-apple-health-export.ts"), + "utf8", +); +const statusRouteSource = readFileSync( + join( + process.cwd(), + "src/app/api/import/apple-health-export/[jobId]/status/route.ts", + ), + "utf8", +); + +describe("Apple Health import — the archive's export stamp reaches the column", () => { + it("subscribes to the parser's export-date hook", () => { + expect(workerSource).toMatch(/onExportDate:\s*async\s*\(exportedAt\)/); + }); + + it("writes the stamp onto the job row it is running", () => { + expect(workerSource).toMatch( + /onExportDate:[\s\S]{0,400}?prisma\s*\.?\s*importJob[\s\S]{0,200}?\.update\([\s\S]{0,200}?exportedAt/, + ); + }); + + it("keeps the write best-effort so provenance cannot fail an import", () => { + expect(workerSource).toMatch( + /onExportDate:[\s\S]{0,400}?exportedAt[\s\S]{0,120}?\.catch\(/, + ); + }); + + it("no longer discards the element that carries the stamp", () => { + // The ignore list is where `ExportDate` spent its first year. + const ignoreBlock = /Known elements we intentionally ignore/.exec( + parserSource, + ); + expect(ignoreBlock).not.toBeNull(); + const before = parserSource.slice( + Math.max(0, (ignoreBlock?.index ?? 0) - 400), + ignoreBlock?.index ?? 0, + ); + expect(before).not.toMatch(/name === "ExportDate" \|\|/); + }); + + it("still surfaces the column the contract promises", () => { + expect(statusRouteSource).toMatch(/exportedAt:\s*row\.exportedAt/); + }); +}); diff --git a/src/lib/jobs/apple-health-import-worker.ts b/src/lib/jobs/apple-health-import-worker.ts index 60cc8b8e9..c610a4825 100644 --- a/src/lib/jobs/apple-health-import-worker.ts +++ b/src/lib/jobs/apple-health-import-worker.ts @@ -417,6 +417,16 @@ export async function handleAppleHealthImport( }, }); }, + // The instant the Health app stamped on the archive. Persisted as + // soon as the parser reads it, so the status endpoint answers with + // it for the rest of the run and a job that later fails still + // records which export it was working from. Best-effort: the + // stamp is provenance, and losing it must not fail an import. + onExportDate: async (exportedAt) => { + await prisma.importJob + .update({ where: { id: importJobId }, data: { exportedAt } }) + .catch(() => {}); + }, })), ecg: { discovered: 0, diff --git a/src/lib/measurements/__tests__/import-apple-health-export.test.ts b/src/lib/measurements/__tests__/import-apple-health-export.test.ts index a2a018d1b..ddb4c3c6c 100644 --- a/src/lib/measurements/__tests__/import-apple-health-export.test.ts +++ b/src/lib/measurements/__tests__/import-apple-health-export.test.ts @@ -1226,3 +1226,55 @@ describe("streamParseExportXml — cycle module off", () => { expect(result.totals.recordsRead).toBe(2); }); }); + +/** + * `ImportJob.exportedAt` — the instant the Health app stamped on the + * archive. The status endpoint and the published contract have both + * promised it since v1.4.34, but `` sat on the parser's + * ignore list, so the column could only ever answer null. These pin the + * capture: remove the `onExportDate` call from the parser and the first + * case goes red; drop the NaN guard and the third does. + */ +describe("streamParseExportXml — the archive's own export stamp", () => { + async function parseFor(xml: string) { + const tmp = mkdtempSync(join(tmpdir(), "healthlog-parser-export-date-")); + const xmlPath = join(tmp, "export.xml"); + writeFileSync(xmlPath, xml); + const stamps: Date[] = []; + await streamParseExportXml({ + xmlPath, + userId: "user-1", + userTimezone: "Europe/Berlin", + // eslint-disable-next-line @typescript-eslint/no-explicit-any + prisma: makeFakePrisma() as any, + onExportDate: (exportedAt) => { + stamps.push(exportedAt); + }, + }); + return stamps; + } + + it("hands the caller the instant Apple wrote on the export", async () => { + const stamps = await parseFor(tinyExportXml()); + + expect(stamps).toHaveLength(1); + // "2026-05-15 14:32:01 +0200" — the fixture's own stamp. + expect(stamps[0]?.toISOString()).toBe("2026-05-15T12:32:01.000Z"); + }); + + it("stays silent for an archive that carries no stamp", async () => { + const stamps = await parseFor( + `\n`, + ); + + expect(stamps).toEqual([]); + }); + + it("stays silent rather than guessing at an unreadable stamp", async () => { + const stamps = await parseFor( + `\n\n \n`, + ); + + expect(stamps).toEqual([]); + }); +}); diff --git a/src/lib/measurements/import-apple-health-export.ts b/src/lib/measurements/import-apple-health-export.ts index df9ae28e6..26a1ce0bb 100644 --- a/src/lib/measurements/import-apple-health-export.ts +++ b/src/lib/measurements/import-apple-health-export.ts @@ -307,6 +307,13 @@ export interface StreamParseInput { * continues even if the hook throws. */ onProgress?: (snapshot: ImportJobProgress) => Promise | void; + /** + * Called once, mid-run, with the instant Apple stamped on the archive + * (``, the first child of ``). + * Never called for an archive that omits the element or writes an + * unreadable value. Best-effort, like `onProgress`. + */ + onExportDate?: (exportedAt: Date) => Promise | void; /** * Optional override of the spot-row flush batch size. Defaults to * `SPOT_FLUSH_BATCH`. Lower values make the upsert path more @@ -339,6 +346,7 @@ export async function streamParseExportXml( userTimezone, prisma, onProgress, + onExportDate, spotBatchSize = SPOT_FLUSH_BATCH, workoutBatchSize = WORKOUT_FLUSH_BATCH, } = input; @@ -775,6 +783,21 @@ export async function streamParseExportXml( let pendingError: Error | null = null; let currentWorkout: PreparedWorkout | null = null; + // Apple stamps the archive with its own export instant as the first + // child of ``. The element used to be on the ignore list, + // so `ImportJob.exportedAt` — a column the status endpoint and the + // published contract both promise — could only ever answer null. The + // value is captured in the synchronous tag handler and handed to + // `onExportDate` from the async drain, so the worker can persist it + // while the run is still going rather than only on the terminal write. + let pendingExportDate: Date | null = null; + const flushExportDate = async (): Promise => { + const stamped = pendingExportDate; + if (!stamped) return; + pendingExportDate = null; + if (onExportDate) await onExportDate(stamped); + }; + parser.onerror = (err) => { pendingError = err instanceof Error ? err : new Error(String(err)); }; @@ -983,8 +1006,17 @@ export async function streamParseExportXml( return; } + if (name === "ExportDate") { + // `value` carries Apple's own format ("2026-05-15 14:32:01 +0200"), + // the same shape every `Record` start/end date uses. An archive + // without the element, or with a value the runtime cannot read, + // leaves the column null rather than guessing at an instant. + const stamped = new Date(attrs.value ?? ""); + if (!Number.isNaN(stamped.getTime())) pendingExportDate = stamped; + return; + } + if ( - name === "ExportDate" || name === "Me" || name === "Correlation" || name === "ActivitySummary" || @@ -1049,9 +1081,11 @@ export async function streamParseExportXml( } // Drain any batches the parser filled to keep memory bounded. const shouldFlush = + pendingExportDate !== null || spotBatch.length >= spotBatchSize || workoutBatch.length >= workoutBatchSize; const drain = async (): Promise => { + await flushExportDate(); if (spotBatch.length >= spotBatchSize) await flushSpotBatch(); if (workoutBatch.length >= workoutBatchSize) await flushWorkoutBatch(); @@ -1090,6 +1124,10 @@ export async function streamParseExportXml( await pipeline(readable, sink); if (pendingError) throw pendingError; + // An archive small enough to arrive in a single chunk can finish + // before any drain ran; the stamp still has to reach the caller. + await flushExportDate(); + await emitProgress("upserting"); // Drain any partial spot/workout batches that did not hit the flush // threshold during the parse.