diff --git a/src/lib/jobs/__tests__/lab-biomarker-backfill.test.ts b/src/lib/jobs/__tests__/lab-biomarker-backfill.test.ts new file mode 100644 index 000000000..463b9a0aa --- /dev/null +++ b/src/lib/jobs/__tests__/lab-biomarker-backfill.test.ts @@ -0,0 +1,63 @@ +import { describe, expect, it } from "vitest"; + +import { groupRowsByAnalyte } from "../lab-biomarker-backfill"; + +function row( + id: string, + analyte: string, + unit: string, + takenAt: string, + opts: { low?: number | null; high?: number | null; panel?: string | null } = {}, +) { + return { + id, + analyte, + unit, + referenceLow: opts.low ?? null, + referenceHigh: opts.high ?? null, + panel: opts.panel ?? null, + takenAt: new Date(takenAt), + }; +} + +describe("groupRowsByAnalyte", () => { + it("collapses case/spacing variants of the same analyte into one group", () => { + const groups = groupRowsByAnalyte([ + row("a", "LDL", "mg/dL", "2026-01-01T00:00:00Z"), + row("b", "ldl", "mg/dl", "2026-02-01T00:00:00Z"), + row("c", " LDL ", "mg/dL", "2026-03-01T00:00:00Z"), + ]); + expect(groups).toHaveLength(1); + expect(groups[0].ids.sort()).toEqual(["a", "b", "c"]); + }); + + it("uses the most-recently-taken reading for the canonical name/unit/range", () => { + // Same analyte key ("ldl"), case + unit + range drift across reports — + // the newest report's spelling/unit/range wins. + const groups = groupRowsByAnalyte([ + row("old", "ldl", "mg/dl", "2026-01-01T00:00:00Z", { high: 100 }), + row("new", "LDL", "mg/dL", "2026-06-01T00:00:00Z", { high: 116 }), + ]); + expect(groups).toHaveLength(1); + expect(groups[0].name).toBe("LDL"); + expect(groups[0].unit).toBe("mg/dL"); + expect(groups[0].referenceHigh).toBe(116); + }); + + it("keeps distinct analytes in separate groups", () => { + const groups = groupRowsByAnalyte([ + row("a", "LDL", "mg/dL", "2026-01-01T00:00:00Z"), + row("b", "HDL", "mg/dL", "2026-01-01T00:00:00Z"), + ]); + expect(groups).toHaveLength(2); + }); + + it("skips a row with a blank analyte", () => { + const groups = groupRowsByAnalyte([ + row("a", " ", "mg/dL", "2026-01-01T00:00:00Z"), + row("b", "LDL", "mg/dL", "2026-01-01T00:00:00Z"), + ]); + expect(groups).toHaveLength(1); + expect(groups[0].name).toBe("LDL"); + }); +}); diff --git a/src/lib/jobs/lab-biomarker-backfill.ts b/src/lib/jobs/lab-biomarker-backfill.ts new file mode 100644 index 000000000..a9cd8d6bb --- /dev/null +++ b/src/lib/jobs/lab-biomarker-backfill.ts @@ -0,0 +1,217 @@ +/** + * v1.18.1 — one-shot backfill that links legacy `LabResult` rows to a + * user-scoped `Biomarker` catalog entry. + * + * Before the catalog existed, every lab reading carried its own free-text + * `analyte` / `unit` / `reference*`. Recording "LDL" three times forked the + * "same" marker into "LDL" / "ldl" / "LDL-C" with possibly diverging units + * and ranges. This pass heals that at the source: for each user, it groups + * every un-linked live reading by `lower(analyte)`, creates ONE `Biomarker` + * per group (using the most-recently-taken reading's spelling / unit / + * reference bounds — "last lab report wins"), and stamps `biomarkerId` on + * every row in the group. + * + * Idempotent across reboots: the discovery predicate is "the user has at + * least one live reading with `biomarkerId IS NULL`". A completed pass leaves + * every reading linked, so the user drops out of the discovery set. A reboot + * mid-pass re-runs from scratch — `Biomarker` creation upserts on + * `(userId, name)` and the row-link `updateMany` is `biomarkerId: null`-gated, + * so the result converges. A reading whose marker already exists (e.g. the + * user defined it by hand) re-uses that row rather than minting a duplicate. + * + * Modelled on `sleep-timeline-backfill.ts` / `whoop-backfill.ts`. The queue + * name MUST be registered in `allQueues` in `src/lib/jobs/reminder-worker.ts` + * or pg-boss never provisions it and the boot enqueue silently never drains + * (the v1.4.37 dead-queue class). + */ +import { prisma } from "@/lib/db"; +import { getGlobalBoss } from "@/lib/jobs/boss-instance"; +import { annotate } from "@/lib/logging/context"; + +export const LAB_BIOMARKER_BACKFILL_QUEUE = "lab-biomarker-backfill"; + +/** + * Serial concurrency — the per-user pass is a short burst of grouped writes; + * concurrency-1 keeps it off the request pool, matching the other backfill + * queues. + */ +export const LAB_BIOMARKER_BACKFILL_CONCURRENCY = 1; + +export interface LabBiomarkerBackfillPayload { + userId: string; + enqueuedAt: string; +} + +/** A single legacy reading the grouping reads. */ +interface LegacyRow { + id: string; + analyte: string; + unit: string; + referenceLow: number | null; + referenceHigh: number | null; + panel: string | null; + takenAt: Date; +} + +interface AnalyteGroup { + /** Canonical spelling = the most-recently-taken reading's `analyte`. */ + name: string; + unit: string; + referenceLow: number | null; + referenceHigh: number | null; + panel: string | null; + ids: string[]; +} + +/** + * Reduce the user's un-linked rows into one group per `lower(analyte)`. The + * canonical name / unit / range come from the most-recently-TAKEN reading in + * each group ("last lab report wins"); every row id rides into the group so a + * single `updateMany` can link them. Exported for unit testing. + */ +export function groupRowsByAnalyte(rows: LegacyRow[]): AnalyteGroup[] { + const byKey = new Map(); + for (const r of rows) { + const key = r.analyte.trim().toLowerCase(); + if (!key) continue; + const bucket = byKey.get(key); + if (bucket) bucket.push(r); + else byKey.set(key, [r]); + } + + const groups: AnalyteGroup[] = []; + for (const bucket of byKey.values()) { + // Newest reading defines the canonical name / unit / range. + const newest = bucket.reduce((a, b) => + b.takenAt.getTime() >= a.takenAt.getTime() ? b : a, + ); + groups.push({ + name: newest.analyte.trim(), + unit: newest.unit, + referenceLow: newest.referenceLow, + referenceHigh: newest.referenceHigh, + panel: newest.panel, + ids: bucket.map((r) => r.id), + }); + } + return groups; +} + +/** + * Backfill one user: group un-linked live readings, create/reuse a Biomarker + * per group, link the rows. Returns the count of markers touched + rows + * linked. + */ +export async function runLabBiomarkerBackfillForUser( + userId: string, +): Promise<{ markers: number; linked: number }> { + const rows = await prisma.labResult.findMany({ + where: { userId, deletedAt: null, biomarkerId: null }, + select: { + id: true, + analyte: true, + unit: true, + referenceLow: true, + referenceHigh: true, + panel: true, + takenAt: true, + }, + }); + + if (rows.length === 0) { + return { markers: 0, linked: 0 }; + } + + const groups = groupRowsByAnalyte(rows); + + let markers = 0; + let linked = 0; + for (const group of groups) { + // Re-use a hand-defined marker of the same name if one exists; otherwise + // mint it. `(userId, name)` is unique, so a concurrent re-run converges. + const biomarker = await prisma.biomarker.upsert({ + where: { userId_name: { userId, name: group.name } }, + // An existing marker keeps its own (possibly user-tuned) unit / range. + update: {}, + create: { + userId, + name: group.name, + unit: group.unit, + lowerBound: group.referenceLow, + upperBound: group.referenceHigh, + panel: group.panel, + }, + }); + markers += 1; + + const { count } = await prisma.labResult.updateMany({ + where: { id: { in: group.ids }, userId, biomarkerId: null }, + data: { biomarkerId: biomarker.id }, + }); + linked += count; + } + + annotate({ + action: { + name: "labs.biomarker.backfill.complete", + details: { markers, linked }, + }, + }); + return { markers, linked }; +} + +/** + * Boot-time discovery. Finds every user holding at least one un-linked live + * lab reading and enqueues one job each. Idempotent across reboots: a + * completed pass links every reading, dropping the user from the discovery + * set. `singletonKey` coalesces duplicate sends. + * + * Best-effort: errors come back through the result value so worker boot never + * fails because of a backfill miss. + */ +export async function enqueueBootTimeLabBiomarkerBackfill(): Promise<{ + enqueued: number; + skipped: number; + error: string | null; +}> { + const boss = getGlobalBoss(); + if (!boss) { + return { enqueued: 0, skipped: 0, error: null }; + } + + try { + const rows = await prisma.labResult.findMany({ + where: { deletedAt: null, biomarkerId: null }, + select: { userId: true }, + distinct: ["userId"], + }); + + if (rows.length === 0) { + return { enqueued: 0, skipped: 0, error: null }; + } + + let enqueued = 0; + let skipped = 0; + for (const { userId } of rows) { + const payload: LabBiomarkerBackfillPayload = { + userId, + enqueuedAt: new Date().toISOString(), + }; + const jobId = await boss.send(LAB_BIOMARKER_BACKFILL_QUEUE, payload, { + retryLimit: 3, + retryDelay: 60, + retryBackoff: true, + singletonKey: `lab-biomarker-backfill|${userId}`, + }); + 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), + }; + } +} diff --git a/src/lib/jobs/reminder-worker.ts b/src/lib/jobs/reminder-worker.ts index db6b4d5b2..672614d04 100644 --- a/src/lib/jobs/reminder-worker.ts +++ b/src/lib/jobs/reminder-worker.ts @@ -28,6 +28,13 @@ import { enqueueBootTimeSleepTimelineBackfill, type SleepTimelineBackfillPayload, } from "@/lib/jobs/sleep-timeline-backfill"; +import { + LAB_BIOMARKER_BACKFILL_QUEUE, + LAB_BIOMARKER_BACKFILL_CONCURRENCY, + runLabBiomarkerBackfillForUser, + enqueueBootTimeLabBiomarkerBackfill, + type LabBiomarkerBackfillPayload, +} from "@/lib/jobs/lab-biomarker-backfill"; import { reportWorkerError } from "@/lib/jobs/report-worker-error"; import { markWorkerStarted, recordError } from "@/lib/jobs/worker-status"; import { setGlobalBoss } from "@/lib/jobs/boss-instance"; @@ -656,6 +663,11 @@ export async function startReminderWorker() { // pg-boss never provisions it and the boot enqueue silently never drains // (the v1.4.37 dead-queue class). SLEEP_TIMELINE_BACKFILL_QUEUE, + // v1.18.1 — one-shot backfill that links legacy free-text lab readings to + // a user-scoped Biomarker catalog entry (group by `lower(analyte)`). The + // queue MUST be registered here or pg-boss never provisions it and the + // boot enqueue silently never drains (the v1.4.37 dead-queue class). + LAB_BIOMARKER_BACKFILL_QUEUE, // v1.17.0 — Nightscout CGM poll sync. Poll-only (no webhook, no OAuth, no // backfill queue — the hourly window walks the recent SGV set). The queue // MUST be registered here or pg-boss never provisions it and the schedule @@ -1114,6 +1126,25 @@ export async function startReminderWorker() { } }, ); + // v1.18.1 — one-shot lab-biomarker backfill. The boot enqueue below sends + // one job per user holding un-linked free-text lab readings; this handler + // groups them by `lower(analyte)`, mints/reuses a Biomarker per group, and + // links the rows. Idempotent — a re-run links only what is still un-linked. + await boss.work( + LAB_BIOMARKER_BACKFILL_QUEUE, + { localConcurrency: LAB_BIOMARKER_BACKFILL_CONCURRENCY }, + async (jobs) => { + for (const job of jobs) { + const { userId } = job.data; + const { markers, linked } = + await runLabBiomarkerBackfillForUser(userId); + workerLog( + "info", + `[lab-biomarker-backfill] user=${userId} markers=${markers} linked=${linked}`, + ); + } + }, + ); // v1.17.0 — Nightscout CGM poll-cohort sync. The hourly cron tick (no // `userId`) walks every configured instance; one user's unreachable host is // warned, not fatal. @@ -1977,6 +2008,33 @@ export async function startReminderWorker() { ); } + // v1.18.1 — fire-and-forget boot discovery for the one-shot lab-biomarker + // backfill. Finds every user holding an un-linked live lab reading and + // enqueues one job each. Idempotent across reboots: a completed pass links + // every reading, dropping the user from the discovery set. Errors come back + // through the helper's result value — worker boot never fails on a miss. + try { + const { enqueued, skipped, error } = + await enqueueBootTimeLabBiomarkerBackfill(); + if (error) { + workerLog( + "error", + `[lab-biomarker-backfill] boot discovery failed: ${error}`, + ); + } else { + workerLog( + "info", + `[lab-biomarker-backfill] boot discovery: enqueued=${enqueued} skipped=${skipped}`, + ); + } + } catch (err) { + workerLog( + "error", + "[lab-biomarker-backfill] boot discovery threw an unexpected error", + err, + ); + } + // v1.7.0 — fire-and-forget boot discovery for the daily-mean // consolidation pass. Finds every user holding live per-sample // high-frequency mean-type rows and enqueues one job per account.