diff --git a/CHANGELOG.md b/CHANGELOG.md index 9048aa07d..152d56968 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,45 @@ # Changelog +## [1.40.1] — 2026-10-04 + +Restore opens older encrypted values again; faster pages on a long-used +server; a day of pulse is the mean of its hours; Apple Health imports leave +out HealthLog's own samples; more memory for the app by default. + +### Fixed + +- **Restore key probe.** A legacy unprefixed ciphertext was read as a base64 + string and refused a restore the key opens; legacy values open as strings, + probes come only from sections that count, binary probes use their AAD. + The drill checks every account of the newest date. +- **Daily pulse (#1107, migration 0372).** A day is the mean of its local + hours' means and a window the mean of its days, in every reader; stored + pulse rollups are recomputed. Count, min and max stay over every reading. +- **Apple Health import (#1109).** Records carrying HealthLog's origin + marker or a HealthLog id are left out and counted; cycle samples too. +- **Blood pressure series.** One source per day for both values. +- **Backups.** Heartbeat on both backup jobs; per-account attempts recorded + (migration 0373), interrupted accounts go last; the preview is built from + the writer's counts. +- **Document previews.** A permanent render failure is recorded (migration 0374) instead of re-queued at every start. +- **Umami.** Events from a tracker built with a custom endpoint are proxied + to the configured Umami host. +- **Display.** Rest Mode shows its start date; the favicon is transparent; + the sidebar no longer flashes scrollbars while it expands. + +### Changed + +- **Performance.** One Prisma client per process instead of one per bundle; + the rollup coverage probe uses a loose index scan and concurrent readers + share one probe; timezone offsets are memoised per hour; chart text is + measured outside ``; Insights mounts lower sections near the + viewport; a nav prefetch is not re-sent on mount; MCP single-metric reads + build only the requested metric; the backup schema loads on first use. +- **Lab list (#1057).** The range bar and name use a wide card. +- **Compose.** `APP_MEMORY_LIMIT` defaults to `1536m` (792 MB heap). +- Dependencies: next 16.3.8, eslint-config-next 16.3.8, zod-openapi 6.0.2, + testcontainers 12.2.0, p-limit 7.3.3. + ## [1.40.0] — 2026-10-04 Back up the encryption key and refuse a key that cannot open the data; diff --git a/docs/api/openapi.yaml b/docs/api/openapi.yaml index bb71a2e8e..54c7d32f5 100644 --- a/docs/api/openapi.yaml +++ b/docs/api/openapi.yaml @@ -1,7 +1,7 @@ openapi: 3.1.0 info: title: HealthLog API - version: 1.40.0 + version: 1.40.1 description: >- Self-hosted personal-health-tracking PWA — public API surface for the iOS native client and external ingest. diff --git a/docs/self-hosting/getting-started.md b/docs/self-hosting/getting-started.md index 88b88f0eb..c55033afd 100644 --- a/docs/self-hosting/getting-started.md +++ b/docs/self-hosting/getting-started.md @@ -13,7 +13,12 @@ already installed, longer if a TLS-fronted public hostname is in scope unchanged. - **2 GB RAM, 10 GB free disk.** The Postgres data volume grows with measurement history; the Apple Health import worker can briefly - hold a 1.5 GB upload in `/tmp` while it parses. + hold a 1.5 GB upload in `/tmp` while it parses. The app container is + capped at `APP_MEMORY_LIMIT`, `1536m` by default (Node then sizes its + heap at about 792 MB). On a 2 GB host that leaves too little for + Postgres and the system, so set `APP_MEMORY_LIMIT=1g` in `.env` + (a heap of about 524 MB); a long-used account then runs closer to + that heap, and the weekly backup with it. - **Outbound HTTPS** to `ghcr.io` (image pulls) plus whichever integration endpoints you plan to enable (Withings, OpenAI, etc.). The opt-in environmental-context module additionally reaches diff --git a/docs/self-hosting/scaling.md b/docs/self-hosting/scaling.md index 8bde98d1b..6776d44a3 100644 --- a/docs/self-hosting/scaling.md +++ b/docs/self-hosting/scaling.md @@ -163,7 +163,11 @@ the knob as `DB_CONNECTION_LIMIT`, baked into `DATABASE_URL` as `connection_limit`; `DATABASE_POOL_MAX` remains a working explicit override. Resolution order: `DB_CONNECTION_LIMIT`, then `DATABASE_POOL_MAX`, then the `connection_limit` URL parameter, then -the default of 20. +the default of 20. In the bundled single-container setup +(`HEALTHLOG_PROCESS_TYPE=all`) the web requests and the background +worker run in one process and share that one Prisma pool, so a budget +sized for the web traffic alone leaves the worker's jobs queueing for +the same connections. Plan total Postgres slots as **container_count × budget**. A stock Postgres 16 container ships with `max_connections = 100`, so diff --git a/e2e/ai-optional-account.spec.ts b/e2e/ai-optional-account.spec.ts index 3e7f138fe..6a27a80cd 100644 --- a/e2e/ai-optional-account.spec.ts +++ b/e2e/ai-optional-account.spec.ts @@ -25,6 +25,7 @@ import { seedStoredBriefing, } from "./setup/ai-optional-fixture"; import { expect, test } from "./setup/test"; +import { revealDeferredSectionsSettled } from "./utils/deferred-sections"; test.use({ storageState: AI_OPTIONAL_STORAGE_STATE_PATH }); test.describe.configure({ mode: "serial" }); @@ -36,12 +37,18 @@ test.afterAll(async () => { await resetAiOptionalAccount(); }); +/** + * The overview with every section mounted. Sections past the first three + * render as an empty sentinel until they near the viewport, so a count of + * zero taken before they mount would pass whatever they paint. + */ async function openOverview(page: Page): Promise { await page.goto("/insights"); await expect(page.locator('[data-slot="insights-hero-strip"]')).toBeVisible({ timeout: 20_000, }); await page.waitForLoadState("networkidle"); + await revealDeferredSectionsSettled(page); } test("J5: withdrawing AI consent deletes the notes written under it", async ({ @@ -80,12 +87,12 @@ test("J4: the Coach memory stays readable and deletable with the Coach hidden", const title = `Stored conversation ${Date.now()}`; await seedHiddenCoachWithConversation(title); - // Hide Coach takes the Coach page and its launcher away... - await page.goto("/insights"); - await expect(page.locator('[data-slot="insights-hero-strip"]')).toBeVisible({ - timeout: 20_000, - }); + // Hide Coach takes the Coach page, its launcher and every per-card + // hand-off away... + await openOverview(page); await expect(page.locator('[data-slot="coach-fab"]')).toHaveCount(0); + await expect(page.locator('[data-slot="ask-coach-action"]')).toHaveCount(0); + await expect(page.locator('[data-slot="ask-coach-icon"]')).toHaveCount(0); // ...and Settings, AI lists what the Coach stored. await page.goto("/settings/ai"); diff --git a/e2e/ai-optional-overview.spec.ts b/e2e/ai-optional-overview.spec.ts index dac4533bd..ba0158dd0 100644 --- a/e2e/ai-optional-overview.spec.ts +++ b/e2e/ai-optional-overview.spec.ts @@ -21,9 +21,19 @@ import type { Page, Request, Response } from "@playwright/test"; import { STORAGE_STATE_PATH } from "./setup/global-setup"; import { aiBlockUnavailable, serveAiBlock } from "./setup/ai-capabilities"; import { expect, test } from "./setup/test"; +import { revealDeferredSectionsSettled } from "./utils/deferred-sections"; test.use({ storageState: STORAGE_STATE_PATH }); +/** + * Mount every overview section, then let their reads land. Sections past the + * first three render as an empty sentinel until they near the viewport, so a + * count of zero taken before this would pass whatever those sections paint. + */ +async function mountWholeOverview(page: Page): Promise { + await revealDeferredSectionsSettled(page); +} + /** Routes that call a model or serve model-written text. */ const AI_ROUTE = new RegExp( [ @@ -89,6 +99,7 @@ test.describe("J1: AI switched off by the operator", () => { ); // Let every section's reads land before judging what painted. await page.waitForLoadState("networkidle"); + await mountWholeOverview(page); // A read refused for want of AI paints one of these on the overview: // the section boundary, the error card, or the dense error row. @@ -137,6 +148,7 @@ test.describe("J3: no AI provider", () => { timeout: 20_000, }); await page.waitForLoadState("networkidle"); + await mountWholeOverview(page); await expect(page.locator('[data-slot="coach-fab"]')).toHaveCount(0); await expect(page.locator('[data-slot="daily-briefing"]')).toHaveCount(0); // The per-card Coach hand-offs paint only while the Coach is available. @@ -163,6 +175,7 @@ test.describe("J3: no AI provider", () => { { timeout: 20_000 }, ); await page.waitForLoadState("networkidle"); + await mountWholeOverview(page); await expect(page.locator('[data-slot="ai-setup-hint"]')).toHaveCount(0); expect(network.aiRequests).toEqual([]); diff --git a/e2e/utils/deferred-sections.ts b/e2e/utils/deferred-sections.ts index 0ca4748ea..247ab9d80 100644 --- a/e2e/utils/deferred-sections.ts +++ b/e2e/utils/deferred-sections.ts @@ -36,3 +36,40 @@ export async function revealDeferredSections(page: Page): Promise { .toBe(0); await page.evaluate(() => document.querySelector("main")?.scrollTo(0, 0)); } + +/** + * `revealDeferredSections`, then wait until the reads the newly mounted + * sections fired have landed: no request in flight for 750 ms. A + * `waitForLoadState("networkidle")` cannot do this: the page reached that + * state once already, so it resolves at once, and a count of zero taken then + * passes before a deferred section has painted anything. + */ +export async function revealDeferredSectionsSettled(page: Page): Promise { + let inFlight = 0; + const started = () => { + inFlight += 1; + }; + const ended = () => { + inFlight -= 1; + }; + page.on("request", started); + page.on("requestfinished", ended); + page.on("requestfailed", ended); + try { + await revealDeferredSections(page); + let quietSince = Date.now(); + await expect + .poll( + () => { + if (inFlight > 0) quietSince = Date.now(); + return Date.now() - quietSince >= 750; + }, + { timeout: 20_000, intervals: [100] }, + ) + .toBe(true); + } finally { + page.off("request", started); + page.off("requestfinished", ended); + page.off("requestfailed", ended); + } +} diff --git a/package.json b/package.json index 3655fa286..32ef4824a 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "healthlog", - "version": "1.40.0", + "version": "1.40.1", "description": "Self-hosted personal-health-tracking PWA with Withings integration, AI insights, and doctor-report PDF export.", "license": "PolyForm-Noncommercial-1.0.0", "homepage": "https://healthlog.dev", diff --git a/public/sw.js b/public/sw.js index 32351a83e..0c7bb5af9 100644 --- a/public/sw.js +++ b/public/sw.js @@ -36,7 +36,7 @@ try { // v1.4.38.4 → v1.4.42. Do not hand-edit; bump `package.json` and rebuild. const CACHE_VERSION = (typeof self !== "undefined" && self.__APP_VERSION__) || - /* @sw-version-fallback */ "v1.40.0"; + /* @sw-version-fallback */ "v1.40.1"; const STATIC_CACHE = `healthlog-static-${CACHE_VERSION}`; const PAGE_CACHE = `healthlog-pages-${CACHE_VERSION}`; // v1.18.6 — read-only data cache for a curated allowlist of safe GET `/api/*` diff --git a/src/lib/apple-health/own-origin.ts b/src/lib/apple-health/own-origin.ts index 34f6f2f8c..1a60a05aa 100644 --- a/src/lib/apple-health/own-origin.ts +++ b/src/lib/apple-health/own-origin.ts @@ -4,9 +4,13 @@ * from the app, and any Withings, import or manual row the server-to-Health * mirror writes. Builds from before the marker existed wrote the id alone. * - * Such a sample is already in the database under its own source. When the same - * sample comes back through an Apple Health export it is a second copy of that - * reading, so the export import leaves it out. + * Such a sample is usually already in the database under its own source. When + * it comes back through an Apple Health export it is a second copy of that + * reading, so the export import leaves it out, but only once its + * `HKExternalUUID` names a row of the importing account (a cycle sample: once + * that account holds a day-log on its day). The marker alone says HealthLog + * wrote the sample, not that this instance still holds it: after a move to a + * new instance without a backup, the export is how those values come back. * * This is not the predicate the app's own sync uses. The app decides "our own * echo" from `HKExternalUUID` plus the sample's authoring source being the app diff --git a/src/lib/export/__tests__/backup-preview.test.ts b/src/lib/export/__tests__/backup-preview.test.ts index 8016c6c4c..b82ac0b6f 100644 --- a/src/lib/export/__tests__/backup-preview.test.ts +++ b/src/lib/export/__tests__/backup-preview.test.ts @@ -101,8 +101,8 @@ describe("createBackupPreviewCollector", () => { const expectedKeys = new BackupKeyIdCollector(); expectedKeys.visit(FILE); - it("says what a full read of the copy says", () => { - const collector = createBackupPreviewCollector(); + it("says what a full read of the copy says", async () => { + const collector = await createBackupPreviewCollector(); observeAsTheWriterDoes(FILE, collector.observe); const result = collector.finish(countsOf(FILE)); expect(result).not.toBeNull(); @@ -112,8 +112,8 @@ describe("createBackupPreviewCollector", () => { expect(result!.keys.keyIds()).toEqual(["v3"]); }); - it("answers null for a copy that does not name its version, owner and date", () => { - const collector = createBackupPreviewCollector(); + it("answers null for a copy that does not name its version, owner and date", async () => { + const collector = await createBackupPreviewCollector(); const { userId, ...rest } = FILE; void userId; observeAsTheWriterDoes(rest, collector.observe); @@ -121,6 +121,64 @@ describe("createBackupPreviewCollector", () => { }); }); +describe("createBackupPreviewCollector against the schema", () => { + // A copy the schema refuses must not be described as restorable: a null + // preview sends the preview route to read the copy, which says why not. + async function previewOf(file: Record) { + const collector = await createBackupPreviewCollector(); + observeAsTheWriterDoes(file, collector.observe); + return collector.finish(countsOf(FILE)); + } + + it("answers null when a bulk row violates the schema", async () => { + const measurements = FILE.measurements.map((m, i) => + i === 1 ? { ...m, measuredAt: "not a date" } : m, + ); + expect(await previewOf({ ...FILE, measurements })).toBeNull(); + }); + + it("answers null when a whole section violates the schema", async () => { + const labResults = [{ ...FILE.labResults[0], takenAt: 42 }]; + expect(await previewOf({ ...FILE, labResults })).toBeNull(); + }); + + it("answers null when the header does not parse", async () => { + expect( + await previewOf({ ...FILE, exportedAt: "yesterday-ish" }), + ).toBeNull(); + }); + + it("answers null for a canonical copy with a measurement without an id", async () => { + const measurements = FILE.measurements.map((m, i) => + i === 0 ? { ...m, id: undefined } : m, + ); + expect(await previewOf({ ...FILE, measurements })).toBeNull(); + }); + + it("agrees with the whole-document parse on every case above", () => { + // The oracle: each refused file is refused by the full schema too. + const bad = [ + { + ...FILE, + measurements: FILE.measurements.map((m, i) => + i === 1 ? { ...m, measuredAt: "not a date" } : m, + ), + }, + { ...FILE, labResults: [{ ...FILE.labResults[0], takenAt: 42 }] }, + { ...FILE, exportedAt: "yesterday-ish" }, + { + ...FILE, + measurements: FILE.measurements.map((m, i) => + i === 0 ? { ...m, id: undefined } : m, + ), + }, + ]; + for (const file of bad) { + expect(backupPayloadSchema.safeParse(file).success).toBe(false); + } + }); +}); + describe("storedPreviewFor", () => { const chunked = { data: null, chunkCount: 4, chunkStreamId: "s1" }; const preview = () => diff --git a/src/lib/export/backup-preview.ts b/src/lib/export/backup-preview.ts index 91eebab21..446998d36 100644 --- a/src/lib/export/backup-preview.ts +++ b/src/lib/export/backup-preview.ts @@ -35,7 +35,10 @@ import { } from "@/lib/export/backup-key-ids"; import type { FullBackupCounts } from "@/lib/export/full-backup-payload"; import type { StoredBackupRef } from "@/lib/export/stored-backup"; -import type { BackupSummary } from "@/lib/validations/backup-summary"; +import { + BACKUP_SCHEMA_VERSION, + type BackupSummary, +} from "@/lib/validations/backup-summary"; export interface BackupPreview { version: 1; @@ -207,7 +210,9 @@ export interface BackupPreviewCollector { observe(member: string, value: unknown): void; /** * The counts and key uses, once the writer has finished; null when the - * copy did not name its schema version, owner and date. + * copy did not name its schema version, owner and date, or when a section + * did not pass the backup schema. A null preview sends the preview route + * to read the copy itself, which then says why it cannot be restored. */ finish(counts: FullBackupCounts): { summary: BackupSummary; @@ -215,19 +220,64 @@ export interface BackupPreviewCollector { } | null; } +type SchemaModule = typeof import("@/lib/validations/backup"); + +/** + * A check of one member of the payload against `backupPayloadSchema`, as the + * writer shows it: a whole section, or one row of a bulk table. An array + * section is checked an element at a time, so the check holds one element's + * copy at most, never a section's. Each value is checked in its JSON form, + * exactly as it lands in the copy. A member the schema does not name passes, + * as it does under the schema's passthrough. + */ +function sectionChecker( + mod: SchemaModule, +): (member: string, value: unknown) => boolean { + const shape = mod.backupPayloadSchema.shape as Record; + const asWritten = (value: unknown): unknown => + value === undefined ? undefined : JSON.parse(JSON.stringify(value)); + const unwrap = (schema: z.ZodType): z.ZodType => + schema instanceof z.ZodDefault + ? unwrap(schema.unwrap() as z.ZodType) + : schema; + return (member, value) => { + const declared = shape[member]; + if (!declared) return true; + const section = unwrap(declared); + if (section instanceof z.ZodArray) { + const element = section.element as z.ZodType; + const rows = Array.isArray(value) ? value : [value]; + return rows.every((row) => element.safeParse(asWritten(row)).success); + } + return declared.safeParse(asWritten(value)).success; + }; +} + /** * Work out a weekly copy's preview from what its writer already has in hand: - * each section as it is written, for the key uses, and the writer's own - * counts. Nothing is parsed and nothing is kept but the key samples, so the - * preview costs the pass no memory that grows with the record. + * each section as it is written, for the key uses and the schema check, and + * the writer's own counts. Nothing is parsed back and nothing is kept but the + * key samples, so the preview costs the pass no memory that grows with the + * record. The check loads the backup schema (`backup-summary.ts` says why it + * is loaded on first use), a fixed cost of the module, not of the record. */ -export function createBackupPreviewCollector(): BackupPreviewCollector { +export async function createBackupPreviewCollector(): Promise { + const check = sectionChecker(await import("@/lib/validations/backup")); const keys = new BackupKeyIdCollector(); const header: Partial< Record<"schemaVersion" | "userId" | "exportedAt", string> > = {}; + let valid = true; + let measurementWithoutId = false; return { observe(member, value) { + if (valid && !check(member, value)) valid = false; + if (member === "measurements") { + const rows: unknown[] = Array.isArray(value) ? value : [value]; + if (rows.some((row) => !(row as { id?: unknown } | null)?.id)) { + measurementWithoutId = true; + } + } if ( (member === "schemaVersion" || member === "userId" || @@ -242,6 +292,12 @@ export function createBackupPreviewCollector(): BackupPreviewCollector { finish(counts) { const { schemaVersion, userId, exportedAt } = header; if (!schemaVersion || !userId || !exportedAt) return null; + if (!valid) return null; + // The schema's one whole-document rule: a canonical copy names every + // measurement by a stable id. + if (schemaVersion === BACKUP_SCHEMA_VERSION && measurementWithoutId) { + return null; + } const summary = { schemaVersion, userId, exportedAt } as BackupSummary; for (const field of SUMMARY_COUNTS) summary[field] = counts[field]; return { summary, keys }; diff --git a/src/lib/jobs/__tests__/backup-pass-attempts.test.ts b/src/lib/jobs/__tests__/backup-pass-attempts.test.ts new file mode 100644 index 000000000..722fec5ff --- /dev/null +++ b/src/lib/jobs/__tests__/backup-pass-attempts.test.ts @@ -0,0 +1,118 @@ +import { describe, expect, it } from "vitest"; + +import { + heldBackAccounts, + orderInterruptedLast, +} from "../backup-pass-attempts"; + +type Row = { userId: string; startedAt: Date; finishedAt: Date | null }; + +/** + * A weekly-style pass over `ids`: oldest copy first, held-back accounts last, + * at most `budget` accounts per run, `kills` the accounts whose record kills + * the process. Returns the accounts copied in each run. + */ +function simulate(opts: { + ids: string[]; + budget: number; + kills: Set; + runs: number; + interruptedFirst?: string; +}): string[][] { + let clock = 0; + const attempts = new Map(); + const copiedAt = new Map(); + if (opts.interruptedFirst) { + attempts.set(opts.interruptedFirst, { + userId: opts.interruptedFirst, + startedAt: new Date(++clock), + finishedAt: null, + }); + } + const copied: string[][] = []; + for (let run = 0; run < opts.runs; run += 1) { + const order = orderInterruptedLast( + [...opts.ids] + .map((id) => ({ id })) + .sort( + (a, b) => (copiedAt.get(a.id) ?? -1) - (copiedAt.get(b.id) ?? -1), + ), + heldBackAccounts([...attempts.values()]), + ); + const thisRun: string[] = []; + for (const { id } of order.slice(0, opts.budget)) { + const row = attempts.get(id) ?? { + userId: id, + startedAt: new Date(0), + finishedAt: null, + }; + row.startedAt = new Date(++clock); + attempts.set(id, row); + if (opts.kills.has(id)) break; // the process dies; the run ends here + row.finishedAt = new Date(++clock); + copiedAt.set(id, clock); + thisRun.push(id); + } + copied.push(thisRun); + } + return copied; +} + +describe("heldBackAccounts", () => { + const at = (ms: number) => new Date(ms); + + it("holds back an interrupted attempt that is the newest of the pass", () => { + const held = heldBackAccounts([ + { userId: "x", startedAt: at(30), finishedAt: at(10) }, + { userId: "a", startedAt: at(20), finishedAt: at(21) }, + ]); + expect([...held.keys()]).toEqual(["x"]); + }); + + it("releases it once a later run has started on another account", () => { + const held = heldBackAccounts([ + { userId: "x", startedAt: at(30), finishedAt: null }, + { userId: "a", startedAt: at(40), finishedAt: at(41) }, + ]); + expect(held.size).toBe(0); + }); + + it("ignores finished attempts", () => { + const held = heldBackAccounts([ + { userId: "a", startedAt: at(40), finishedAt: at(41) }, + ]); + expect(held.size).toBe(0); + }); +}); + +describe("the order across runs", () => { + const ids = ["x", "a", "b", "c", "d"]; + + it("keeps a record that kills the process from costing the others their copies", () => { + // `x` has the oldest copy and kills every run that reaches it. + const runs = simulate({ ids, budget: 99, kills: new Set(["x"]), runs: 6 }); + // The first run meets it unwarned; from then on it goes last every run, + // because each crash makes its attempt the newest again. + expect(runs[0]).toEqual([]); + for (const run of runs.slice(1)) { + expect([...run].sort()).toEqual(["a", "b", "c", "d"]); + } + }); + + it("still copies an interrupted account when every run stops on its budget", () => { + // `x` was interrupted once (a restart, not its record). Every run stops + // after three accounts, so kept last for good it would never be reached. + const runs = simulate({ + ids, + budget: 3, + kills: new Set(), + runs: 4, + interruptedFirst: "x", + }); + const firstCopy = runs.findIndex((run) => run.includes("x")); + expect(firstCopy).toBeGreaterThanOrEqual(0); + expect(firstCopy).toBeLessThanOrEqual(1); + // And every account gets a copy within the four runs. + expect(new Set(runs.flat())).toEqual(new Set(ids)); + }); +}); diff --git a/src/lib/jobs/__tests__/restore-drill.test.ts b/src/lib/jobs/__tests__/restore-drill.test.ts index d223c5a45..81eb035ef 100644 --- a/src/lib/jobs/__tests__/restore-drill.test.ts +++ b/src/lib/jobs/__tests__/restore-drill.test.ts @@ -7,7 +7,9 @@ import { runRestoreDrill, RESTORE_DRILL_CRON, RESTORE_DRILL_QUEUE, + RESTORE_DRILL_SEND_OPTIONS, } from "../restore-drill"; +import { QUEUE_RUNTIME } from "../queue-runtime"; vi.mock("../report-worker-error", () => ({ reportWorkerError: vi.fn().mockResolvedValue(undefined), @@ -86,6 +88,36 @@ describe("runRestoreDrill", () => { expect(RESTORE_DRILL_CRON).toBe("11 4 1 * *"); }); + it("runs as a locked long pass that pages each account once per run", () => { + // Every account's object is read, so the pg-boss default of fifteen + // minutes is no ceiling for it; no retry, so a failed account is paged + // once per run instead of three times. + expect(RESTORE_DRILL_SEND_OPTIONS.expireInSeconds).toBeGreaterThan(15 * 60); + expect(RESTORE_DRILL_SEND_OPTIONS.retryLimit).toBe(0); + expect(QUEUE_RUNTIME[RESTORE_DRILL_QUEUE]).toMatchObject({ + runtime: "long", + exclusive: "lockedPass", + }); + }); + + it("stops between accounts on its budget and counts the ones it did not reach", async () => { + const s3 = makeS3Mock({ + "2026-06-01/user-aaa.json.enc": backupObject({ userId: "user-aaa" }), + "2026-06-01/user-mmm.json.enc": backupObject({ userId: "user-mmm" }), + "2026-06-01/user-zzz.json.enc": backupObject({ userId: "user-zzz" }), + }); + let asked = 0; + const report = await runRestoreDrill( + s3, + new Date("2026-06-02T04:11:00Z"), + () => ++asked > 1, + ); + expect(report.accounts.map((a) => a.objectKey)).toEqual([ + "2026-06-01/user-aaa.json.enc", + ]); + expect(report.unchecked).toBe(2); + }); + it("fetches, decrypts, and parses the newest backup object", async () => { const s3 = makeS3Mock({ "2026-05-30/user-old.json.enc": backupObject(), diff --git a/src/lib/jobs/backup-pass-attempts.ts b/src/lib/jobs/backup-pass-attempts.ts index 05c0d919b..389252334 100644 --- a/src/lib/jobs/backup-pass-attempts.ts +++ b/src/lib/jobs/backup-pass-attempts.ts @@ -16,6 +16,16 @@ * takes that account last, after everybody else has their copy, and the admin * backups page names it. Every write and read here is best effort: the marker * is bookkeeping about a backup, and must never be the reason one fails. + * + * Last only once. A pass that regularly stops on its time budget never reaches + * the end of its list, so an account kept last for as long as its attempt + * stays open would never be attempted again, and would never get a copy. An + * interrupted account is therefore held back only until a later run has + * started on another account (`accountsToTakeLast`): from then on it takes its + * ordinary place again. If it kills the process again, its new attempt is the + * newest one and it goes last once more. So a crash loop costs the other + * accounts at most every second run, and the account itself is attempted at + * least every second run. */ import type { PrismaClient } from "@/generated/prisma/client"; @@ -83,7 +93,51 @@ export async function readInterruptedBackupAttempts( } /** - * `accounts` in their order, with every interrupted one moved behind the + * The interrupted accounts a pass should take last: those whose interrupted + * attempt is still the newest attempt of the pass, i.e. no later run has + * started on another account since. Rows are one per account. Pure, so the + * rule can be tested apart from the database. + */ +export function heldBackAccounts( + rows: ReadonlyArray<{ + userId: string; + startedAt: Date; + finishedAt: Date | null; + }>, +): Map { + const out = new Map(); + for (const row of rows) { + const open = row.finishedAt === null || row.finishedAt < row.startedAt; + if (!open) continue; + const passedOver = rows.some( + (other) => other.userId !== row.userId && other.startedAt > row.startedAt, + ); + if (!passedOver) out.set(row.userId, row.startedAt); + } + return out; +} + +/** + * What a pass orders by: the interrupted accounts it takes last + * (`heldBackAccounts`). Empty when the table cannot be read. + */ +export async function readAccountsToTakeLast( + prisma: Db, + pass: BackupPass, +): Promise> { + try { + const rows = await prisma.backupPassAttempt.findMany({ + where: { pass }, + select: { userId: true, startedAt: true, finishedAt: true }, + }); + return heldBackAccounts(rows); + } catch { + return new Map(); + } +} + +/** + * `accounts` in their order, with every held-back one moved behind the * rest; among those, the one interrupted longest ago first. */ export function orderInterruptedLast( diff --git a/src/lib/jobs/job-outcome.ts b/src/lib/jobs/job-outcome.ts index 2ef5e5cc0..33e994a32 100644 --- a/src/lib/jobs/job-outcome.ts +++ b/src/lib/jobs/job-outcome.ts @@ -200,6 +200,7 @@ export const JOB_FACT_ALLOWLIST: ReadonlySet = new Set([ "restore_drill_medications", "restore_drill_mood_entries", "restore_drill_plaintext_bytes", + "restore_drill_unchecked", "restore_drill_stale", "restore_fail_cause", "restore_failed", diff --git a/src/lib/jobs/offhost-backup.ts b/src/lib/jobs/offhost-backup.ts index ad9cf04dd..05a9f23b5 100644 --- a/src/lib/jobs/offhost-backup.ts +++ b/src/lib/jobs/offhost-backup.ts @@ -39,7 +39,7 @@ import { markBackupAttemptFinished, markBackupAttemptStarted, orderInterruptedLast, - readInterruptedBackupAttempts, + readAccountsToTakeLast, } from "@/lib/jobs/backup-pass-attempts"; import { BACKUP_HEARTBEAT_SECONDS } from "@/lib/jobs/data-backup-policy"; import { annotate, getEvent } from "@/lib/logging/context"; @@ -930,14 +930,15 @@ export async function runOffhostBackup( // By id, with an account whose last walk never came back behind the rest: // a record that killed the process once would otherwise lead every retry - // and keep every account after it out of the bucket - // (`backup-pass-attempts.ts`). + // and keep every account after it out of the bucket. Behind them for one + // run only, so a pass that stops on its budget still reaches it the run + // after (`backup-pass-attempts.ts`). const users = orderInterruptedLast( await prisma.user.findMany({ select: { id: true }, orderBy: { id: "asc" }, }), - await readInterruptedBackupAttempts(prisma, OFFHOST_BACKUP_QUEUE), + await readAccountsToTakeLast(prisma, OFFHOST_BACKUP_QUEUE), ); const doneThisRun = new Set(); if (runStartedAt) { diff --git a/src/lib/jobs/queue-runtime.ts b/src/lib/jobs/queue-runtime.ts index 18177c274..3dd254b8f 100644 --- a/src/lib/jobs/queue-runtime.ts +++ b/src/lib/jobs/queue-runtime.ts @@ -36,6 +36,7 @@ import { DOCUMENT_AI_RUN_EXPIRE_SECONDS } from "@/lib/jobs/document-ai-run"; import { DATA_BACKUP_SEND_OPTIONS } from "@/lib/jobs/data-backup-policy"; import { OFFHOST_BACKUP_EXPIRE_SECONDS } from "@/lib/jobs/offhost-backup"; import { BACKUP_RESTORE_EXPIRE_SECONDS } from "@/lib/jobs/backup-restore"; +import { RESTORE_DRILL_EXPIRE_SECONDS } from "@/lib/jobs/restore-drill"; import { APPLE_HEALTH_IMPORT_SEND_OPTIONS } from "@/lib/jobs/apple-health-import-worker"; import { MEDICATION_INTAKE_IMPORT_SEND_OPTIONS } from "@/lib/jobs/medication-intake-import"; @@ -267,7 +268,14 @@ export const QUEUE_RUNTIME: Readonly> = { exclusive: "lockedPass", why: "Uploads every account's record off-host.", }, - "data-restore-drill": short("Fetches and stream-parses one off-host object."), + "data-restore-drill": { + runtime: "long", + expireInSeconds: RESTORE_DRILL_EXPIRE_SECONDS, + expiryVia: ["RESTORE_DRILL_SEND_OPTIONS"], + stop: { file: "lib/jobs/restore-drill.ts", fn: "handleRestoreDrill" }, + exclusive: "lockedPass", + why: "Fetches and stream-parses every account's newest off-host object.", + }, "offhost-backup-purge": short("At most 200 delete requests per run."), "backup-restore": { runtime: "long", diff --git a/src/lib/jobs/reminder/backup-handlers.ts b/src/lib/jobs/reminder/backup-handlers.ts index d4bb4b1a4..c8bc8fa83 100644 --- a/src/lib/jobs/reminder/backup-handlers.ts +++ b/src/lib/jobs/reminder/backup-handlers.ts @@ -27,7 +27,7 @@ import { markBackupAttemptFinished, markBackupAttemptStarted, orderInterruptedLast, - readInterruptedBackupAttempts, + readAccountsToTakeLast, } from "@/lib/jobs/backup-pass-attempts"; import { DATA_BACKUP_QUEUE } from "@/lib/jobs/data-backup-policy"; @@ -154,7 +154,7 @@ export async function handleDataBackup( where: { type: "WEEKLY_AUTO" }, select: { userId: true, createdAt: true }, }), - readInterruptedBackupAttempts(prisma, DATA_BACKUP_QUEUE), + readAccountsToTakeLast(prisma, DATA_BACKUP_QUEUE), ]); const copiedAt = new Map( copies.map((copy) => [copy.userId, copy.createdAt.getTime()]), @@ -163,7 +163,9 @@ export async function handleDataBackup( // attempt never came back goes behind all of them: it is the one with // the oldest copy, so it would lead every retry, and a record that // killed the process once would keep every account after it from a - // copy (`backup-pass-attempts.ts`). + // copy. Only until a later run has started on another account, so a + // pass that stops on its budget still reaches it the run after + // (`backup-pass-attempts.ts`). const users = orderInterruptedLast( [...accounts].sort( (a, b) => (copiedAt.get(a.id) ?? -1) - (copiedAt.get(b.id) ?? -1), @@ -205,7 +207,7 @@ export async function handleDataBackup( // so its size does not depend on this process's memory (#1031). // The restore preview is taken from the writer's own counts and // the sections as they are written, never by parsing the copy. - const preview = createBackupPreviewCollector(); + const preview = await createBackupPreviewCollector(); let counts: FullBackupCounts | undefined; const { bytes: storedBytes } = await storeBackupBlob( prisma, diff --git a/src/lib/jobs/reminder/register-maintenance.ts b/src/lib/jobs/reminder/register-maintenance.ts index 6fd906114..d774495d4 100644 --- a/src/lib/jobs/reminder/register-maintenance.ts +++ b/src/lib/jobs/reminder/register-maintenance.ts @@ -83,6 +83,7 @@ import { handleRestoreDrill, RESTORE_DRILL_CRON, RESTORE_DRILL_QUEUE, + RESTORE_DRILL_SEND_OPTIONS, } from "@/lib/jobs/restore-drill"; import { BACKUP_RESTORE_QUEUE, @@ -551,8 +552,10 @@ const allQueues = [ // a bulk DELETE or UPDATE that timed out will time out again on an immediate // retry, so pg-boss's default two would run the same doomed statement three // times a night. The queues without it are the ones whose failure is -// transient — the TLS probe, the off-host upload, the restore drill, the -// per-user environment fetch — and those still want the default retries. +// transient — the TLS probe, the off-host upload, the per-user environment +// fetch — and those still want the default retries. The restore drill sends +// `retryLimit: 0` in its own options for the other reason: it pages each +// failed account itself, and a retry would page the same accounts again. const schedules: ScheduleEntry[] = [ // v1.25 (W-ENV) — daily 02:10 Europe/Berlin discovery tick (empty payload) // that fans out one per-user environment fetch per opted-in account. @@ -571,7 +574,7 @@ const schedules: ScheduleEntry[] = [ cronIsTheRetry, ], [OFFHOST_BACKUP_QUEUE, OFFHOST_BACKUP_CRON, OFFHOST_BACKUP_SEND_OPTIONS], - [RESTORE_DRILL_QUEUE, RESTORE_DRILL_CRON], + [RESTORE_DRILL_QUEUE, RESTORE_DRILL_CRON, RESTORE_DRILL_SEND_OPTIONS], // Nightly backstop for the purge the deletion kicks straight away: a // request the bucket refused, or one written while the queue was down. [OFFHOST_PURGE_QUEUE, OFFHOST_PURGE_CRON, cronIsTheRetry], @@ -832,11 +835,13 @@ export async function registerMaintenanceQueues( { localConcurrency: 1, includeMetadata: true }, lockedPass(OFFHOST_BACKUP_QUEUE, () => WHOLE_PASS, handleOffhostBackup), ); + // Locked: a drill that outlives a restart or a manual run never reads the + // bucket beside another one. await createAndWork( boss, RESTORE_DRILL_QUEUE, { localConcurrency: 1 }, - handleRestoreDrill, + lockedPass(RESTORE_DRILL_QUEUE, () => WHOLE_PASS, handleRestoreDrill), ); await createAndWork( boss, diff --git a/src/lib/jobs/restore-drill.ts b/src/lib/jobs/restore-drill.ts index caec85d00..1a6854bc1 100644 --- a/src/lib/jobs/restore-drill.ts +++ b/src/lib/jobs/restore-drill.ts @@ -21,6 +21,13 @@ * Schedule: 04:11 on the 1st of each month (`11 4 1 * *`), after the * 02:30 nightly upload and the 03:xx cleanup window, on a minute slot * no other cron uses. + * + * The drill reads every account's object, so it runs as a long pass: an + * explicit expiry (`RESTORE_DRILL_EXPIRE_SECONDS`), a stop between accounts + * on the job's budget, a lock so two drills never overlap (`lockedPass` at + * the binding), and no retry. A retry would re-read the same objects and + * page the same failed accounts a second and third time; each account is + * reported once per run, and the next drill is the retry. */ import type { Job } from "pg-boss"; import { BackupJsonError, scanBackupJson } from "@/lib/export/backup-json-scan"; @@ -38,12 +45,28 @@ import { type S3Like, } from "@/lib/jobs/offhost-backup"; import { reportWorkerError } from "@/lib/jobs/report-worker-error"; +import { jobBudget } from "@/lib/jobs/job-budget"; import { jobDone, jobFailed, type JobOutcome } from "@/lib/jobs/job-outcome"; import { withBackgroundEvent } from "@/lib/logging/background"; export const RESTORE_DRILL_QUEUE = "data-restore-drill"; export const RESTORE_DRILL_CRON = "11 4 1 * *"; +/** + * Four hours, the off-host upload's own window: the drill downloads and + * stream-parses every object that upload wrote in one night. + */ +export const RESTORE_DRILL_EXPIRE_SECONDS = 4 * 60 * 60; + +/** + * What the monthly schedule sends with: the expiry above and no retry + * (`retryLimit: 0`, the maintenance schedules' `cronIsTheRetry`). + */ +export const RESTORE_DRILL_SEND_OPTIONS = { + expireInSeconds: RESTORE_DRILL_EXPIRE_SECONDS, + retryLimit: 0, +} as const; + /** * A drill run that finds the newest backup older than this is reported * as a failure even when decrypt + parse succeed: the nightly uploader @@ -80,6 +103,8 @@ export interface RestoreDrillReport { accounts: RestoreDrillAccount[]; /** The accounts whose object did not pass. */ failed: RestoreDrillAccount[]; + /** Objects of the newest date the run stopped before, on its budget. */ + unchecked: number; } function countArray(value: unknown): number { @@ -178,6 +203,7 @@ async function drillObject( export async function runRestoreDrill( s3Override?: S3Like, now: Date = new Date(), + shouldStop: () => boolean = () => false, ): Promise { const cfg = loadOffhostConfig(); if (!cfg) { @@ -208,6 +234,8 @@ export async function runRestoreDrill( const accounts: RestoreDrillAccount[] = []; for (const objectKey of newest) { + // Between accounts, never inside one: the job's budget is spent. + if (shouldStop()) break; const account: RestoreDrillAccount = { objectKey, ok: true, @@ -239,6 +267,7 @@ export async function runRestoreDrill( stale: ageDays > MAX_BACKUP_AGE_DAYS, accounts, failed: accounts.filter((a) => !a.ok), + unchecked: newest.length - accounts.length, }; } @@ -252,10 +281,10 @@ export async function runRestoreDrill( export async function handleRestoreDrill( jobs: Job[], ): Promise { - void jobs; + const shouldStop = jobBudget(jobs); return withBackgroundEvent("job.restore_drill", async (evt) => { try { - const report = await runRestoreDrill(); + const report = await runRestoreDrill(undefined, new Date(), shouldStop); const total = (pick: (a: RestoreDrillAccount) => number) => report.accounts.reduce((sum, a) => sum + pick(a), 0); const meta = { @@ -270,6 +299,7 @@ export async function handleRestoreDrill( restore_drill_intake_events: total((a) => a.recordCounts.intakeEvents), restore_drill_mood_entries: total((a) => a.recordCounts.moodEntries), restore_drill_stale: report.stale, + restore_drill_unchecked: report.unchecked, }; for (const [key, value] of Object.entries(meta)) evt.addMeta(key, value); evt.addMeta( @@ -301,9 +331,18 @@ export async function handleRestoreDrill( { dateKey: report.dateKey, ageDays: report.ageDays }, ); } + if (report.unchecked > 0) { + await reportWorkerError( + RESTORE_DRILL_QUEUE, + new Error( + `Restore drill stopped on its time budget with ${report.unchecked} of ${report.accounts.length + report.unchecked} backup objects unchecked.`, + ), + { dateKey: report.dateKey, unchecked: report.unchecked }, + ); + } if (report.failed.length > 0) { - // Already paged above, per account. A retry reads the same objects - // and cannot change the verdict before the next upload. + // Already paged above, per account. The queue does not retry: a + // retry reads the same objects and would page the same accounts again. return jobFailed( `restore drill: ${report.failed.length} of ${report.accounts.length} accounts failed`, new Error(report.failed.map((a) => a.objectKey).join(", ")), diff --git a/src/lib/measurements/day-mean.ts b/src/lib/measurements/day-mean.ts index 38b2e2970..9d90908bc 100644 --- a/src/lib/measurements/day-mean.ts +++ b/src/lib/measurements/day-mean.ts @@ -245,34 +245,32 @@ function weightedRowsSql( if (!/^[a-z_][a-z0-9_]*$/.test(source)) { throw new Error(`source must be a CTE name: ${source}`); } - const hourOf = (alias: string) => - `date_trunc('hour', ${frameTs(frame, `${alias}."measured_at"`)})`; - const dayOf = (alias: string) => - `date_trunc('day', ${frameTs(frame, `${alias}."measured_at"`)})`; - const src = bySource ? `, t."source"` : ""; - const part = bySource ? `, x."source"` : ""; - const join = bySource ? ` AND hw."source" = s."source"` : ""; + // One scan of the CTE and window functions only. The earlier shape joined + // the CTE to an hour grouping of itself; with stale or missing statistics + // (a fresh import below the autoanalyze threshold) the planner ran that + // join as a nested loop and a year of pulse ran into the statement timeout. + // + // `dm_hour_n` counts the readings in a row's hour and `dm_hour_first` marks + // one row per hour, so the sum of `dm_hour_first` over a day is the day's + // hour count. Both columns ride along on the row; every caller aggregates. + const partition = (alias: string, unit: "hour" | "day") => { + const keys = bySource + ? `${alias}."type", ${alias}."source"` + : `${alias}."type"`; + const ts = frameTs(frame, `${alias}."measured_at"`); + return `PARTITION BY ${keys}, date_trunc('${unit}', ${ts})`; + }; return `( SELECT s.*, (CASE WHEN ${isHourlyMeanTypeSql('s."type"')} - THEN 1.0 / (hw.n * hw.hours) + THEN 1.0 / (s.dm_hour_n * SUM(s.dm_hour_first) OVER (${partition("s", "day")})) ELSE 1.0 END)::double precision AS day_weight - FROM ${source} s - LEFT JOIN ( - SELECT x."type"${part}, x.local_hour, x.n, - COUNT(*) OVER (PARTITION BY x."type"${part}, x.local_day) AS hours - FROM ( - SELECT t."type"${src}, - ${hourOf("t")} AS local_hour, - ${dayOf("t")} AS local_day, - COUNT(*) AS n - FROM ${source} t - WHERE ${isHourlyMeanTypeSql('t."type"')} - GROUP BY ${bySource ? "1, 2, 3, 4" : "1, 2, 3"} - ) x - ) hw - ON hw."type" = s."type"${join} - AND hw.local_hour = ${hourOf("s")} + FROM ( + SELECT t.*, + COUNT(*) OVER (${partition("t", "hour")}) AS dm_hour_n, + (ROW_NUMBER() OVER (${partition("t", "hour")}) = 1)::int AS dm_hour_first + FROM ${source} t + ) s )`; } diff --git a/src/lib/measurements/import-apple-health-export.ts b/src/lib/measurements/import-apple-health-export.ts index dc280d285..f75e2a53e 100644 --- a/src/lib/measurements/import-apple-health-export.ts +++ b/src/lib/measurements/import-apple-health-export.ts @@ -122,11 +122,14 @@ export interface ImportJobResult { clinical: { skipped: number }; /** * Records left out because HealthLog wrote them into Apple Health itself and - * already holds them under their own source, each counted once however often - * the export repeats it: `byMarker` carry the app's origin marker, - * `byExternalId` carry a HealthLog row id as `HKExternalUUID` (samples from - * before the marker existed), `matchedManual` match a manual entry (same type - * and value, within 2 s) without either. + * this account already holds them under their own source, each counted once + * however often the export repeats it: `byMarker` carry the app's origin + * marker and a HealthLog row id of this account (a cycle sample: a day-log of + * this account on its day), `byExternalId` carry such a row id without the + * marker (samples from before the marker existed), `matchedManual` match a + * manual entry (same type and value, within 2 s). A marked sample this + * account does not hold (a move to a new instance without a backup) is + * imported like any other. */ writtenByHealthLog: { byMarker: number; @@ -186,8 +189,11 @@ interface PreparedMeasurement { * at the close tag, once its `` children have been read. */ interface PendingRecord { - /** Queue the record; `healthLogId` is the `HKExternalUUID` it carried. */ - commit: (healthLogId: string | null) => void; + /** + * Queue the record; `healthLogId` is the `HKExternalUUID` it carried and + * `marked` whether it (or its correlation) carried the origin marker. + */ + commit: (healthLogId: string | null, marked: boolean) => void; ownOrigin: boolean; healthLogId: string | null; /** The record's sample identity, computed only when it is left out. */ @@ -369,7 +375,11 @@ export interface StreamParseInput { /** Prisma client to flush rows through. */ prisma: Pick< PrismaClient, - "measurement" | "workout" | "$transaction" | "$queryRawUnsafe" + | "measurement" + | "workout" + | "cycleDayLog" + | "$transaction" + | "$queryRawUnsafe" >; /** * Live progress hook. Called every `PROGRESS_TICK_RECORDS` records @@ -449,6 +459,15 @@ export async function streamParseExportXml( sampleKey: string; ownOrigin: boolean; } | null = null; + // Marked cycle samples, held to the end of the parse: one counts as written + // by HealthLog only when this account holds a day-log on its day. + const markedCycle: Array<{ + hkType: string; + dayKey: string; + rawValue: string | undefined; + protectionUsed?: boolean; + sampleKey: string; + }> = []; // Cumulative fold: type -> local day -> hashed source identity -> subtotal. const cumulativeBucket = new Map< @@ -471,13 +490,25 @@ export async function streamParseExportXml( countedOwn.add(sampleKey); writtenByHealthLog[reason] += 1; }; - // The `HKExternalUUID` a queued spot row carried, checked at flush. + // The `HKExternalUUID` a queued spot row carried, checked at flush, and the + // queued rows that carried the origin marker. const healthLogIdOf = new Map(); + const markedRows = new Set(); + // Marked cumulative samples that named a HealthLog row id, held to the end + // of the parse and added to their day only when the id is not this + // account's. Small: one per mirrored total. + const markedCumulative: Array<{ + type: MeasurementType; + healthLogId: string; + sampleKey: string; + add: () => void; + }> = []; // A `` (a blood pressure reading) wraps its records, and the // marker may sit on the correlation itself, before or after them. Its records // are held until it closes, then all kept or all left out together. let currentCorrelation: { ownOrigin: boolean; + healthLogId: string | null; held: PendingRecord[]; } | null = null; // Spot-row batch awaiting flush. @@ -536,12 +567,29 @@ export async function streamParseExportXml( }; // ── Flush helpers ────────────────────────────────────────── - // A sample the app wrote before it stamped its origin marker still carries - // the HealthLog row id as `HKExternalUUID`. A hit on one of this account's - // rows (deleted ones included: the sample is ours either way) of the same - // kind is an exact match. Blood pressure is one row per half, while the app - // stamps both samples of a reading with one id (the systolic row's, the - // diastolic row's in older builds), so either half matches either id. + // A sample the app wrote carries the HealthLog row id as `HKExternalUUID`, + // with the origin marker or (before the marker existed) without it. A hit on + // one of this account's rows (deleted ones included: the sample is ours + // either way) of the same kind is an exact match. The marker alone proves + // nothing about THIS account: after a move to a new instance without a + // backup, re-importing the export must bring the mirrored values back. + // Blood pressure is one row per half, while the app stamps both samples of a + // reading with one id (the systolic row's, the diastolic row's in older + // builds), so either half matches either id. + const ownRowKinds = async ( + ids: ReadonlySet, + ): Promise> => { + const out = new Map(); + const list = [...ids]; + for (let i = 0; i < list.length; i += 1000) { + const known = await prisma.measurement.findMany({ + where: { userId, id: { in: list.slice(i, i + 1000) } }, + select: { id: true, type: true }, + }); + for (const row of known) out.set(row.id, kindOf(row.type)); + } + return out; + }; const withoutHealthLogIds = async ( rows: PreparedMeasurement[], ): Promise => { @@ -551,17 +599,16 @@ export async function streamParseExportXml( if (id) ids.add(id); } if (ids.size === 0) return rows; - const known = await prisma.measurement.findMany({ - where: { userId, id: { in: [...ids] } }, - select: { id: true, type: true }, - }); - if (known.length === 0) return rows; - const typeOf = new Map(known.map((row) => [row.id, row.type])); + const kindOfId = await ownRowKinds(ids); + if (kindOfId.size === 0) return rows; return rows.filter((row) => { const id = healthLogIdOf.get(row); - const rowType = id ? typeOf.get(id) : undefined; - if (!rowType || kindOf(rowType) !== kindOf(row.type)) return true; - countOwn("byExternalId", row.externalId); + const rowKind = id ? kindOfId.get(id) : undefined; + if (!rowKind || rowKind !== kindOf(row.type)) return true; + countOwn( + markedRows.has(row) ? "byMarker" : "byExternalId", + row.externalId, + ); return false; }); }; @@ -663,7 +710,10 @@ export async function streamParseExportXml( const chunk = await withoutManualMirrors( await withoutHealthLogIds(incoming), ); - for (const row of incoming) healthLogIdOf.delete(row); + for (const row of incoming) { + healthLogIdOf.delete(row); + markedRows.delete(row); + } for (const row of chunk) widenSpan(row.measuredAt); const insertedArrivals: Array<{ id: string; @@ -993,6 +1043,61 @@ export async function streamParseExportXml( } }; + const consumeCycle = (rec: { + hkType: string; + dayKey: string; + rawValue: string | undefined; + protectionUsed?: boolean; + }): void => { + const consumed = cycleAccumulator.consume( + rec.hkType, + rec.dayKey, + rec.rawValue, + rec.protectionUsed, + ); + if (!consumed) { + // Recognised identifier but unrecognised value — count it under + // `unknown` with the reason tag so operators can spot it. + unknown[`${rec.hkType}::cycle_unmapped`] = + (unknown[`${rec.hkType}::cycle_unmapped`] ?? 0) + 1; + } + }; + + // Marked cumulative samples: left out when their id is a row of this + // account of the same kind, added to their day otherwise. + const settleMarkedCumulative = async (): Promise => { + if (markedCumulative.length === 0) return; + const kindOfId = await ownRowKinds( + new Set(markedCumulative.map((m) => m.healthLogId)), + ); + for (const m of markedCumulative.splice(0, markedCumulative.length)) { + if (kindOfId.get(m.healthLogId) === kindOf(m.type)) { + countOwn("byMarker", m.sampleKey); + } else { + m.add(); + } + } + }; + + // Marked cycle samples: left out when this account holds a day-log on + // their day (deleted ones included), folded in otherwise. + const settleMarkedCycle = async (): Promise => { + if (markedCycle.length === 0) return; + const days = [...new Set(markedCycle.map((rec) => rec.dayKey))]; + const held = new Set(); + for (let i = 0; i < days.length; i += 1000) { + const logs = await prisma.cycleDayLog.findMany({ + where: { userId, date: { in: days.slice(i, i + 1000) } }, + select: { date: true }, + }); + for (const log of logs) held.add(log.date); + } + for (const rec of markedCycle.splice(0, markedCycle.length)) { + if (held.has(rec.dayKey)) countOwn("byMarker", rec.sampleKey); + else consumeCycle(rec); + } + }; + // ── SAX parser configuration ──────────────────────────────── const parser = sax.parser(true, { trim: true }); let pendingError: Error | null = null; @@ -1121,27 +1226,44 @@ export async function streamParseExportXml( attrs.startDate, attrs.endDate, ), - commit: (healthLogId) => { + commit: (healthLogId, marked) => { if (CUMULATIVE_HK_TYPES.has(mapped.type)) { const dayKey = dayKeyForUserTz(mapped.takenAt, userTimezone); const sourceHash = hashCumulativeSourceIdentity( attrs.sourceName, attrs.device, ); - let byDay = cumulativeBucket.get(mapped.type); - if (!byDay) { - byDay = new Map(); - cumulativeBucket.set(mapped.type, byDay); + const add = () => { + let byDay = cumulativeBucket.get(mapped.type); + if (!byDay) { + byDay = new Map(); + cumulativeBucket.set(mapped.type, byDay); + } + let bySource = byDay.get(dayKey); + if (!bySource) { + bySource = new Map(); + byDay.set(dayKey, bySource); + } + bySource.set( + sourceHash, + (bySource.get(sourceHash) ?? 0) + mapped.value, + ); + }; + if (marked && healthLogId) { + markedCumulative.push({ + type: mapped.type, + healthLogId, + sampleKey: hashSampleKey( + hkType, + attrs.value ?? "", + attrs.startDate, + attrs.endDate, + ), + add, + }); + } else { + add(); } - let bySource = byDay.get(dayKey); - if (!bySource) { - bySource = new Map(); - byDay.set(dayKey, bySource); - } - bySource.set( - sourceHash, - (bySource.get(sourceHash) ?? 0) + mapped.value, - ); } else { // Spot row: derive a stable externalId, queue for flush. const externalId = hashSampleKey( @@ -1163,6 +1285,7 @@ export async function streamParseExportXml( }; spotBatch.push(row); if (healthLogId) healthLogIdOf.set(row, healthLogId); + if (marked) markedRows.add(row); } }, }; @@ -1244,7 +1367,7 @@ export async function streamParseExportXml( } if (name === "Correlation") { - currentCorrelation = { ownOrigin: false, held: [] }; + currentCorrelation = { ownOrigin: false, healthLogId: null, held: [] }; return; } @@ -1255,8 +1378,12 @@ export async function streamParseExportXml( else if (currentCycleRecord) currentCycleRecord.ownOrigin = true; else if (currentCorrelation) currentCorrelation.ownOrigin = true; } - if (pendingRecord && attrs.key === HK_EXTERNAL_UUID_METADATA_KEY) { - pendingRecord.healthLogId = attrs.value || null; + if (attrs.key === HK_EXTERNAL_UUID_METADATA_KEY) { + // On a record, or on the correlation that wraps it. + if (pendingRecord) pendingRecord.healthLogId = attrs.value || null; + else if (currentCorrelation && !currentCycleRecord) { + currentCorrelation.healthLogId = attrs.value || null; + } } // Attach the SexualActivity protection flag to the open cycle record. // Apple writes `HKMetadataKeySexualActivityProtectionUsed` with a @@ -1306,18 +1433,16 @@ export async function streamParseExportXml( const rec = pendingRecord; pendingRecord = null; if (currentCorrelation) currentCorrelation.held.push(rec); - else if (rec.ownOrigin) countOwn("byMarker", rec.sampleKey()); - else rec.commit(rec.healthLogId); + else rec.commit(rec.healthLogId, rec.ownOrigin); } if (tagName === "Correlation" && currentCorrelation) { const correlation = currentCorrelation; currentCorrelation = null; for (const rec of correlation.held) { - if (correlation.ownOrigin || rec.ownOrigin) { - countOwn("byMarker", rec.sampleKey()); - } else { - rec.commit(rec.healthLogId); - } + rec.commit( + rec.healthLogId ?? correlation.healthLogId, + correlation.ownOrigin || rec.ownOrigin, + ); } } if (tagName === "Workout" && currentWorkout) { @@ -1328,22 +1453,12 @@ export async function streamParseExportXml( const rec = currentCycleRecord; currentCycleRecord = null; if (rec.ownOrigin) { - // A day HealthLog mirrored into Apple Health from its own day-log. - countOwn("byMarker", rec.sampleKey); + // A day HealthLog mirrored into Apple Health from a day-log; decided + // at the end of the parse against this account's day-logs. + markedCycle.push(rec); return; } - const consumed = cycleAccumulator.consume( - rec.hkType, - rec.dayKey, - rec.rawValue, - rec.protectionUsed, - ); - if (!consumed) { - // Recognised identifier but unrecognised value — count it under - // `unknown` with the reason tag so operators can spot it. - unknown[`${rec.hkType}::cycle_unmapped`] = - (unknown[`${rec.hkType}::cycle_unmapped`] ?? 0) + 1; - } + consumeCycle(rec); } }; @@ -1421,7 +1536,9 @@ export async function streamParseExportXml( // threshold during the parse. await flushSpotBatch(); await flushWorkoutBatch(); + await settleMarkedCumulative(); await flushCumulativeBuckets(); + await settleMarkedCycle(); // v1.15.0 — fold the accumulated reproductive samples into CYCLE // day-logs. Gated on cycle-tracking being enabled for the account so a diff --git a/src/lib/queries/__tests__/prefetch-medications.test.ts b/src/lib/queries/__tests__/prefetch-medications.test.ts index 22ee0f750..dc004b368 100644 --- a/src/lib/queries/__tests__/prefetch-medications.test.ts +++ b/src/lib/queries/__tests__/prefetch-medications.test.ts @@ -125,6 +125,28 @@ describe("refetchMedicationsOnMount", () => { unsubscribe(); }); + it("refetches a prefetched read that was invalidated before the mount", async () => { + // Hover the nav link, then log a dose elsewhere before the page mounts: + // the dose invalidates the medication reads, and the prefetched answer + // no longer holds the compliance the user just changed. + const { client, calls } = setup(); + prefetchMedicationsList(client); + await vi.waitFor(() => expect(calls("/api/medications")).toBe(1)); + await vi.waitFor(() => + expect(client.getQueryState(queryKeys.medications())?.status).toBe( + "success", + ), + ); + await client.invalidateQueries({ queryKey: queryKeys.medications() }); + expect(client.getQueryState(queryKeys.medications())?.isInvalidated).toBe( + true, + ); + + const { refetched, unsubscribe } = await mountList(client); + expect(refetched).toBe(true); + unsubscribe(); + }); + it("still refetches on mount when the cache holds an older read", async () => { const { client } = setup(); const queryFn = vi.fn(async () => []); diff --git a/src/lib/queries/prefetch-medications.ts b/src/lib/queries/prefetch-medications.ts index 6058fc137..d78e63c52 100644 --- a/src/lib/queries/prefetch-medications.ts +++ b/src/lib/queries/prefetch-medications.ts @@ -93,11 +93,18 @@ export function prefetchMedicationsList( * the identical request a second time, a few hundred milliseconds after the * first answer landed, on every visit to the page. A return to the page with * an older cache entry, or a direct load, still refetches on mount. + * + * An invalidated entry always refetches: a dose logged elsewhere in the app + * between the prefetch and the mount invalidates these reads, and the + * prefetched answer is then the stale one. */ export function refetchMedicationsOnMount( read: MedicationsRead, -): (query: { state: { dataUpdatedAt: number } }) => boolean | "always" { +): (query: { + state: { dataUpdatedAt: number; isInvalidated: boolean }; +}) => boolean | "always" { return (query) => { + if (query.state.isInvalidated) return "always"; const startedAt = prefetchStartedAt.get(read); const answeredThisVisit = startedAt !== undefined && diff --git a/src/lib/rollups/measurement-rollups.ts b/src/lib/rollups/measurement-rollups.ts index df8e52bb2..8f6643007 100644 --- a/src/lib/rollups/measurement-rollups.ts +++ b/src/lib/rollups/measurement-rollups.ts @@ -597,7 +597,7 @@ async function runRollupAggregate( * every reading, so the band and the windowed slope are untouched. * * Hours and days are UTC, matching the UTC day this tier buckets by; migration - * 0373 applies the same statistic to rows stored before it. + * 0372 applies the same statistic to rows stored before it. */ async function applyHourlyMeanDay( db: Pick, diff --git a/tests/fixtures/apple-health/export-own-origin.synthetic.xml b/tests/fixtures/apple-health/export-own-origin.synthetic.xml index aff5f87ac..fa9491018 100644 --- a/tests/fixtures/apple-health/export-own-origin.synthetic.xml +++ b/tests/fixtures/apple-health/export-own-origin.synthetic.xml @@ -9,9 +9,10 @@ verbatim; that has to be checked against a real export (see #1109). Records, in order: - 1. weight 80.1 carries the HealthLog origin marker -> left out + 1. weight 80.1 origin marker and a HealthLog row id -> left out + when the importing account holds that row 2. weight 80.2 user-entered, no marker -> imported - 3. weight 80.3 marker among other metadata entries -> left out + 3. weight 80.3 marker among other metadata entries, no id -> imported 4. weight 80.4 no metadata at all -> imported 5. weight 80.5 origin key with another value -> imported --> diff --git a/tests/integration/apple-health-import-own-samples.test.ts b/tests/integration/apple-health-import-own-samples.test.ts index a24b663de..e252a6fea 100644 --- a/tests/integration/apple-health-import-own-samples.test.ts +++ b/tests/integration/apple-health-import-own-samples.test.ts @@ -9,7 +9,12 @@ * twice and the rollups count it twice. * * Asserted against a real Postgres: - * - a record carrying the marker is left out, and counted; + * - a record carrying the marker and a row id of this account is left out, + * and counted; the marker alone proves nothing about this account (a move + * to a new instance without a backup), so a marked record whose id is + * missing, unknown or another account's is imported, a deleted row of this + * account still counts, and a marked cycle day counts only when the + * account holds a day-log on it; * - a record without it lands as before; * - the marker is read from a child `` that follows the * record's open tag, so the record is committed at its close tag; @@ -56,13 +61,46 @@ ${records} const weight = (value: number, end: string, children = "") => ` ${children}`; -const OWN = ``; +const MARKER = ``; +const ext = (id: string) => + ``; +/** The row the app mirrored: a weight of this account, its id on the sample. */ +const OWN_WEIGHT_ID = "own-weight-row"; +const OWN_SYS_ID = "own-sys-row"; +const OWN = `${MARKER}${ext(OWN_WEIGHT_ID)}`; +const OWN_BP = `${MARKER}${ext(OWN_SYS_ID)}`; const OTHER = ``; +/** + * A row of this account the samples point at. Withings and ten minutes off the + * samples, so neither the series under test nor the manual-value match sees it. + */ +async function seedOwnRow( + userId: string, + id: string, + type: "WEIGHT" | "BLOOD_PRESSURE_SYS" | "ACTIVITY_STEPS", + deletedAt: Date | null = null, +) { + await prisma.measurement.create({ + data: { + id, + userId, + type, + value: type === "ACTIVITY_STEPS" ? 1000 : 70, + unit: + type === "WEIGHT" ? "kg" : type === "ACTIVITY_STEPS" ? "steps" : "mmHg", + source: "WITHINGS", + measuredAt: new Date("2026-05-14T06:10:00.000Z"), + deletedAt, + }, + }); +} + async function run(records: string) { const user = await prisma.user.create({ data: { username: "own-samples", email: "own@example.test", role: "USER" }, }); + await seedOwnRow(user.id, OWN_WEIGHT_ID, "WEIGHT"); const result = await streamParseExportXml({ xmlPath: writeXml(records), userId: user.id, @@ -139,6 +177,7 @@ describe("blood pressure correlations and self-closing records", () => { const user = await prisma.user.create({ data: { username: "own-bp", email: "bp@example.test", role: "USER" }, }); + await seedOwnRow(user.id, OWN_SYS_ID, "BLOOD_PRESSURE_SYS"); const result = await streamParseExportXml({ xmlPath: writeXml(records), userId: user.id, @@ -148,6 +187,7 @@ describe("blood pressure correlations and self-closing records", () => { const rows = await prisma.measurement.findMany({ where: { userId: user.id, + source: "APPLE_HEALTH", type: { in: ["BLOOD_PRESSURE_SYS", "BLOOD_PRESSURE_DIA"] }, }, orderBy: [{ type: "asc" }], @@ -173,7 +213,7 @@ describe("blood pressure correlations and self-closing records", () => { [bpRecord("Systolic", 121, AT), bpRecord("Diastolic", 79, AT)].join( "\n", ), - `\n ${OWN}`, + `\n ${OWN_BP}`, ), ); expect(values).toEqual([]); @@ -186,7 +226,7 @@ describe("blood pressure correlations and self-closing records", () => { bpRecord("Diastolic", 79, AT), ].join("\n"); const { values } = await runBp( - ` \n${records}\n ${OWN}\n `, + ` \n${records}\n ${OWN_BP}\n `, ); expect(values).toEqual([]); }); @@ -195,7 +235,7 @@ describe("blood pressure correlations and self-closing records", () => { const { values } = await runBp( correlation( [ - bpRecord("Systolic", 121, AT, `\n ${OWN}\n `), + bpRecord("Systolic", 121, AT, `\n ${OWN_BP}\n `), bpRecord("Diastolic", 79, AT), ].join("\n"), ), @@ -210,7 +250,7 @@ describe("blood pressure correlations and self-closing records", () => { [bpRecord("Systolic", 121, AT), bpRecord("Diastolic", 79, AT)].join( "\n", ), - `\n ${OWN}`, + `\n ${OWN_BP}`, ), bpRecord("Systolic", 130, "2026-05-14 09:00:00 +0200"), ].join("\n"), @@ -230,10 +270,11 @@ describe("the synthetic export fixture", () => { // `export-own-origin.synthetic.xml` is invented, in the shape Apple's DTD // documents. It shows the parser reads a file laid out like an export; it // cannot show that Apple writes a custom metadata key verbatim. - it("leaves out the two marked records and imports the other three", async () => { + it("leaves out the marked record this account holds and imports the other four", async () => { const user = await prisma.user.create({ data: { username: "own-fixture", email: "fx@example.test", role: "USER" }, }); + await seedOwnRow(user.id, "00000000-0000-0000-0000-000000000001", "WEIGHT"); const result = await streamParseExportXml({ xmlPath: join( process.cwd(), @@ -247,9 +288,9 @@ describe("the synthetic export fixture", () => { where: { userId: user.id, type: "WEIGHT", source: "APPLE_HEALTH" }, orderBy: { measuredAt: "asc" }, }); - expect(rows.map((r) => r.value)).toEqual([80.2, 80.4, 80.5]); + expect(rows.map((r) => r.value)).toEqual([80.2, 80.3, 80.4, 80.5]); expect(result.writtenByHealthLog).toEqual({ - byMarker: 2, + byMarker: 1, byExternalId: 0, matchedManual: 0, }); @@ -421,8 +462,6 @@ describe("the manual-mirror lookup stays off the hot path", () => { describe("a sample whose HKExternalUUID is a HealthLog row id", () => { const AT = "2026-05-14 08:00:00 +0200"; - const ext = (id: string) => - ``; async function runWith( seed: (userId: string) => Promise>, @@ -524,11 +563,12 @@ describe("a sample whose HKExternalUUID is a HealthLog row id", () => { describe("each sample left out is counted once", () => { it("does not count a marked record twice when it appears top level and inside its correlation", async () => { const AT = "2026-05-14 08:00:00 +0200"; - const sys = bpRecord("Systolic", 121, AT, `\n ${OWN}\n `); - const dia = bpRecord("Diastolic", 79, AT, `\n ${OWN}\n `); + const sys = bpRecord("Systolic", 121, AT, `\n ${OWN_BP}\n `); + const dia = bpRecord("Diastolic", 79, AT, `\n ${OWN_BP}\n `); const user = await prisma.user.create({ data: { username: "own-once", email: "once@example.test", role: "USER" }, }); + await seedOwnRow(user.id, OWN_SYS_ID, "BLOOD_PRESSURE_SYS"); const result = await streamParseExportXml({ xmlPath: writeXml( [sys, dia, correlation([sys, dia].join("\n"))].join("\n"), @@ -542,7 +582,7 @@ describe("each sample left out is counted once", () => { }); describe("a cycle sample HealthLog wrote itself", () => { - it("is left out of the day-log import", async () => { + it("is left out on a day this account holds, imported on a day it does not", async () => { const user = await prisma.user.create({ data: { username: "own-cycle", @@ -552,11 +592,20 @@ describe("a cycle sample HealthLog wrote itself", () => { timezone: "Europe/Berlin", }, }); + // The day the app mirrored from: a day-log of this account. + await prisma.cycleDayLog.create({ + data: { userId: user.id, date: "2026-03-02", flow: "MEDIUM" }, + }); const flow = (day: string, children = "") => ` ${children}`; const result = await streamParseExportXml({ xmlPath: writeXml( - [flow("2026-03-02", OWN), flow("2026-03-03")].join("\n"), + [ + flow("2026-03-02", MARKER), + flow("2026-03-03"), + // Marked, but no day-log here: a move to a new instance. + flow("2026-03-04", MARKER), + ].join("\n"), ), userId: user.id, userTimezone: "Europe/Berlin", @@ -566,7 +615,119 @@ describe("a cycle sample HealthLog wrote itself", () => { where: { userId: user.id }, orderBy: { date: "asc" }, }); - expect(days.map((d) => d.date)).toEqual(["2026-03-03"]); + expect(days.map((d) => d.date)).toEqual([ + "2026-03-02", + "2026-03-03", + "2026-03-04", + ]); + // The held day stays the account's own: the marked sample did not + // re-import it as an Apple Health day. + expect(days[0].source).toBe("MANUAL"); + expect(result.writtenByHealthLog.byMarker).toBe(1); + }); +}); + +describe("a marked record this account does not hold", () => { + // A move to a new instance without a backup: the export still carries the + // marker on every value the app mirrored, but none of the rows exist here. + it("is imported when the id is missing, unknown or another account's", async () => { + const other = await prisma.user.create({ + data: { username: "own-move-2", email: "mv2@example.test", role: "USER" }, + }); + await seedOwnRow(other.id, "foreign-weight-row", "WEIGHT"); + const user = await prisma.user.create({ + data: { username: "own-move", email: "mv@example.test", role: "USER" }, + }); + const result = await streamParseExportXml({ + xmlPath: writeXml( + [ + weight(80.1, "2026-05-14 08:00:00 +0200", MARKER), + weight( + 80.2, + "2026-05-14 09:00:00 +0200", + `${MARKER}${ext("00000000-0000-4000-8000-000000000009")}`, + ), + weight( + 80.3, + "2026-05-14 10:00:00 +0200", + `${MARKER}${ext("foreign-weight-row")}`, + ), + ].join("\n"), + ), + userId: user.id, + userTimezone: "Europe/Berlin", + prisma, + }); + const rows = await prisma.measurement.findMany({ + where: { userId: user.id, type: "WEIGHT", source: "APPLE_HEALTH" }, + orderBy: { measuredAt: "asc" }, + }); + expect(rows.map((r) => r.value)).toEqual([80.1, 80.2, 80.3]); + expect(result.writtenByHealthLog).toEqual({ + byMarker: 0, + byExternalId: 0, + matchedManual: 0, + }); + }); + + it("is still left out when its id names a deleted row of this account", async () => { + const user = await prisma.user.create({ + data: { + username: "own-deleted", + email: "del@example.test", + role: "USER", + }, + }); + await seedOwnRow(user.id, OWN_WEIGHT_ID, "WEIGHT", new Date()); + const result = await streamParseExportXml({ + xmlPath: writeXml(weight(80.1, "2026-05-14 08:00:00 +0200", OWN)), + userId: user.id, + userTimezone: "Europe/Berlin", + prisma, + }); + const rows = await prisma.measurement.findMany({ + where: { userId: user.id, type: "WEIGHT", source: "APPLE_HEALTH" }, + }); + expect(rows).toHaveLength(0); + expect(result.writtenByHealthLog.byMarker).toBe(1); + }); + + it("adds a marked cumulative sample to its day unless its id is this account's", async () => { + const user = await prisma.user.create({ + data: { username: "own-steps", email: "st@example.test", role: "USER" }, + }); + await seedOwnRow(user.id, "own-steps-row", "ACTIVITY_STEPS"); + const steps = (value: number, at: string, children: string) => + ` ${children}`; + const result = await streamParseExportXml({ + xmlPath: writeXml( + [ + steps( + 700, + "2026-05-14 08:00:00 +0200", + `${MARKER}${ext("own-steps-row")}`, + ), + steps( + 300, + "2026-05-14 09:00:00 +0200", + `${MARKER}${ext("not-a-row")}`, + ), + steps(200, "2026-05-14 10:00:00 +0200", ""), + ].join("\n"), + ), + userId: user.id, + userTimezone: "Europe/Berlin", + prisma, + }); + const rows = await prisma.measurement.findMany({ + where: { + userId: user.id, + type: "ACTIVITY_STEPS", + source: "APPLE_HEALTH", + }, + }); + // 300 + 200: the 700 the app mirrored from this account's row stays out. + expect(rows.map((r) => r.value)).toEqual([500]); expect(result.writtenByHealthLog.byMarker).toBe(1); }); }); diff --git a/tests/integration/backup-preview-memory.test.ts b/tests/integration/backup-preview-memory.test.ts index fbe8eb6f6..569ec6e7b 100644 --- a/tests/integration/backup-preview-memory.test.ts +++ b/tests/integration/backup-preview-memory.test.ts @@ -118,6 +118,11 @@ describe("weekly copy preview under a memory budget", () => { }); it("holds nothing of the record once the producer is done", async () => { + // The backup schema the preview checks each section against is a module + // loaded once per process (about 12 MB, `backup-summary.ts`), not memory + // that grows with the record: load it before the baseline, as a running + // server has long since done. + await import("@/lib/validations/backup"); const baseline = liveHeapBytes(); let peakHeld = 0; let heldAtEnd = 0; @@ -127,7 +132,7 @@ describe("weekly copy preview under a memory budget", () => { rawPeak = Math.max(rawPeak, process.memoryUsage().heapUsed); }, 5); // Wired exactly as the weekly job wires it (`backup-handlers.ts`). - const preview = createBackupPreviewCollector(); + const preview = await createBackupPreviewCollector(); let counts: FullBackupCounts | undefined; try { await storeBackupBlob( diff --git a/tests/integration/backup-round-trip.test.ts b/tests/integration/backup-round-trip.test.ts index 045447121..28e9b83bc 100644 --- a/tests/integration/backup-round-trip.test.ts +++ b/tests/integration/backup-round-trip.test.ts @@ -1495,7 +1495,7 @@ describe("every model the plan claims two-ended survives a real restore", () => // below then covers BOTH, since the row that gets restored comes from the // streamed bytes. let streamedJson = ""; - const previewCollector = createBackupPreviewCollector(); + const previewCollector = await createBackupPreviewCollector(); const streamedCounts = await streamFullBackupJson( prisma, OWNER_ID, @@ -3779,7 +3779,7 @@ describe("every column of every two-ended model survives a real restore", () => const exportedAt = new Date("2026-08-01T00:00:00.000Z"); let json = ""; - const previewCollector = createBackupPreviewCollector(); + const previewCollector = await createBackupPreviewCollector(); const counts = await streamFullBackupJson( prisma, OWNER_ID, diff --git a/tests/integration/day-weight-unanalysed-scale.test.ts b/tests/integration/day-weight-unanalysed-scale.test.ts new file mode 100644 index 000000000..977f32575 --- /dev/null +++ b/tests/integration/day-weight-unanalysed-scale.test.ts @@ -0,0 +1,232 @@ +/** + * The shared pulse day weight (`dayWeightedRows`) must not depend on planner + * statistics. A new account importing a year of pulse lands below the + * autoanalyze threshold, so the first reads run against a table the planner + * believes is empty. The earlier shape joined the CTE to an hour grouping of + * itself; on such a table the planner chose a nested loop and the all-time + * pulse aggregate ran into the 60 s production statement timeout. + * + * This file seeds a year of dense pulse (well over 100k rows) with autovacuum + * off for the table and no ANALYZE, then: + * + * - times the real all-time reader (`readAllTimeExtremes`) and a bare + * day-weighted mean, each well under the timeout; + * - after ANALYZE, compares the weights per day, per frame and per source + * against the earlier self-join SQL kept below as the oracle. + */ +import { afterAll, beforeAll, describe, expect, it } from "vitest"; + +import { getPrismaClient, truncateAllTables } from "./setup"; +import type { Prisma } from "@/generated/prisma/client"; +import { readAllTimeExtremes } from "@/lib/insights/feature-blocks"; +import { + SESSION_DAY_FRAME, + dayWeightedRows, + zoneDayFrame, + type DayFrame, +} from "@/lib/measurements/day-mean"; + +const prisma = getPrismaClient(); +const USER = "day-weight-scale"; +const DAY = 86_400_000; +const MINUTE = 60_000; +const DAYS = 365; +/** Generous against a 60 s production timeout, far above what one scan takes. */ +const BOUND_MS = 15_000; + +const start = + Date.UTC( + new Date().getUTCFullYear(), + new Date().getUTCMonth(), + new Date().getUTCDate(), + ) - + DAYS * DAY; + +/** + * About 330 readings a day, unevenly spread so hours carry different counts, + * plus a second source on every fifth day so the per-source weights differ. + */ +function seedRows(): Prisma.MeasurementCreateManyInput[] { + const rows: Prisma.MeasurementCreateManyInput[] = []; + let i = 0; + for (let d = 0; d < DAYS; d += 1) { + for (let m = 0; m < 24 * 60; m += 4) { + if ((m + d) % 13 === 0) continue; + const hour = Math.floor(m / 60); + // Night hours sparse, day hours dense. + if (hour < 6 && m % 12 !== 0) continue; + rows.push({ + id: `${USER}-a-${i}`, + userId: USER, + type: "PULSE", + unit: "bpm", + source: "APPLE_HEALTH", + value: 55 + ((i * 7 + d) % 50), + measuredAt: new Date(start + d * DAY + m * MINUTE + (i % 3) * 1000), + }); + i += 1; + } + if (d % 5 === 0) { + for (const h of [7, 7, 13, 22]) { + rows.push({ + id: `${USER}-w-${i}`, + userId: USER, + type: "PULSE", + unit: "bpm", + source: "WITHINGS", + value: 60 + (i % 20), + measuredAt: new Date( + start + d * DAY + h * 3_600_000 + (i % 50) * MINUTE, + ), + }); + i += 1; + } + } + } + return rows; +} + +/** The earlier self-join, kept verbatim as the oracle for the weights. */ +function oracleWeightedRows( + source: string, + frame: DayFrame, + bySource: boolean, +): string { + const ts = (column: string) => + frame.kind === "session" + ? column + : `((${column} AT TIME ZONE 'UTC') AT TIME ZONE ${frame.tzSql})`; + const hourOf = (alias: string) => + `date_trunc('hour', ${ts(`${alias}."measured_at"`)})`; + const dayOf = (alias: string) => + `date_trunc('day', ${ts(`${alias}."measured_at"`)})`; + const src = bySource ? `, t."source"` : ""; + const part = bySource ? `, x."source"` : ""; + const join = bySource ? ` AND hw."source" = s."source"` : ""; + return `( + SELECT s.*, + (CASE WHEN (s."type")::text IN ('PULSE') + THEN 1.0 / (hw.n * hw.hours) + ELSE 1.0 END)::double precision AS day_weight + FROM ${source} s + LEFT JOIN ( + SELECT x."type"${part}, x.local_hour, x.n, + COUNT(*) OVER (PARTITION BY x."type"${part}, x.local_day) AS hours + FROM ( + SELECT t."type"${src}, + ${hourOf("t")} AS local_hour, + ${dayOf("t")} AS local_day, + COUNT(*) AS n + FROM ${source} t + WHERE (t."type")::text IN ('PULSE') + GROUP BY ${bySource ? "1, 2, 3, 4" : "1, 2, 3"} + ) x + ) hw + ON hw."type" = s."type"${join} + AND hw.local_hour = ${hourOf("s")} + )`; +} + +const SRC_CTE = `WITH src AS ( + SELECT m."id", m."type", m."source", m."measured_at", m."value" + FROM measurements m + WHERE m."user_id" = $1 AND m."deleted_at" IS NULL AND m."type" = 'PULSE' +)`; + +async function timed(fn: () => Promise): Promise<[T, number]> { + const t0 = performance.now(); + const out = await fn(); + return [out, performance.now() - t0]; +} + +let rowCount = 0; + +beforeAll(async () => { + await truncateAllTables(prisma); + await prisma.$executeRawUnsafe( + "ALTER TABLE measurements SET (autovacuum_enabled = false)", + ); + await prisma.user.create({ data: { id: USER, username: USER } }); + const rows = seedRows(); + rowCount = rows.length; + for (let k = 0; k < rows.length; k += 20_000) { + await prisma.measurement.createMany({ data: rows.slice(k, k + 20_000) }); + } +}, 180_000); + +afterAll(async () => { + await prisma.$executeRawUnsafe( + "ALTER TABLE measurements RESET (autovacuum_enabled)", + ); +}); + +describe("day weight on an unanalysed table", () => { + it("seeds a year of dense pulse", () => { + expect(rowCount).toBeGreaterThan(100_000); + }); + + it("answers the all-time pulse aggregate well under the timeout", async () => { + const [out, ms] = await timed(() => readAllTimeExtremes(USER, ["PULSE"])); + console.info(`[day-weight] readAllTimeExtremes ${ms.toFixed(0)} ms`); + expect(out.get("PULSE")?.mean).toBeGreaterThan(55); + expect(ms).toBeLessThan(BOUND_MS); + }); + + it("answers a bare day-weighted mean well under the timeout", async () => { + const [rows, ms] = await timed(() => + prisma.$queryRawUnsafe>( + `${SRC_CTE} + SELECT (SUM(w."value" * w.day_weight) / SUM(w.day_weight))::double precision AS mean + FROM ${dayWeightedRows("src", zoneDayFrame("$2"), { bySource: true })} w`, + USER, + "Europe/Berlin", + ), + ); + console.info(`[day-weight] zone + bySource mean ${ms.toFixed(0)} ms`); + expect(rows[0].mean).toBeGreaterThan(55); + expect(ms).toBeLessThan(BOUND_MS); + }); +}); + +describe("day weight against the earlier self-join", () => { + beforeAll(async () => { + // Statistics now, so the oracle plans the way it did in production. + await prisma.$executeRawUnsafe("ANALYZE measurements"); + }, 120_000); + + const cases: Array<[string, DayFrame, boolean, unknown[]]> = [ + ["session frame", SESSION_DAY_FRAME, false, []], + ["session frame by source", SESSION_DAY_FRAME, true, []], + ["Berlin frame", zoneDayFrame("$2"), false, ["Europe/Berlin"]], + ["Kolkata frame by source", zoneDayFrame("$2"), true, ["Asia/Kolkata"]], + ]; + + it.each(cases)( + "%s gives every row the same weight", + async (_label, frame, bySource, extra) => { + const query = (sub: string) => + prisma.$queryRawUnsafe>( + `${SRC_CTE} + SELECT w."id" AS id, w.day_weight AS w + FROM ${sub} w + ORDER BY w."id"`, + USER, + ...extra, + ); + const fresh = await query(dayWeightedRows("src", frame, { bySource })); + const oracle = await query(oracleWeightedRows("src", frame, bySource)); + expect(fresh.length).toBe(rowCount); + expect(fresh.length).toBe(oracle.length); + let mismatches = 0; + for (let k = 0; k < fresh.length; k += 1) { + if ( + fresh[k].id !== oracle[k].id || + Math.abs(fresh[k].w - oracle[k].w) > 1e-12 + ) { + mismatches += 1; + } + } + expect(mismatches).toBe(0); + }, + ); +}); diff --git a/tests/integration/mcp-metric-series-source-snapshot.test.ts b/tests/integration/mcp-metric-series-source-snapshot.test.ts index 3759189dc..ab95bc21a 100644 --- a/tests/integration/mcp-metric-series-source-snapshot.test.ts +++ b/tests/integration/mcp-metric-series-source-snapshot.test.ts @@ -218,13 +218,10 @@ beforeAll(async () => { // The mood feature warms its rollup tier fire-and-forget on first read; // settle it now so both builds read the same tier. await ensureUserMoodRollupsFresh(userId); - // Planner statistics for the rows just written. Without them the - // all-time pulse aggregate (`readAllTimeExtremes`, day-weighted for pulse) - // plans against an empty table and runs for ~14 s per call until - // autovacuum's analyze happens to land, which on a loaded runner took the - // first case past its 60 s budget. A production table always has - // statistics, so this measures what production runs. - await prisma.$executeRawUnsafe("ANALYZE measurements, measurement_rollups"); + // No ANALYZE here, on purpose: a fresh import below the autoanalyze + // threshold is what production runs too, and the all-time pulse aggregate + // must not depend on planner statistics + // (`day-weight-unanalysed-scale.test.ts`). }, 120_000); describe("get_metric_series built from one source's rows", () => {