feat(labs): backfill job linking legacy readings to the catalog

Add a one-shot pg-boss job that heals the per-row free-text divergence:
for each user it groups un-linked live readings by lower(analyte),
mints (or reuses) one Biomarker per group from the most-recent report's
spelling/unit/range, and stamps biomarkerId on every row. Registered in
allQueues with a boot-time discovery enqueue; idempotent across reboots
(a completed pass drops the user from the discovery set).

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");
+  });
+});

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<string, LegacyRow[]>();
+  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),
+    };
+  }
+}

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<LabBiomarkerBackfillPayload>(
+    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.