diff --git a/CHANGELOG.md b/CHANGELOG.md index d49fe80a1..3f1823ec9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,45 @@ # Changelog +## [1.4.35.1] — 2026-05-17 — Auto-converging rollup backfill on worker boot + +Follow-on to v1.4.35. Removes the operator action that was implicit +in the foundation release: instead of self-hosters having to run +`pnpm tsx scripts/backfill-rollups.ts` after upgrading, the worker +boot now discovers any account with measurements but no rollup +coverage and enqueues a one-shot full fold via pg-boss. Idempotent +across reboots (the discovery query only matches uncovered users) and +zero reader-path impact (serial concurrency, runs in the background). + +### Boot-time backfill + +- **New `rollup-full-backfill` queue** (`ROLLUP_FULL_BACKFILL_QUEUE`, + concurrency = 1) carrying `{ userId, enqueuedAt }`. The handler + invokes `recomputeUserRollups(userId)` with the default 5-year + window across all four granularities — equivalent to the manual + CLI run, but inside the worker process so no `tsx` install + gymnastics in production images. +- **`enqueueBootTimeRollupBackfill`** runs once at + `startReminderWorker` boot after every `boss.work` subscription is + in place. The discovery query finds users with at least one row in + `measurements` and zero rows in `measurement_rollups`, then sends + one job per uncovered account with `singletonKey: + boot-backfill|{userId}` so a fast restart while a fold is queued + doesn't pile up duplicates. +- **Best-effort:** any error during discovery is captured in the + return value and logged through `workerLog`; the worker boot never + fails because the backfill missed. + +### Tests + +- 5 new unit cases pin the helper's contract (silent no-op without + boss, one job per uncovered user, singleton-key coalesce counted + as `skipped`, error surfacing, zero-users path). 2 new integration + cases verify against the real testcontainer: only the uncovered + user lands on the queue when a third already-folded user co-exists; + zero jobs enqueued when every account is already covered. +- Unit suite 4280 → 4285. Integration suite 228 → 230. `pnpm + typecheck` + `pnpm lint` clean. + ## [1.4.35] — 2026-05-17 — Persistent measurement rollups + partial read-swap Foundation release for the persistent `measurement_rollups` cache diff --git a/package.json b/package.json index e4ddb29d7..8b57e2d40 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "healthlog", - "version": "1.4.35", + "version": "1.4.35.1", "description": "Self-hosted personal-health-tracking PWA with Withings integration, AI insights, and doctor-report PDF export.", "license": "AGPL-3.0-only", "homepage": "https://healthlog.dev", diff --git a/src/lib/jobs/reminder-worker.ts b/src/lib/jobs/reminder-worker.ts index 282495861..0d0b76fd7 100644 --- a/src/lib/jobs/reminder-worker.ts +++ b/src/lib/jobs/reminder-worker.ts @@ -53,9 +53,13 @@ import { type AppleHealthImportPayload, } from "@/lib/jobs/apple-health-import-worker"; import { + ROLLUP_FULL_BACKFILL_QUEUE, + ROLLUP_FULL_BACKFILL_CONCURRENCY, ROLLUP_RECOMPUTE_QUEUE, ROLLUP_RECOMPUTE_CONCURRENCY, + enqueueBootTimeRollupBackfill, recomputeUserRollups, + type RollupFullBackfillPayload, type RollupRecomputePayload, } from "@/lib/measurements/rollups"; import { expireStaleInUseItems } from "@/lib/medications/inventory/service"; @@ -1466,6 +1470,7 @@ export async function startReminderWorker() { MEDICATION_INVENTORY_EXPIRE_QUEUE, APPLE_HEALTH_IMPORT_QUEUE, ROLLUP_RECOMPUTE_QUEUE, + ROLLUP_FULL_BACKFILL_QUEUE, ]; for (const q of allQueues) { @@ -1666,6 +1671,54 @@ export async function startReminderWorker() { }, ); + // v1.4.35.1 — boot-time full-fold worker. The boot enqueue helper + // below sends one job per uncovered user; this handler runs the + // full `recomputeUserRollups` against the default 5-year window + // across every granularity. Serial concurrency so the populator + // never crowds the dashboard request pool. + await boss.work( + ROLLUP_FULL_BACKFILL_QUEUE, + { localConcurrency: ROLLUP_FULL_BACKFILL_CONCURRENCY }, + async (jobs) => { + for (const job of jobs) { + const { userId } = job.data; + const { rowsUpserted, durationMs } = await recomputeUserRollups(userId); + workerLog( + "info", + `[rollup-full-backfill] user=${userId} rows=${rowsUpserted} duration=${durationMs}ms`, + ); + } + }, + ); + + // v1.4.35.1 — fire-and-forget boot-time backfill discovery. Finds + // every user with measurements but no rollup coverage and enqueues + // a full-fold per account. Idempotent across reboots: the discovery + // query only matches accounts with zero rollup rows, so once a + // fold completes the user drops off the list. Errors are returned + // through the helper's result value — the worker boot never fails + // because of a backfill miss. + try { + const { enqueued, skipped, error } = await enqueueBootTimeRollupBackfill(); + if (error) { + workerLog( + "error", + `[rollup-full-backfill] boot discovery failed: ${error}`, + ); + } else if (enqueued > 0 || skipped > 0) { + workerLog( + "info", + `[rollup-full-backfill] boot discovery: enqueued=${enqueued} skipped=${skipped}`, + ); + } + } catch (err) { + workerLog( + "error", + "[rollup-full-backfill] boot discovery threw an unexpected error", + err, + ); + } + return boss; } diff --git a/src/lib/measurements/__tests__/rollups.test.ts b/src/lib/measurements/__tests__/rollups.test.ts index 44eac9d2c..305ad249f 100644 --- a/src/lib/measurements/__tests__/rollups.test.ts +++ b/src/lib/measurements/__tests__/rollups.test.ts @@ -72,8 +72,10 @@ void mocks.findMany; void queryRaw; import { + ROLLUP_FULL_BACKFILL_QUEUE, ROLLUP_RECOMPUTE_QUEUE, collapseToTypeDayKeys, + enqueueBootTimeRollupBackfill, enqueueRollupRecompute, ensureUserRollupsFresh, recomputeBucketsForMeasurement, @@ -280,3 +282,70 @@ describe("ensureUserRollupsFresh", () => { expect(result.recomputed).toBe(false); }); }); + +describe("enqueueBootTimeRollupBackfill", () => { + it("is a silent no-op when no boss is attached", async () => { + getGlobalBossMock.mockReturnValue(null); + const result = await enqueueBootTimeRollupBackfill(); + expect(result).toEqual({ enqueued: 0, skipped: 0, error: null }); + expect(queryRaw).not.toHaveBeenCalled(); + expect(bossSend).not.toHaveBeenCalled(); + }); + + it("enqueues one full-fold job per user who has measurements but no rollups", async () => { + getGlobalBossMock.mockReturnValue({ send: bossSend }); + queryRaw.mockResolvedValueOnce([ + { id: "user-a" }, + { id: "user-b" }, + { id: "user-c" }, + ]); + bossSend + .mockResolvedValueOnce("job-a") + .mockResolvedValueOnce("job-b") + .mockResolvedValueOnce("job-c"); + + const result = await enqueueBootTimeRollupBackfill(); + + expect(result).toEqual({ enqueued: 3, skipped: 0, error: null }); + expect(bossSend).toHaveBeenCalledTimes(3); + // The queue name is the boot-fold queue, not the per-bucket queue. + for (const call of bossSend.mock.calls) { + expect(call[0]).toBe(ROLLUP_FULL_BACKFILL_QUEUE); + } + // Singleton key per user — coalesces across rapid reboots. + expect(bossSend.mock.calls[0][2].singletonKey).toBe("boot-backfill|user-a"); + expect(bossSend.mock.calls[1][2].singletonKey).toBe("boot-backfill|user-b"); + expect(bossSend.mock.calls[2][2].singletonKey).toBe("boot-backfill|user-c"); + }); + + it("counts a `boss.send` returning null as 'skipped' (singleton coalesce)", async () => { + getGlobalBossMock.mockReturnValue({ send: bossSend }); + queryRaw.mockResolvedValueOnce([{ id: "user-a" }, { id: "user-b" }]); + bossSend.mockResolvedValueOnce(null).mockResolvedValueOnce("job-b"); + + const result = await enqueueBootTimeRollupBackfill(); + + expect(result).toEqual({ enqueued: 1, skipped: 1, error: null }); + }); + + it("returns the error message when the discovery query throws", async () => { + getGlobalBossMock.mockReturnValue({ send: bossSend }); + queryRaw.mockRejectedValueOnce(new Error("pool exhausted")); + + const result = await enqueueBootTimeRollupBackfill(); + + expect(result.enqueued).toBe(0); + expect(result.error).toBe("pool exhausted"); + expect(bossSend).not.toHaveBeenCalled(); + }); + + it("returns { enqueued: 0 } when no users need backfill", async () => { + getGlobalBossMock.mockReturnValue({ send: bossSend }); + queryRaw.mockResolvedValueOnce([]); + + const result = await enqueueBootTimeRollupBackfill(); + + expect(result).toEqual({ enqueued: 0, skipped: 0, error: null }); + expect(bossSend).not.toHaveBeenCalled(); + }); +}); diff --git a/src/lib/measurements/rollups.ts b/src/lib/measurements/rollups.ts index fed8be94d..1f488f2e5 100644 --- a/src/lib/measurements/rollups.ts +++ b/src/lib/measurements/rollups.ts @@ -68,6 +68,28 @@ export interface RollupRecomputePayload { enqueuedAt: string; } +/** + * v1.4.35.1 — boot-time full-fold queue. The write-path hooks keep + * the DAY bucket current per measurement, and `ensureUserRollupsFresh` + * warm-on-read covers the trailing 90-day DAY window on cold mount, + * but neither path folds the deep WEEK / MONTH / YEAR history for an + * account that has measurements pre-dating the v1.4.35 deploy. This + * queue exists so the worker boot can enqueue one full + * `recomputeUserRollups(userId)` per uncovered user — no operator + * action needed on self-hosted instances. + */ +export const ROLLUP_FULL_BACKFILL_QUEUE = "rollup-full-backfill"; + +/** Boot-fold worker concurrency. Serial so we don't crowd the pool. */ +export const ROLLUP_FULL_BACKFILL_CONCURRENCY = 1; + +/** Payload the boot-time backfill helper enqueues. */ +export interface RollupFullBackfillPayload { + userId: string; + /** Wall-clock kick-off for debugging only. */ + enqueuedAt: string; +} + /** Postgres `date_trunc` unit per granularity. */ const DATE_TRUNC_UNIT: Record = { DAY: "day", @@ -600,6 +622,85 @@ export async function ensureUserRollupsFresh( } } +/** + * v1.4.35.1 — boot-time enqueue helper. Finds users with measurements + * but zero rollup coverage and enqueues one full-fold job per account + * onto `ROLLUP_FULL_BACKFILL_QUEUE`. Designed to be called once at + * worker boot so every self-hosted instance auto-converges on the + * persistent rollup tier without operator intervention. + * + * Idempotent across reboots: + * - the discovery query only matches accounts with no rollup rows at + * all, so once the fold completes the user drops off the list and + * subsequent boots skip them + * - pg-boss `singletonKey` coalesces duplicate sends within the + * queue, so a fast restart while a backfill is queued doesn't + * double up + * + * Best-effort: errors are swallowed and reported through the return + * value so the worker boot never fails because of a backfill miss. + */ +export async function enqueueBootTimeRollupBackfill(): Promise<{ + enqueued: number; + skipped: number; + error: string | null; +}> { + const boss = getGlobalBoss(); + if (!boss) { + return { enqueued: 0, skipped: 0, error: null }; + } + + try { + // Users with at least one measurement but no rollup row yet. The + // `LIMIT 1` inside each `EXISTS` keeps the planner from materialising + // the full child relation just to answer existence — important on + // power-user accounts with 100k+ measurements. + const users = await prisma.$queryRaw>` + SELECT u."id" + FROM users u + WHERE EXISTS ( + SELECT 1 FROM measurements m WHERE m."user_id" = u."id" LIMIT 1 + ) + AND NOT EXISTS ( + SELECT 1 FROM measurement_rollups r WHERE r."user_id" = u."id" LIMIT 1 + ) + `; + + if (users.length === 0) { + return { enqueued: 0, skipped: 0, error: null }; + } + + let enqueued = 0; + let skipped = 0; + for (const { id } of users) { + const payload: RollupFullBackfillPayload = { + userId: id, + enqueuedAt: new Date().toISOString(), + }; + const jobId = await boss.send(ROLLUP_FULL_BACKFILL_QUEUE, payload, { + retryLimit: 3, + retryDelay: 60, + retryBackoff: true, + // Coalesce: if a backfill for this user is already queued and + // we restart, pg-boss returns null instead of duplicating. + singletonKey: `boot-backfill|${id}`, + }); + if (jobId) { + enqueued += 1; + } else { + skipped += 1; + } + } + return { enqueued, skipped, error: null }; + } catch (err) { + return { + enqueued: 0, + skipped: 0, + error: err instanceof Error ? err.message : String(err), + }; + } +} + /** * Collapse a list of `(type, day)` keys to the set of distinct * `(type, dayStart)` pairs. Used by the batch-ingest hooks so a 500- diff --git a/tests/integration/measurement-rollups.test.ts b/tests/integration/measurement-rollups.test.ts index e10ee40a4..4b66cab29 100644 --- a/tests/integration/measurement-rollups.test.ts +++ b/tests/integration/measurement-rollups.test.ts @@ -27,6 +27,8 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { getPrismaClient, truncateAllTables } from "./setup"; import { + ROLLUP_FULL_BACKFILL_QUEUE, + enqueueBootTimeRollupBackfill, recomputeBucketsForMeasurement, recomputeUserRollups, ALL_GRANULARITIES, @@ -38,12 +40,25 @@ vi.mock("@/lib/db-compat", () => ({ ensureDbCompatibility: vi.fn().mockResolvedValue(undefined), })); +// Stub boss-instance with a vi.fn so we can swap the return per test. +// The integration suite runs with `isolate: false`, so the mock state +// is shared with sibling files (e.g. admin-backups-audit.test.ts also +// mocks boss-instance). Re-setting the implementation inside +// beforeEach restores our default before each case. vi.mock("@/lib/jobs/boss-instance", () => ({ - getGlobalBoss: () => null, // no pg-boss in the test container + getGlobalBoss: vi.fn(() => null), })); +import { getGlobalBoss } from "@/lib/jobs/boss-instance"; + +const bossSend = vi.fn(); + beforeEach(async () => { await truncateAllTables(getPrismaClient()); + bossSend.mockReset(); + // Default: pg-boss not attached. Individual tests that exercise the + // enqueue path override this with a stub that records sends. + vi.mocked(getGlobalBoss).mockReturnValue(null as never); }); afterEach(() => { @@ -503,4 +518,116 @@ describe("measurement rollups — integration", () => { // = 81.6667 → round2 = 81.67. expect(after.summaries.WEIGHT.mean).toBe(81.67); }); + + // ─── v1.4.35.1 boot-time backfill ────────────────────────── + // + // Self-hosted instances upgrading to v1.4.35 onwards must auto- + // converge on the persistent rollup tier without operator action. + // The discovery query matches users with measurements but zero + // rollup rows; one full-fold job is enqueued per such account. + + it("enqueues one full-fold job per user with measurements but no rollups", async () => { + const prisma = getPrismaClient(); + // Three users: two with measurements (one already-folded, one + // uncovered), one with no measurements at all (skipped entirely). + const covered = await prisma.user.create({ + data: { + username: "boot-backfill-covered", + email: "boot-backfill-covered@example.test", + role: "USER", + }, + }); + const uncovered = await prisma.user.create({ + data: { + username: "boot-backfill-uncovered", + email: "boot-backfill-uncovered@example.test", + role: "USER", + }, + }); + const empty = await prisma.user.create({ + data: { + username: "boot-backfill-empty", + email: "boot-backfill-empty@example.test", + role: "USER", + }, + }); + void empty; + + const measuredAt = new Date("2026-04-01T08:00:00.000Z"); + await prisma.measurement.createMany({ + data: [ + { + userId: covered.id, + type: "WEIGHT", + value: 80, + unit: "kg", + source: "MANUAL", + measuredAt, + }, + { + userId: uncovered.id, + type: "WEIGHT", + value: 80, + unit: "kg", + source: "MANUAL", + measuredAt, + }, + ], + }); + + // Pre-fold the covered user so the discovery query skips them. + await recomputeUserRollups(covered.id, { granularities: ["DAY"] }); + + bossSend.mockResolvedValue("job-id"); + vi.mocked(getGlobalBoss).mockReturnValue({ send: bossSend } as never); + + const result = await enqueueBootTimeRollupBackfill(); + + // Only the uncovered user lands on the queue. The covered user + // already has DAY rollups; the empty user has no measurements. + expect(result.enqueued).toBe(1); + expect(result.skipped).toBe(0); + expect(result.error).toBeNull(); + expect(bossSend).toHaveBeenCalledTimes(1); + expect(bossSend.mock.calls[0][0]).toBe(ROLLUP_FULL_BACKFILL_QUEUE); + expect(bossSend.mock.calls[0][1]).toMatchObject({ userId: uncovered.id }); + expect(bossSend.mock.calls[0][2].singletonKey).toBe( + `boot-backfill|${uncovered.id}`, + ); + + // Reset boss stub for downstream tests. + vi.mocked(getGlobalBoss).mockReturnValue(null as never); + }); + + it("returns { enqueued: 0 } when every user is already covered", async () => { + const prisma = getPrismaClient(); + const user = await prisma.user.create({ + data: { + username: "boot-backfill-allcovered", + email: "boot-backfill-allcovered@example.test", + role: "USER", + }, + }); + await prisma.measurement.create({ + data: { + userId: user.id, + type: "WEIGHT", + value: 80, + unit: "kg", + source: "MANUAL", + measuredAt: new Date("2026-04-01T08:00:00.000Z"), + }, + }); + await recomputeUserRollups(user.id, { granularities: ["DAY"] }); + + bossSend.mockResolvedValue("job-id"); + vi.mocked(getGlobalBoss).mockReturnValue({ send: bossSend } as never); + + const result = await enqueueBootTimeRollupBackfill(); + + expect(result.enqueued).toBe(0); + expect(bossSend).not.toHaveBeenCalled(); + + vi.mocked(getGlobalBoss).mockReturnValue(null as never); + }); });