From 79843f87702d3073388f18824e08d0e18d0063b4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Thu, 23 Jul 2026 09:23:28 +0200 Subject: [PATCH 1/4] fix(import): stream the Apple Health ZIP extraction and sweep orphaned import jobs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Closes #588. The ZIP extractor decompressed export.xml synchronously into one in-memory buffer, then wrote it with a blocking writeFileSync — on the same event loop the web server runs in the default single-container deployment. A real multi-year export could pin the Node main thread for minutes, freezing every other request, while briefly holding roughly 2x the decompressed size in heap. A process that died mid-extraction never reached the failure path, so the ImportJob row stayed stuck in "unpacking" with "0 rows imported" forever and no error ever surfaced. Extraction now streams through zlib's async inflate into a file write stream, moving the decompression work off the event loop and bounding memory to the chunk size. The zip-bomb ceiling moves from inflateRawSync's maxOutputLength to an equivalent byte-counting transform on the streamed output. Orphan-ImportJob reconciliation used to run once, at worker boot. docker-compose's restart:unless-stopped does bring a crashed worker back, but a fast restart can land before the stuck row's heartbeat goes stale or pg-boss stops reporting the job as live — and because the boot-time pass never re-runs on a healthy worker, that row was never revisited again. A 15-minute cron now re-runs the same reconcile so a genuinely abandoned import converges to a visible failure instead of staying silently stuck. --- .../import/__tests__/unzip-export-xml.test.ts | 87 ++++++-- src/lib/import/unzip-export-xml.ts | 149 +++++++++++--- .../apple-health-import-reconcile.test.ts | 186 ++++++++++++++++++ src/lib/jobs/apple-health-import-worker.ts | 48 ++++- src/lib/jobs/reminder/register-maintenance.ts | 23 +++ .../__tests__/stream-to-disk.test.ts | 2 +- 6 files changed, 455 insertions(+), 40 deletions(-) create mode 100644 src/lib/jobs/__tests__/apple-health-import-reconcile.test.ts diff --git a/src/lib/import/__tests__/unzip-export-xml.test.ts b/src/lib/import/__tests__/unzip-export-xml.test.ts index 3151558c5..f006a3d45 100644 --- a/src/lib/import/__tests__/unzip-export-xml.test.ts +++ b/src/lib/import/__tests__/unzip-export-xml.test.ts @@ -3,16 +3,27 @@ import { writeFileSync, mkdtempSync, readFileSync } from "node:fs"; import { join } from "node:path"; import { tmpdir } from "node:os"; import { deflateRawSync } from "node:zlib"; +import { pipeline } from "node:stream/promises"; +import { Readable } from "node:stream"; -import { extractExportXml, readCentralDirectory } from "../unzip-export-xml"; +import { + createByteCap, + extractExportXml, + readCentralDirectory, +} from "../unzip-export-xml"; /** - * Hand-build a single-entry deflate-compressed ZIP archive with the - * supplied filename + payload. Enough to exercise the central-directory - * walker + extractor without pulling in a full zip library. + * Hand-build a single-entry ZIP archive with the supplied filename + + * payload. Enough to exercise the central-directory walker + extractor + * without pulling in a full zip library. `method` selects compression + * method 8 (deflate, default) or 0 (stored — bytes copied verbatim). */ -function buildMinimalZip(filename: string, payload: Buffer): Buffer { - const compressed = deflateRawSync(payload); +function buildMinimalZip( + filename: string, + payload: Buffer, + method: 0 | 8 = 8, +): Buffer { + const compressed = method === 8 ? deflateRawSync(payload) : payload; const crc32 = (() => { // Pre-computed lookup table for CRC-32. node:zlib has crc32 in 22+ // but we keep the function inline to stay version-agnostic. @@ -38,7 +49,7 @@ function buildMinimalZip(filename: string, payload: Buffer): Buffer { localHeader.writeUInt32LE(0x04034b50, 0); localHeader.writeUInt16LE(20, 4); // version localHeader.writeUInt16LE(0, 6); // flags - localHeader.writeUInt16LE(8, 8); // method: deflate + localHeader.writeUInt16LE(method, 8); // method localHeader.writeUInt16LE(0, 10); // time localHeader.writeUInt16LE(0, 12); // date localHeader.writeUInt32LE(crc32, 14); @@ -53,7 +64,7 @@ function buildMinimalZip(filename: string, payload: Buffer): Buffer { cdh.writeUInt16LE(20, 4); // version made by cdh.writeUInt16LE(20, 6); // version needed cdh.writeUInt16LE(0, 8); // flags - cdh.writeUInt16LE(8, 10); // method + cdh.writeUInt16LE(method, 10); // method cdh.writeUInt16LE(0, 12); // time cdh.writeUInt16LE(0, 14); // date cdh.writeUInt32LE(crc32, 16); @@ -95,7 +106,7 @@ describe("readCentralDirectory", () => { }); describe("extractExportXml", () => { - it("extracts the export.xml member to a temp file", () => { + it("extracts a deflated export.xml member to a temp file", async () => { const payload = Buffer.from( ``, ); @@ -104,7 +115,7 @@ describe("extractExportXml", () => { const zipPath = join(tmp, "export.zip"); writeFileSync(zipPath, zip); - const out = extractExportXml(zipPath); + const out = await extractExportXml(zipPath); expect(out.xmlBytes).toBe(payload.length); expect(out.otherMembers).toHaveLength(0); @@ -112,13 +123,65 @@ describe("extractExportXml", () => { expect(extracted.toString("utf8")).toBe(payload.toString("utf8")); }); - it("throws when the export.xml member is missing", () => { + // v1.32.1 (issue #588) — the stored (uncompressed) path now runs + // through the same streaming pipeline as deflate; a stored member + // must still round-trip byte-exact through it. + it("extracts a stored (uncompressed) export.xml member to a temp file", async () => { + const payload = Buffer.from( + ``, + ); + const zip = buildMinimalZip("apple_health_export/export.xml", payload, 0); + const tmp = mkdtempSync(join(tmpdir(), "healthlog-unzip-")); + const zipPath = join(tmp, "export.zip"); + writeFileSync(zipPath, zip); + + const out = await extractExportXml(zipPath); + expect(out.xmlBytes).toBe(payload.length); + + const extracted = readFileSync(out.xmlPath); + expect(extracted.toString("utf8")).toBe(payload.toString("utf8")); + }); + + it("throws when the export.xml member is missing", async () => { const payload = Buffer.from("noop"); const zip = buildMinimalZip("other-file.txt", payload); const tmp = mkdtempSync(join(tmpdir(), "healthlog-unzip-")); const zipPath = join(tmp, "export.zip"); writeFileSync(zipPath, zip); - expect(() => extractExportXml(zipPath)).toThrow(/missing/); + await expect(extractExportXml(zipPath)).rejects.toThrow(/missing/); + }); +}); + +// v1.32.1 (issue #588) — the streaming inflate Transform has no +// `maxOutputLength` option, so the zip-bomb defence moved to a +// hand-rolled byte-counting stage. Exercise it directly at a tiny +// limit; the real 8 GiB ceiling is not practical to trip from a +// fixture. +describe("createByteCap", () => { + it("passes bytes through unchanged while under the limit", async () => { + const chunks: Buffer[] = []; + await pipeline( + Readable.from([Buffer.from("hello"), Buffer.from("world")]), + createByteCap(10), + async function collect(source) { + for await (const chunk of source) chunks.push(chunk as Buffer); + }, + ); + expect(Buffer.concat(chunks).toString("utf8")).toBe("helloworld"); + }); + + it("rejects the stream once the running total exceeds the limit", async () => { + await expect( + pipeline( + Readable.from([Buffer.from("hello"), Buffer.from("world")]), + createByteCap(6), + async function drain(source) { + for await (const _chunk of source) { + /* drain */ + } + }, + ), + ).rejects.toThrow(/exceeds the 6-byte cap/); }); }); diff --git a/src/lib/import/unzip-export-xml.ts b/src/lib/import/unzip-export-xml.ts index 7333f2d09..6379ce0ec 100644 --- a/src/lib/import/unzip-export-xml.ts +++ b/src/lib/import/unzip-export-xml.ts @@ -7,21 +7,49 @@ * 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. + * ships `node:zlib` 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`. + * - Deflated entries (compression method 8) — streamed through + * `zlib.createInflateRaw()`. * - 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). * + * v1.32.1 (issue #588) — extracting the member used to run through + * `inflateRawSync()` on a single in-memory buffer holding the WHOLE + * inflated `export.xml` (up to `MAX_DECOMPRESSED_BYTES`) at the same + * time the compressed-archive buffer was still resident, then + * `writeFileSync()`'d the result in one blocking call. `extractExportXml` + * runs on the same event loop as the web server in the default + * single-container deployment (`src/instrumentation.ts` starts the + * pg-boss worker in-process), so that synchronous path could pin the + * Node main thread for minutes on a real multi-year export — freezing + * every other request the app was serving — while briefly needing + * roughly 2x the decompressed size in JS heap. A user hitting this saw + * an import that never leaves "Unpacking the archive…": the phase + * label is written to `ImportJob.status` BEFORE extraction starts and + * only advances once `extractExportXml()` returns, so a process that + * dies mid-extraction (OOM, restart) before that return leaves the row + * silently stuck with no failure ever recorded. The member now streams + * through `zlib.createInflateRaw()` into a file write stream — + * decompression work moves off the JS event loop onto the libuv + * threadpool in bounded chunks, and peak memory is bounded by the + * chunk size rather than the whole decompressed payload. The + * `MAX_DECOMPRESSED_BYTES` zip-bomb defence moves with it: the + * streaming inflate API has no `maxOutputLength` option, so a + * byte-counting transform enforces the cap on the real output as it + * flows. + * * Locks per `.planning/research/v1434-r-1-xml-import.md` §6.1. */ -import { readFileSync, writeFileSync } from "node:fs"; -import { inflateRawSync } from "node:zlib"; +import { readFileSync, createWriteStream, statSync, unlinkSync } from "node:fs"; +import { createInflateRaw } from "node:zlib"; +import { Readable, Transform } from "node:stream"; +import { pipeline } from "node:stream/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; @@ -44,11 +72,15 @@ const MAX_DECOMPRESSED_BYTES = 8 * 1024 * 1024 * 1024; * Pre-flight refusal threshold for the central-directory's advertised * ratio. Legitimate Apple Health XML compresses at maybe 10–20× under * DEFLATE; anything claiming a 200× expansion is a synthesized bomb. - * This is a coarse signal — the `maxOutputLength` ceiling on the - * inflate call below is the load-bearing defence. + * This is a coarse signal — the byte-counting cap on the streamed + * inflate output (`streamEntryToFile()` below) is the load-bearing + * defence. */ const MAX_COMPRESSION_RATIO = 200; +/** Chunk size fed into the inflate/write pipeline per `write()` call. */ +const STREAM_CHUNK_BYTES = 1 * 1024 * 1024; + /** A single entry within the archive's central directory. */ interface CentralDirectoryEntry { fileName: string; @@ -78,7 +110,9 @@ export interface UnzipResult { * Throws when the member is missing, encrypted, or compressed with * an unsupported method. */ -export function extractExportXml(archivePath: string): UnzipResult { +export async function extractExportXml( + archivePath: string, +): Promise { const buf = readFileSync(archivePath); const entries = readCentralDirectory(buf); @@ -105,8 +139,8 @@ export function extractExportXml(archivePath: string): UnzipResult { // Pre-flight zip-bomb defence. The central directory's advertised // `uncompressedSize` is attacker-controlled (a malicious archive can // lie), so this catches honest-but-oversized payloads early; the - // load-bearing defence is the `maxOutputLength` cap inside - // `extractEntry()` which trips on the actual inflate output. + // load-bearing defence is the byte-counting cap inside + // `streamEntryToFile()` which trips on the actual inflate output. if (exportXmlEntry.uncompressedSize > MAX_DECOMPRESSED_BYTES) { throw new Error( `export.xml declares an uncompressed size of ${exportXmlEntry.uncompressedSize} bytes` + @@ -126,13 +160,11 @@ export function extractExportXml(archivePath: string): UnzipResult { ); } - 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 xmlBytes = await streamEntryToFile(buf, exportXmlEntry, xmlPath); const otherMembers = entries .filter((e) => e !== exportXmlEntry) @@ -140,7 +172,7 @@ export function extractExportXml(archivePath: string): UnzipResult { return { xmlPath, - xmlBytes: xmlBytes.length, + xmlBytes, otherMembers, }; } @@ -260,8 +292,59 @@ export function readCentralDirectory(buf: Buffer): CentralDirectoryEntry[] { return entries; } -/** Extract the bytes of a single ZIP entry into a Buffer. */ -function extractEntry(buf: Buffer, entry: CentralDirectoryEntry): Buffer { +/** + * Split a Buffer into fixed-size slices (no copying — `subarray` views + * share the backing memory) so the inflate/write pipeline processes the + * entry in bounded chunks instead of one giant `write()` call. + */ +function* chunkBuffer(buf: Buffer, size: number): Generator { + for (let offset = 0; offset < buf.length; offset += size) { + yield buf.subarray(offset, Math.min(offset + size, buf.length)); + } +} + +/** + * Byte-counting passthrough that refuses to forward more than `limit` + * bytes — the streaming equivalent of `inflateRawSync`'s + * `maxOutputLength`, since the Transform-based zlib API exposes no such + * option. Trips on the actual bytes flowing through even when the + * central-directory metadata lied about the uncompressed size. Exported + * so a unit test can exercise the cap directly against a small limit — + * the real `MAX_DECOMPRESSED_BYTES` ceiling (8 GiB) is not practical to + * trip from a test fixture. + */ +export function createByteCap(limit: number): Transform { + let total = 0; + return new Transform({ + transform(chunk: Buffer, _enc, callback) { + total += chunk.length; + if (total > limit) { + callback( + new Error( + `Decompressed export.xml exceeds the ${limit}-byte cap` + + " — refusing as a suspected zip bomb.", + ), + ); + return; + } + callback(null, chunk); + }, + }); +} + +/** + * Stream a single ZIP entry's bytes out to `destPath`, decompressing + * through `zlib.createInflateRaw()` when needed. Runs entirely off the + * JS main thread (stream I/O + the libuv-threadpool-backed zlib + * binding), so a multi-GB member no longer blocks the event loop the + * way the old `inflateRawSync()` + `writeFileSync()` pair did. Returns + * the number of bytes written. + */ +async function streamEntryToFile( + buf: Buffer, + entry: CentralDirectoryEntry, + destPath: string, +): Promise { const local = entry.localHeaderOffset; if (buf.readUInt32LE(local) !== LOCAL_FILE_HEADER) { throw new Error( @@ -271,18 +354,32 @@ function extractEntry(buf: Buffer, entry: CentralDirectoryEntry): Buffer { 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); + const compressed = buf.subarray(dataStart, dataStart + entry.compressedSize); - if (entry.compressionMethod === 0) { - return compressed; + const source = Readable.from(chunkBuffer(compressed, STREAM_CHUNK_BYTES)); + const dest = createWriteStream(destPath); + const cap = createByteCap(MAX_DECOMPRESSED_BYTES); + + try { + if (entry.compressionMethod === 0) { + await pipeline(source, cap, dest); + } else { + // Method 8 — DEFLATE, streamed through the async zlib Transform. + await pipeline(source, createInflateRaw(), cap, dest); + } + } catch (err) { + // A mid-stream failure (byte cap trip, corrupt deflate stream) can + // leave a partial file on disk — the old sync path only ever wrote + // once inflate had already fully succeeded. Clean up so a rejected + // archive doesn't leak a partial `/tmp` file. + try { + unlinkSync(destPath); + } catch { + // ignore — best-effort + } + throw err; } - // Method 8 — DEFLATE. `maxOutputLength` instructs `node:zlib` to - // refuse expansion past the cap with a thrown error, defeating - // zip-bomb amplification even when the central-directory metadata - // lies about the uncompressed size. - return inflateRawSync(compressed, { - maxOutputLength: MAX_DECOMPRESSED_BYTES, - }); + return statSync(destPath).size; } /** diff --git a/src/lib/jobs/__tests__/apple-health-import-reconcile.test.ts b/src/lib/jobs/__tests__/apple-health-import-reconcile.test.ts new file mode 100644 index 000000000..13e240628 --- /dev/null +++ b/src/lib/jobs/__tests__/apple-health-import-reconcile.test.ts @@ -0,0 +1,186 @@ +/** + * v1.32.1 (issue #588) — periodic orphan-`ImportJob` reconcile. + * + * Before this change `reconcileOrphanImportJobs()` only ran once, at + * worker boot. A worker that crashed mid-extraction (OOM, restart) and + * came back up BEFORE the stuck row's heartbeat went stale (30 min) or + * pg-boss stopped reporting the backing job as live left that row + * stuck in `unpacking` / `parsing` / `upserting` forever — an import + * that never leaves "Unpacking the archive… 0 rows imported" with no + * failure ever surfaced, because nothing re-evaluated the row after + * that one boot-time pass. These tests cover both the periodic-tick + * wiring (queue provisioned, cron scheduled, handler bound) and the + * reconcile logic itself (a genuinely-abandoned row flips to `failed`; + * a row another worker is still actively running does not). + */ +import { readFileSync } from "node:fs"; +import { join } from "node:path"; +import { beforeEach, describe, expect, it, vi } from "vitest"; + +const mocks = vi.hoisted(() => ({ + findMany: vi.fn(), + updateMany: vi.fn(), + getGlobalBoss: vi.fn(), + getJobById: vi.fn(), +})); + +vi.mock("@/lib/db", () => ({ + prisma: { + importJob: { + findMany: mocks.findMany, + updateMany: mocks.updateMany, + }, + }, + toJson: (value: unknown) => value, +})); + +vi.mock("@/lib/jobs/boss-instance", () => ({ + getGlobalBoss: mocks.getGlobalBoss, +})); + +import { + APPLE_HEALTH_IMPORT_PARSER_REVISION, + APPLE_HEALTH_IMPORT_V2_QUEUE, + IMPORT_JOB_RECONCILE_CRON, + IMPORT_JOB_RECONCILE_QUEUE, + handleImportJobReconcileTick, + reconcileOrphanImportJobs, +} from "../apple-health-import-worker"; + +const maintenanceSource = readFileSync( + join(process.cwd(), "src/lib/jobs/reminder/register-maintenance.ts"), + "utf8", +); + +beforeEach(() => { + vi.resetAllMocks(); + mocks.getGlobalBoss.mockReturnValue({ getJobById: mocks.getJobById }); + mocks.updateMany.mockResolvedValue({ count: 0 }); +}); + +describe("periodic reconcile — wiring", () => { + it("provisions the queue, schedules a 15-minute cron, and binds the handler", () => { + expect(IMPORT_JOB_RECONCILE_QUEUE).toBe("apple-health-import-reconcile"); + expect(IMPORT_JOB_RECONCILE_CRON).toBe("*/15 * * * *"); + + const allQueues = maintenanceSource.match( + /const allQueues\s*=\s*\[([\s\S]*?)\];/, + ); + expect(allQueues).not.toBeNull(); + expect(allQueues![1]).toMatch(/\bIMPORT_JOB_RECONCILE_QUEUE\b/); + + const schedules = maintenanceSource.match( + /const schedules[\s\S]*?=\s*\[([\s\S]*?)\];/, + ); + expect(schedules).not.toBeNull(); + expect(schedules![1]).toMatch( + /\[IMPORT_JOB_RECONCILE_QUEUE,\s*IMPORT_JOB_RECONCILE_CRON\]/, + ); + + expect(maintenanceSource).toMatch( + /boss\.work\(\s*IMPORT_JOB_RECONCILE_QUEUE[\s\S]{0,120}handleImportJobReconcileTick/, + ); + }); +}); + +describe("reconcileOrphanImportJobs", () => { + it("flips a row to failed once its heartbeat has gone stale", async () => { + mocks.findMany.mockResolvedValue([ + { + id: "import-stale", + pgBossJobId: "boss-1", + updatedAt: new Date(Date.now() - 31 * 60 * 1000), + }, + ]); + mocks.getJobById.mockResolvedValue({ state: "active" }); + + await reconcileOrphanImportJobs(); + + expect(mocks.updateMany).toHaveBeenCalledWith({ + where: { id: { in: ["import-stale"] } }, + data: { + status: "failed", + failureReason: "interrupted_by_restart", + completedAt: expect.any(Date), + }, + }); + }); + + it("leaves a row alone while its heartbeat is fresh and pg-boss reports the job active", async () => { + mocks.findMany.mockResolvedValue([ + { + id: "import-live", + pgBossJobId: "boss-2", + updatedAt: new Date(Date.now() - 2 * 60 * 1000), + }, + ]); + mocks.getJobById.mockResolvedValue({ state: "active" }); + + await reconcileOrphanImportJobs(); + + expect(mocks.updateMany).not.toHaveBeenCalled(); + expect(mocks.getJobById).toHaveBeenCalledWith( + APPLE_HEALTH_IMPORT_V2_QUEUE, + "boss-2", + ); + }); + + it("flips a row whose backing pg-boss job is gone even with a fresh heartbeat", async () => { + mocks.findMany.mockResolvedValue([ + { + id: "import-gone", + pgBossJobId: "boss-3", + updatedAt: new Date(Date.now() - 60 * 1000), + }, + ]); + mocks.getJobById.mockResolvedValue(null); + + await reconcileOrphanImportJobs(); + + expect(mocks.updateMany).toHaveBeenCalledWith({ + where: { id: { in: ["import-gone"] } }, + data: { + status: "failed", + failureReason: "interrupted_by_restart", + completedAt: expect.any(Date), + }, + }); + }); + + it("only ever scopes the candidate query to the current parser revision", async () => { + mocks.findMany.mockResolvedValue([]); + + await reconcileOrphanImportJobs(); + + expect(mocks.findMany).toHaveBeenCalledWith({ + where: { + parserRevision: APPLE_HEALTH_IMPORT_PARSER_REVISION, + status: { in: ["unpacking", "parsing", "upserting"] }, + }, + select: { id: true, pgBossJobId: true, updatedAt: true }, + }); + }); +}); + +describe("handleImportJobReconcileTick", () => { + it("delegates to reconcileOrphanImportJobs and resolves on success", async () => { + mocks.findMany.mockResolvedValue([]); + + await expect( + handleImportJobReconcileTick([] as never), + ).resolves.toBeUndefined(); + expect(mocks.findMany).toHaveBeenCalledTimes(1); + }); + + it("swallows a reconcile failure instead of throwing", async () => { + mocks.findMany.mockRejectedValue(new Error("db unavailable")); + + // A background sweep this cheap must never spam pg-boss's + // retry/backoff machinery — it just re-runs on the next 15-minute + // tick. The handler must resolve even when the underlying reconcile + // pass throws. + await expect( + handleImportJobReconcileTick([] as never), + ).resolves.toBeUndefined(); + }); +}); diff --git a/src/lib/jobs/apple-health-import-worker.ts b/src/lib/jobs/apple-health-import-worker.ts index 662b144c2..f3919d5bf 100644 --- a/src/lib/jobs/apple-health-import-worker.ts +++ b/src/lib/jobs/apple-health-import-worker.ts @@ -37,6 +37,52 @@ import { type ImportJobResult, } from "@/lib/measurements/import-apple-health-export"; import { recomputeUserRollups } from "@/lib/rollups/measurement-rollups"; +import { withBackgroundEvent } from "@/lib/logging/background"; + +/** + * Queue + cron for the periodic orphan-ImportJob sweep. v1.32.1 + * (issue #588) — `reconcileOrphanImportJobs()` used to run only once, + * at worker boot. `restart: unless-stopped` in `docker-compose.yml` + * DOES bring a crashed/OOM-killed worker back up, so the boot-time + * pass usually catches a dead run — but only when that restart lands + * AFTER the row's heartbeat has already gone stale (30 minutes) or + * pg-boss no longer reports the backing job as live. A fast restart + * lands neither condition true yet, and because the pass runs + * exactly once per boot, a worker that then stays up and healthy + * NEVER re-evaluates that row — it is stuck in `unpacking` / + * `parsing` / `upserting` with "0 rows imported" forever and no + * failure ever surfaced to the status poll. A 15-minute cron closes + * that gap: within two ticks of the heartbeat going stale, the same + * idempotent reconcile flips the row to `failed` with an honest + * reason, so the UI's "Unpacking the archive…" spinner eventually + * turns into a visible, actionable error instead of an indefinite + * silent stall. + */ +export const IMPORT_JOB_RECONCILE_QUEUE = "apple-health-import-reconcile"; +export const IMPORT_JOB_RECONCILE_CRON = "*/15 * * * *"; + +/** + * Cron handler for the periodic orphan-ImportJob sweep. Delegates to + * the same `reconcileOrphanImportJobs()` the boot path calls — the + * heartbeat + live pg-boss-job check already make it safe to re-run + * on a healthy worker (a live import's fresh heartbeat and `active` + * pg-boss state both keep its row untouched). Errors are logged, not + * rethrown — a reconcile miss must never spam pg-boss's retry/backoff + * machinery for a background sweep this cheap to just re-run in 15 + * minutes. + */ +export async function handleImportJobReconcileTick( + jobs: Job[], +): Promise { + void jobs; + await withBackgroundEvent("job.apple_health_import_reconcile", async (evt) => { + try { + await reconcileOrphanImportJobs(); + } catch (err) { + evt.addWarning(`apple-health-import-reconcile failed: ${err}`); + } + }); +} /** Queue name isolated from revision-1 workers during rolling deployments. */ export const APPLE_HEALTH_IMPORT_V2_QUEUE = "apple-health-import-v2"; @@ -260,7 +306,7 @@ export async function handleAppleHealthImport( elapsedMs: 0, }); - const unzip = extractExportXml(uploadPath); + const unzip = await extractExportXml(uploadPath); extractedXmlPath = unzip.xmlPath; // Phase 2 + 3: parsing + upserting (the parser tracks both diff --git a/src/lib/jobs/reminder/register-maintenance.ts b/src/lib/jobs/reminder/register-maintenance.ts index 198784ad9..7008e2e90 100644 --- a/src/lib/jobs/reminder/register-maintenance.ts +++ b/src/lib/jobs/reminder/register-maintenance.ts @@ -41,7 +41,10 @@ import { APPLE_HEALTH_IMPORT_V2_QUEUE, APPLE_HEALTH_IMPORT_LEGACY_QUEUE, APPLE_HEALTH_IMPORT_CONCURRENCY, + IMPORT_JOB_RECONCILE_QUEUE, + IMPORT_JOB_RECONCILE_CRON, handleAppleHealthImport, + handleImportJobReconcileTick, migrateLegacyAppleHealthImport, type AppleHealthImportPayload, } from "@/lib/jobs/apple-health-import-worker"; @@ -314,6 +317,11 @@ const allQueues = [ INTAKE_AUTO_SKIP_QUEUE, APPLE_HEALTH_IMPORT_V2_QUEUE, APPLE_HEALTH_IMPORT_LEGACY_QUEUE, + // v1.32.1 (issue #588) — periodic sweep for ImportJob rows orphaned by a + // worker crash/restart mid-run. Without this entry the 15-minute cron + // below silently no-ops and a stuck "unpacking"/"parsing"/"upserting" row + // that survives past the next worker boot is never revisited. + IMPORT_JOB_RECONCILE_QUEUE, MEDICATION_INTAKE_IMPORT_QUEUE, // v1.8.2 — one-time duplicate dose-slot cleanup. Boot discovery enqueues // one job per user holding two live intake rows that snap to the same @@ -462,6 +470,13 @@ const schedules: ScheduleEntry[] = [ // Document vault — daily 04:10 Europe/Berlin purge for tombstoned // documents past the 30-day undo grace. [DOCUMENT_PURGE_QUEUE, DOCUMENT_PURGE_CRON], + // v1.32.1 (issue #588) — every-15-minute orphan-ImportJob sweep. Re-runs + // the same reconcile the boot path uses, so a stuck "unpacking" row + // whose worker crashed/restarted without the boot-time pass catching it + // (heartbeat not yet stale, pg-boss job not yet expired) still converges + // to a visible `failed` state within two ticks instead of staying stuck + // forever on an otherwise-healthy worker. + [IMPORT_JOB_RECONCILE_QUEUE, IMPORT_JOB_RECONCILE_CRON], ]; /** @@ -706,6 +721,14 @@ export async function registerMaintenanceQueues( } }, ); + // v1.32.1 (issue #588) — periodic orphan-ImportJob sweep. Single-flight: + // the underlying `updateMany` is idempotent and two ticks racing the same + // read-then-write pass would just be wasted work. + await boss.work( + IMPORT_JOB_RECONCILE_QUEUE, + { localConcurrency: 1 }, + handleImportJobReconcileTick, + ); await boss.work( MEDICATION_INTAKE_IMPORT_QUEUE, { diff --git a/src/lib/multipart/__tests__/stream-to-disk.test.ts b/src/lib/multipart/__tests__/stream-to-disk.test.ts index 890b46319..42f8329c3 100644 --- a/src/lib/multipart/__tests__/stream-to-disk.test.ts +++ b/src/lib/multipart/__tests__/stream-to-disk.test.ts @@ -431,7 +431,7 @@ describe("streamMultipartToDisk — large multi-chunk binary round-trip", () => expect(written.equals(zip), `zip byte-exact for ${sizes}`).toBe(true); // The load-bearing assertion: the saved file is still a parseable // archive (EOCD intact) and the member round-trips. - const extracted = extractExportXml(result.filePath); + const extracted = await extractExportXml(result.filePath); xmlPath = extracted.xmlPath; expect(readFileSync(extracted.xmlPath).equals(xmlBuf)).toBe(true); } finally { From fb5033545955a789a9a379c6ffd714812d9bff0e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Thu, 23 Jul 2026 09:38:03 +0200 Subject: [PATCH 2/4] fix(insights): stop the pulse page from silently swapping to resting heart rate MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Closes #584. `/insights/pulse` computed hasRestingHr from RESTING_HEART_RATE presence and used it to swap the ENTIRE primary chart and the Coach read strip to that series whenever any resting row existed, while the title, stat strip, capture action, and "show all values" link kept saying PULSE. A user could see a resting-only chart under a page that claimed to be pulse throughout. The chart now always includes PULSE as its first series; when resting data exists, RESTING_HEART_RATE renders as a second, clearly-labelled series (the same multi-series pattern the blood-pressure page uses for systolic/diastolic) instead of replacing it. The Coach read strip stays pinned to PULSE. The page also drops the resting-calibrated target band it used to paint behind the chart — that band is calibrated to steady-state resting readings and would misreport an expected workout spike in raw PULSE data as "out of target"; the personal target belongs on the dedicated resting-pulse page against the clean resting series. --- .../insights/pulse/__tests__/page.test.tsx | 144 ++++++++++++++++++ src/app/insights/pulse/page.tsx | 91 ++++------- .../import/__tests__/unzip-export-xml.test.ts | 4 +- 3 files changed, 176 insertions(+), 63 deletions(-) create mode 100644 src/app/insights/pulse/__tests__/page.test.tsx diff --git a/src/app/insights/pulse/__tests__/page.test.tsx b/src/app/insights/pulse/__tests__/page.test.tsx new file mode 100644 index 000000000..13e2432ff --- /dev/null +++ b/src/app/insights/pulse/__tests__/page.test.tsx @@ -0,0 +1,144 @@ +import { describe, expect, it, vi, beforeEach } from "vitest"; +import { renderToStaticMarkup } from "react-dom/server"; + +/** + * v1.32.1 (issue #584) — `/insights/pulse` must chart the metric its own + * title, capture action, and stat strip claim. `hasRestingHr` used to swap + * the ENTIRE primary chart + Coach read strip to RESTING_HEART_RATE the + * moment any resting-HR row existed, so the page could show a resting-only + * series while every other surface (title, stat strip, capture action, + * "show all values") stayed labelled "Pulse" — a silent full swap. + * RESTING_HEART_RATE may now appear only as a second, clearly-labelled + * series alongside PULSE (the same multi-series pattern the blood-pressure + * page uses for systolic/diastolic) — never a swap. + * + * `SubPageShell` is stubbed to a probe recording its props (including the + * `children` array holding the chart element) so these assert the resolved + * chart/coach-strip props rather than rendered text — responsive classes + * have broken text-based queries in this repo. + */ + +const shellProps = vi.fn(); +vi.mock("@/components/insights/sub-page-shell", () => ({ + SubPageShell: (props: Record) => { + shellProps(props); + return null; + }, +})); + +vi.mock("@/lib/i18n/context", () => ({ + useTranslations: () => ({ + t: (key: string) => key, + }), +})); + +vi.mock("@/hooks/use-auth", () => ({ + useAuth: () => ({ + isAuthenticated: true, + user: { dateOfBirth: null, gender: null, timezone: "UTC" }, + }), +})); + +const analytics = { + current: { summaries: {} as Record }, +}; +vi.mock("@/hooks/use-insights-analytics", () => ({ + useInsightsAnalytics: () => ({ + data: analytics.current, + isLoading: false, + isEmpty: false, + error: null, + refetch: () => {}, + }), +})); + +// `useInsightsLayoutPrefs` is the one real hook left that reaches +// `@tanstack/react-query`; stub `useQuery` so the page renders without a +// `QueryClientProvider` ancestor (mirrors the coach-page-launch-params test). +vi.mock("@tanstack/react-query", () => ({ + useQuery: () => ({ data: undefined, isLoading: false, error: null }), +})); + +import InsightsPulsPage from "../page"; + +interface CapturedShellProps { + captureType?: string; + showAllValuesType?: string; + coachReadStrip?: { props?: Record }; + children?: unknown; +} + +function renderPage( + summaries: Record, +): CapturedShellProps { + analytics.current = { summaries }; + renderToStaticMarkup(); + const calls = shellProps.mock.calls; + return calls[calls.length - 1][0] as CapturedShellProps; +} + +/** Locate the primary `` element among the page's + * children by its distinctive `types` array prop. Never rendered — the + * `SubPageShell` stub does not paint `children` — so this reads the plain + * React element object the page constructed. */ +function findChartElement( + children: unknown, +): { props: { types?: string[] } } | undefined { + const arr = Array.isArray(children) ? children : [children]; + return arr.find( + (el): el is { props: { types?: string[] } } => + Boolean(el) && + typeof el === "object" && + el !== null && + "props" in el && + Array.isArray((el as { props?: { types?: unknown } }).props?.types), + ); +} + +describe("/insights/pulse page — chart identity (issue #584)", () => { + beforeEach(() => { + shellProps.mockClear(); + }); + + it("charts PULSE as the primary (first) series even when resting-HR data exists", () => { + const props = renderPage({ + PULSE: { count: 50 }, + RESTING_HEART_RATE: { count: 30 }, + }); + const chart = findChartElement(props.children); + expect(chart).toBeDefined(); + expect(chart!.props.types).toEqual(["PULSE", "RESTING_HEART_RATE"]); + }); + + it("never swaps the chart to a RESTING_HEART_RATE-only series", () => { + const props = renderPage({ + PULSE: { count: 50 }, + RESTING_HEART_RATE: { count: 30 }, + }); + const chart = findChartElement(props.children); + expect(chart!.props.types).not.toEqual(["RESTING_HEART_RATE"]); + }); + + it("charts PULSE alone when there is no resting-HR data", () => { + const props = renderPage({ PULSE: { count: 50 } }); + const chart = findChartElement(props.children); + expect(chart!.props.types).toEqual(["PULSE"]); + }); + + it("keeps the Coach read strip pinned to PULSE regardless of resting-HR data", () => { + const props = renderPage({ + PULSE: { count: 50 }, + RESTING_HEART_RATE: { count: 30 }, + }); + expect(props.coachReadStrip?.props?.metricType).toBe("PULSE"); + }); + + it("keeps captureType and showAllValuesType pinned to PULSE", () => { + const props = renderPage({ + PULSE: { count: 50 }, + RESTING_HEART_RATE: { count: 30 }, + }); + expect(props.captureType).toBe("PULSE"); + expect(props.showAllValuesType).toBe("PULSE"); + }); +}); diff --git a/src/app/insights/pulse/page.tsx b/src/app/insights/pulse/page.tsx index 0b1b1d313..1f63ab298 100644 --- a/src/app/insights/pulse/page.tsx +++ b/src/app/insights/pulse/page.tsx @@ -35,20 +35,23 @@ const IntradayPulseChart = dynamic( })), { ssr: false, loading: () => }, ); -import { - getAgeFromDateOfBirth, - getPersonalizedPulseTarget, -} from "@/lib/analytics/pulse-targets"; /** * v1.4.25 W4 — `/insights/pulse`. * - * Routed Pulse sub-page. Renders the pulse chart with the personalized - * Karvonen-derived target band plus the per-section AI assessment. - * Note: `chartKey="pulse"` so the chart-cog can override the - * comparison-overlay independently from the dashboard pulse card; the - * MeasurementType filter is `PULSE` (the same field used elsewhere - * in the codebase). + * Routed Pulse sub-page. Renders the raw/workout-inclusive pulse chart + * plus the per-section AI assessment. Note: `chartKey="pulse"` so the + * chart-cog can override the comparison-overlay independently from the + * dashboard pulse card; the MeasurementType filter is `PULSE` (the same + * field used elsewhere in the codebase). + * + * v1.32.1 (issue #584) — this page no longer paints the personalized + * resting-pulse target band (`getPersonalizedPulseTarget()`, CDC/NCHS + * resting-pulse percentiles) behind the chart: that band is calibrated + * to steady-state resting readings and would misreport an expected + * workout spike in the raw PULSE series as "out of target". The + * personal target + its band belong on the dedicated + * `/insights/resting-pulse` page, against the clean resting series. * * Analytics fetch + empty-state branch consume `useInsightsAnalytics()` * + ``. VO₂ max is a cardio-fitness metric, so the page @@ -66,11 +69,16 @@ export default function InsightsPulsPage() { // dedicated cardio-fitness page. const hasVo2 = (analytics?.summaries?.VO2_MAX?.count ?? 0) > 0; const pulseSummary = analytics?.summaries?.PULSE ?? null; - // v1.15.12 A2 — the resting-pulse band is judged against - // RESTING_HEART_RATE (Apple's clean daily resting figure). When the - // user has resting rows the chart shows that series against the band; - // otherwise it charts raw heart rate WITHOUT the resting-band overlay - // (which would flag expected-high workout HR as "outside target"). + // v1.32.1 (issue #584) — this page is titled, captured, and stat-stripped + // as PULSE throughout; the chart must chart PULSE too. `hasRestingHr` used + // to swap the ENTIRE primary chart + Coach read to RESTING_HEART_RATE + // whenever any resting row existed, which silently showed a different + // metric than the page's own title/stat-strip/capture action claimed. + // RESTING_HEART_RATE now only ever appears as a second, clearly-labelled + // series alongside PULSE (same multi-series pattern the blood-pressure + // page uses for systolic/diastolic) — never a full swap. The dedicated + // `/insights/resting-pulse` page owns the resting-only view + its target + // band. const hasRestingHr = (analytics?.summaries?.RESTING_HEART_RATE?.count ?? 0) > 0; @@ -108,44 +116,6 @@ export default function InsightsPulsPage() { ); } - const pulseAge = getAgeFromDateOfBirth(user?.dateOfBirth ?? null); - const pulseTarget = getPersonalizedPulseTarget( - pulseAge, - (user?.gender as "MALE" | "FEMALE" | null | undefined) ?? null, - ); - const pulseBands = [ - { - min: 30, - max: pulseTarget.orangeMin, - color: "var(--destructive)", - opacity: 0.16, - }, - { - min: pulseTarget.orangeMin, - max: pulseTarget.greenMin, - color: "var(--warning)", - opacity: 0.18, - }, - { - min: pulseTarget.greenMin, - max: pulseTarget.greenMax, - color: "var(--success)", - opacity: 0.2, - }, - { - min: pulseTarget.greenMax, - max: pulseTarget.orangeMax, - color: "var(--warning)", - opacity: 0.18, - }, - { - min: pulseTarget.orangeMax, - max: 220, - color: "var(--destructive)", - opacity: 0.16, - }, - ].filter((band) => band.max > band.min); - return ( } coachReadStrip={ - + } diversityNudge={ { Readable.from([Buffer.from("hello"), Buffer.from("world")]), createByteCap(6), async function drain(source) { - for await (const _chunk of source) { - /* drain */ + for await (const chunk of source) { + void chunk; } }, ), From 6715c4a69ae1e7e5f584508156e14c84f43689ff Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Thu, 23 Jul 2026 09:55:23 +0200 Subject: [PATCH 3/4] observability(intraday-pulse): surface per-shape row counts and HR-bucket write outcomes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Related to #585. Investigated the intraday pulse chart's read path end to end: the ZIP-import raw-sample fold, the live-sync uploaded 10-min bucket fold, their merge, and the chart's render logic. All of it treats the two ingestion shapes consistently — the chart has no shape awareness at all, it paints whatever bucket series it receives. No read-side or display-side defect explains a systematic density difference between the two ingestion paths. What was missing is visibility into the write and read sides that would let a future report actually be diagnosed instead of re-investigated from scratch. The batch endpoint's wide-event trace never broke a skip down by reason, and never distinguished an HR-bucket write from any other entry in the same request. The intraday route's own trace surfaced only the final bucket count, with no visibility into which shape (raw sample rows vs uploaded bucket rows vs folded hourly rows) a day's data actually came from. Both traces now carry that breakdown, so the next report of "the chart looks sparser after a given date" is answerable by grep instead of by re-deriving the whole read path again. --- .../__tests__/hr-bucket-observability.test.ts | 269 ++++++++++++++++++ src/app/api/measurements/batch/route.ts | 73 +++++ .../__tests__/intraday-pulse-io.test.ts | 59 ++++ src/lib/analytics/intraday-pulse-io.ts | 21 ++ 4 files changed, 422 insertions(+) create mode 100644 src/app/api/measurements/batch/__tests__/hr-bucket-observability.test.ts diff --git a/src/app/api/measurements/batch/__tests__/hr-bucket-observability.test.ts b/src/app/api/measurements/batch/__tests__/hr-bucket-observability.test.ts new file mode 100644 index 000000000..35300709e --- /dev/null +++ b/src/app/api/measurements/batch/__tests__/hr-bucket-observability.test.ts @@ -0,0 +1,269 @@ +/** + * v1.32.1 (issue #585) — wide-event observability for the go-forward + * aggregated 10-min heart-rate bucket wire contract. + * + * The intraday pulse chart reads exactly the `stats:HKQuantityTypeIdentifier + * HeartRate:<10-min-Z>` rows this batch route accepts (or rejects). A report + * of "the chart looks sparser after the ZIP-import cutoff" was previously + * undiagnosable from server logs: the baseline `measurement.batch.ingest` + * annotation carried only a total `skipped` count with no reason breakdown + * and no visibility into how many of a batch's entries specifically targeted + * an HR bucket. These tests pin the new `measurement.batch.hr-bucket` + * annotation (fires only when the batch actually carried an HR-bucket entry) + * and the `skipped_by_reason` breakdown on the baseline annotation. + */ +import { describe, expect, it, vi, beforeEach } from "vitest"; +import { NextRequest } from "next/server"; + +const mocks = vi.hoisted(() => ({ annotate: vi.fn() })); + +vi.mock("@/lib/db", () => ({ + prisma: { + user: { + update: vi.fn(), + }, + measurement: { + findMany: vi.fn(), + createManyAndReturn: vi.fn(), + updateMany: vi.fn(), + }, + $transaction: vi.fn(async (fn: unknown) => { + if (typeof fn === "function") { + // eslint-disable-next-line @typescript-eslint/no-explicit-any + return (fn as any)(prisma as unknown as { measurement: unknown }); + } + }), + }, +})); + +vi.mock("@/lib/logging/context", async (importOriginal) => { + const actual = await importOriginal(); + return { ...actual, annotate: mocks.annotate }; +}); + +vi.mock("@/lib/measurements/reconcile-external-measurement", () => ({ + reconcileExternalMeasurement: async ( + tx: { + measurement: { + findMany: (args: unknown) => Promise< + Array<{ + id?: string; + type?: string; + source?: string; + externalId?: string | null; + }> + >; + createManyAndReturn: (args: { + data: Record; + }) => Promise>>; + updateMany: (args: { + where: Record; + data: Record; + }) => Promise; + }; + }, + input: Record & { + userId: string; + type: string; + source: string; + externalId: string; + }, + options: { exactExternalMatch?: "update" | "duplicate" } = {}, + ) => { + const existing = await tx.measurement.findMany({}); + const exact = existing.find( + (row: { type?: string; source?: string; externalId?: string | null }) => + row.type === input.type && + row.source === input.source && + row.externalId === input.externalId, + ); + if (exact && options.exactExternalMatch !== "update") { + return { status: "duplicate", row: exact }; + } + if (exact) { + await tx.measurement.updateMany({ + where: { id: exact.id }, + data: { ...input, deletedAt: null }, + }); + return { status: "updated", row: { ...exact, ...input } }; + } + const [inserted] = await tx.measurement.createManyAndReturn({ + data: input, + }); + if (inserted) return { status: "inserted", row: inserted }; + if (options.exactExternalMatch === "update") { + await tx.measurement.updateMany({ where: {}, data: input }); + return { status: "updated", row: input }; + } + return { status: "duplicate" }; + }, +})); + +vi.mock("@/lib/auth/session", () => ({ getSession: vi.fn() })); +vi.mock("@/lib/auth/audit", () => ({ + auditLog: vi.fn().mockResolvedValue(undefined), +})); +vi.mock("@/lib/logging/transports", () => ({ emitIfSampled: vi.fn() })); +vi.mock("@/lib/db-compat", () => ({ + ensureDbCompatibility: vi.fn().mockResolvedValue(undefined), +})); +vi.mock("@/lib/rate-limit", () => ({ checkRateLimit: vi.fn() })); +vi.mock("@/lib/jobs/pr-detection", () => ({ + enqueuePrDetection: vi.fn().mockResolvedValue(undefined), +})); +vi.mock("@/lib/jobs/reminder-satisfy", () => ({ + enqueueReminderSatisfy: vi.fn().mockResolvedValue(undefined), +})); +vi.mock("@/lib/cache/invalidate", () => ({ + invalidateUserMeasurements: vi.fn(), +})); +vi.mock("@/lib/insights/comprehensive-generate", () => ({ + invalidateStatusInsightsForTypes: vi.fn().mockResolvedValue(undefined), +})); +vi.mock("@/lib/rollups/measurement-rollups", () => ({ + recomputeBucketsForMeasurement: vi.fn().mockResolvedValue(undefined), + collapseToTypeDayKeys: vi.fn(() => []), +})); + +vi.mock("next/headers", () => ({ + headers: vi.fn(async () => ({ get: () => null })), + cookies: vi.fn(async () => ({ + get: () => undefined, + set: () => {}, + delete: () => {}, + })), +})); + +import { POST } from "../route"; +import { prisma } from "@/lib/db"; +import { getSession } from "@/lib/auth/session"; +import { checkRateLimit } from "@/lib/rate-limit"; + +const SESSION_OK = { + session: { id: "sess-1", expiresAt: new Date(Date.now() + 3_600_000) }, + user: { id: "user-1", username: "testuser", role: "USER" as const }, +}; + +const FRESH_BUCKET_ID = + "stats:HKQuantityTypeIdentifierHeartRate:2026-06-21T14:00:00.000Z"; +const REPOST_BUCKET_ID = + "stats:HKQuantityTypeIdentifierHeartRate:2026-06-21T15:00:00.000Z"; +const MALFORMED_BUCKET_ID = + "stats:HKQuantityTypeIdentifierHeartRate:2026-06-21T14:35:00.000Z"; +const SAMPLE_ID = "C0FFEE00-0000-4000-8000-000000000000"; + +function makeRequest(body: unknown): NextRequest { + return new NextRequest("http://localhost/api/measurements/batch", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify(body), + }); +} + +function hrEntry(externalId: string, value: number) { + return { + hkIdentifier: "HKQuantityTypeIdentifierHeartRate", + value, + unit: "count/min", + startDate: "2026-06-21T14:00:00.000Z", + endDate: "2026-06-21T14:59:59.000Z", + externalId, + }; +} + +function annotateCallsFor(actionName: string) { + return mocks.annotate.mock.calls.filter( + ([arg]) => + (arg as { action?: { name?: string } })?.action?.name === actionName, + ); +} + +beforeEach(() => { + vi.clearAllMocks(); + vi.mocked(getSession).mockResolvedValue(SESSION_OK as never); + vi.mocked(checkRateLimit).mockResolvedValue({ + allowed: true, + remaining: 60, + resetAt: Date.now() + 60_000, + }); + vi.mocked(prisma.measurement.createManyAndReturn).mockImplementation((async ( + args: unknown, + ) => { + const { data } = args as { + data: + | { type: unknown; source: string; externalId: string | null } + | Array<{ type: unknown; source: string; externalId: string | null }>; + }; + const rows = Array.isArray(data) ? data : [data]; + return rows.map(({ type, source, externalId }) => ({ + type, + source, + externalId, + })); + }) as never); + vi.mocked(prisma.measurement.updateMany).mockResolvedValue({ count: 1 }); +}); + +describe("POST /api/measurements/batch — HR-bucket observability (issue #585)", () => { + it("reports a per-outcome breakdown covering only HR-bucket entries", async () => { + // REPOST_BUCKET_ID already exists — the reconcile mock overwrites it. + vi.mocked(prisma.measurement.findMany).mockResolvedValue([ + { + type: "PULSE", + source: "APPLE_HEALTH", + externalId: REPOST_BUCKET_ID, + }, + ] as never); + + const res = await POST( + makeRequest({ + entries: [ + hrEntry(FRESH_BUCKET_ID, 72), // inserted + hrEntry(REPOST_BUCKET_ID, 78), // updated (overwrite) + hrEntry(MALFORMED_BUCKET_ID, 70), // skipped (malformed) + hrEntry(SAMPLE_ID, 64), // NOT an HR bucket — must be excluded + ], + }), + ); + expect(res.status).toBe(200); + + const calls = annotateCallsFor("measurement.batch.hr-bucket"); + expect(calls).toHaveLength(1); + const meta = (calls[0][0] as { meta: Record }).meta; + expect(meta).toEqual({ + attempted: 3, + inserted: 1, + updated: 1, + duplicate: 0, + skipped: 1, + failed: 0, + processed: 4, + }); + }); + + it("never fires the HR-bucket annotation for a batch with no bucket entries", async () => { + vi.mocked(prisma.measurement.findMany).mockResolvedValue([]); + const res = await POST(makeRequest({ entries: [hrEntry(SAMPLE_ID, 64)] })); + expect(res.status).toBe(200); + expect(annotateCallsFor("measurement.batch.hr-bucket")).toHaveLength(0); + }); + + it("breaks the baseline ingest annotation's skip count down by reason", async () => { + vi.mocked(prisma.measurement.findMany).mockResolvedValue([]); + const res = await POST( + makeRequest({ + entries: [ + hrEntry(MALFORMED_BUCKET_ID, 70), // skipped: malformed_hr_bucket_id + ], + }), + ); + expect(res.status).toBe(200); + + const calls = annotateCallsFor("measurement.batch.ingest"); + expect(calls).toHaveLength(1); + const meta = calls[0][0] as { meta: { skipped_by_reason: unknown } }; + expect(meta.meta.skipped_by_reason).toEqual({ + malformed_hr_bucket_id: 1, + }); + }); +}); diff --git a/src/app/api/measurements/batch/route.ts b/src/app/api/measurements/batch/route.ts index b739ed6b4..a48cf1c66 100644 --- a/src/app/api/measurements/batch/route.ts +++ b/src/app/api/measurements/batch/route.ts @@ -603,6 +603,18 @@ async function postBatch(request: NextRequest): Promise { .filter((r) => r.status === "skipped") .map((r) => ({ index: r.index, reason: r.reason ?? "unknown" })); + // v1.32.1 (issue #585) — per-reason skip breakdown. The baseline + // annotation below only ever carried the total `skipped` count, so an + // operator investigating "why does this account's uploaded data look + // sparser than expected" had no way to see WHICH validation gate an + // entry tripped (malformed bucket id, out-of-range value, unmappable + // identifier, …) without pulling raw rows. Grouping by reason turns + // that into a one-glance wide-event field. + const skippedByReason: Record = {}; + for (const s of skipped) { + skippedByReason[s.reason] = (skippedByReason[s.reason] ?? 0) + 1; + } + await auditLog("measurement.batch.ingest", { userId: user.id, ipAddress: getClientIp(request), @@ -648,6 +660,64 @@ async function postBatch(request: NextRequest): Promise { }); } + // v1.32.1 (issue #585) — dedicated outcome breakdown for entries + // targeting the go-forward aggregated 10-min heart-rate bucket prefix + // (`stats:HKQuantityTypeIdentifierHeartRate:<10-min-Z>`). The intraday + // pulse chart reads exactly these rows for a live-synced day; a report + // of "the chart looks sparser after the ZIP import cutoff" is + // unanswerable from the baseline ingest trace alone, since it never + // distinguished an HR-bucket write from any other entry in the batch. + // Grepping `measurement.batch.hr-bucket` now shows, per sync, how many + // bucket entries the client sent and what happened to each — confirming + // (or ruling out) server-side rejection/dedup as the density-gap cause. + // Only fires when the batch actually carried at least one HR-bucket + // entry so the baseline trace is unchanged for every other sync. + let hrBucketInserted = 0; + let hrBucketUpdated = 0; + let hrBucketDuplicate = 0; + let hrBucketSkipped = 0; + let hrBucketFailed = 0; + for (let index = 0; index < entries.length; index++) { + if (!targetsAggregatedBucket(entries[index].externalId)) continue; + switch (results[index]?.status) { + case "inserted": + hrBucketInserted++; + break; + case "updated": + hrBucketUpdated++; + break; + case "duplicate": + hrBucketDuplicate++; + break; + case "skipped": + hrBucketSkipped++; + break; + case "failed": + hrBucketFailed++; + break; + } + } + const hrBucketAttempted = + hrBucketInserted + + hrBucketUpdated + + hrBucketDuplicate + + hrBucketSkipped + + hrBucketFailed; + if (hrBucketAttempted > 0) { + annotate({ + action: { name: "measurement.batch.hr-bucket" }, + meta: { + attempted: hrBucketAttempted, + inserted: hrBucketInserted, + updated: hrBucketUpdated, + duplicate: hrBucketDuplicate, + skipped: hrBucketSkipped, + failed: hrBucketFailed, + processed: entries.length, + }, + }); + } + // v1.4.25 W16c — kick off PR detection for this user. We always // enqueue when at least one row was written (or the batch had any // measurements to consider) so a single off-day reading still gets @@ -693,6 +763,9 @@ async function postBatch(request: NextRequest): Promise { updated: updatedCount, duplicates: duplicateCount, skipped: skipped.length, + // v1.32.1 (issue #585) — per-reason skip breakdown; see the + // `skippedByReason` build above. + skipped_by_reason: skippedByReason, failed: failedCount, }, }); diff --git a/src/lib/analytics/__tests__/intraday-pulse-io.test.ts b/src/lib/analytics/__tests__/intraday-pulse-io.test.ts index 867c0b48c..57450a5b4 100644 --- a/src/lib/analytics/__tests__/intraday-pulse-io.test.ts +++ b/src/lib/analytics/__tests__/intraday-pulse-io.test.ts @@ -13,6 +13,7 @@ import { describe, expect, it, vi, beforeEach } from "vitest"; const findManyMock = vi.fn(); const workoutFindManyMock = vi.fn(); +const { annotateMock } = vi.hoisted(() => ({ annotateMock: vi.fn() })); vi.mock("@/lib/db", () => ({ prisma: { @@ -21,6 +22,8 @@ vi.mock("@/lib/db", () => ({ }, })); +vi.mock("@/lib/logging/context", () => ({ annotate: annotateMock })); + import { loadIntradayPulse } from "@/lib/analytics/intraday-pulse-io"; import { resolveRestingPulseSeries, @@ -42,6 +45,7 @@ describe("loadIntradayPulse", () => { findManyMock.mockReset(); workoutFindManyMock.mockReset(); workoutFindManyMock.mockResolvedValue([]); + annotateMock.mockReset(); }); it("M2 — queries PULSE/step rows against the exact local-day window, not the padded lead/trail superset", async () => { @@ -209,4 +213,59 @@ describe("loadIntradayPulse", () => { { startMinute: 490, mean: 75, count: 2, min: 70, max: 81 }, ]); }); + + it("v1.32.1 (issue #585) — annotates the per-shape row/bucket breakdown for observability", async () => { + const tz = "UTC"; + const dateKey = "2026-06-15"; + + const restingRows = Array.from({ length: 20 }, (_, i) => ({ + value: 55, + measuredAt: new Date( + `2026-05-${String((i % 28) + 1).padStart(2, "0")}T06:00:00.000Z`, + ), + })); + // One uploaded 10-min bucket row and one raw per-sample row sharing the + // day — the DTO merges them (bucket overlays raw on a slot collision), + // but the annotation must still report each SHAPE's own row count, not + // just the merged bucket total. + const uploadedBucket = { + value: 72, + valueMin: 64, + valueMax: 88, + measuredAt: new Date("2026-06-15T08:09:00.000Z"), + externalId: + "stats:HKQuantityTypeIdentifierHeartRate:2026-06-15T08:00:00.000Z", + }; + const rawSample = { + value: 65, + valueMin: null, + valueMax: null, + measuredAt: new Date("2026-06-15T09:05:00.000Z"), + externalId: "sample:abc123", + }; + + findManyMock.mockImplementation( + async (args: { + where: { type: string; measuredAt?: unknown }; + take?: number; + }) => { + if (args.where.type === "RESTING_HEART_RATE") return restingRows; + if (args.where.type === "PULSE" && args.where.measuredAt) { + return [uploadedBucket, rawSample]; + } + return []; + }, + ); + + await loadIntradayPulse("user-1", tz, dateKey); + + expect(annotateMock).toHaveBeenCalledWith({ + meta: { + intraday_raw_sample_rows: 1, + intraday_uploaded_bucket_rows: 1, + intraday_folded_hourly_rows: 0, + intraday_ten_min_buckets: 2, + }, + }); + }); }); diff --git a/src/lib/analytics/intraday-pulse-io.ts b/src/lib/analytics/intraday-pulse-io.ts index 9a4df4243..ec7987b34 100644 --- a/src/lib/analytics/intraday-pulse-io.ts +++ b/src/lib/analytics/intraday-pulse-io.ts @@ -16,6 +16,7 @@ * digest's `tension_window` builder, so the signal is computed one way. */ import { prisma } from "@/lib/db"; +import { annotate } from "@/lib/logging/context"; import { percentile } from "@/lib/insights/strain-score"; import { resolveRestingPulseSeries } from "@/lib/analytics/resting-pulse"; import { localDayWindow } from "@/lib/measurements/consolidation-tz"; @@ -349,6 +350,26 @@ export async function loadIntradayPulse( }) : null; + // v1.32.1 (issue #585) — per-shape row/bucket breakdown. A report of + // "the chart looks sparser after the ZIP-import cutoff" is otherwise + // undiagnosable from server logs: the route's own annotation only ever + // surfaced the FINAL bucket count, with no visibility into which + // ingestion shape (raw ZIP-imported samples vs uploaded 10-min + // aggregates) the day's data actually came from, nor how many DB rows + // fed each shape before bucketing. Grepping `insights.pulse.intraday` + // now shows both — e.g. `uploaded_bucket_rows: 6` next to + // `ten_min_buckets: 6` confirms the uploaded shape IS being read and + // placed 1:1, narrowing a future report to "the client uploaded too few + // buckets" rather than leaving the read path itself as a live suspect. + annotate({ + meta: { + intraday_raw_sample_rows: rawPulseRows.length, + intraday_uploaded_bucket_rows: uploadedBucketRows.length, + intraday_folded_hourly_rows: foldedPulseRows.length, + intraday_ten_min_buckets: tenMinSeries.length, + }, + }); + return { dateKey, timezone, From ecf2d1a7d5ff1368453db6c261866bab9c7f6514 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Thu, 23 Jul 2026 10:01:09 +0200 Subject: [PATCH 4/4] style: run prettier over the import-worker reconcile addition --- src/lib/jobs/apple-health-import-worker.ts | 17 ++++++++++------- 1 file changed, 10 insertions(+), 7 deletions(-) diff --git a/src/lib/jobs/apple-health-import-worker.ts b/src/lib/jobs/apple-health-import-worker.ts index f3919d5bf..9bf197017 100644 --- a/src/lib/jobs/apple-health-import-worker.ts +++ b/src/lib/jobs/apple-health-import-worker.ts @@ -75,13 +75,16 @@ export async function handleImportJobReconcileTick( jobs: Job[], ): Promise { void jobs; - await withBackgroundEvent("job.apple_health_import_reconcile", async (evt) => { - try { - await reconcileOrphanImportJobs(); - } catch (err) { - evt.addWarning(`apple-health-import-reconcile failed: ${err}`); - } - }); + await withBackgroundEvent( + "job.apple_health_import_reconcile", + async (evt) => { + try { + await reconcileOrphanImportJobs(); + } catch (err) { + evt.addWarning(`apple-health-import-reconcile failed: ${err}`); + } + }, + ); } /** Queue name isolated from revision-1 workers during rolling deployments. */