diff --git a/src/__tests__/labs-module-ai-gate-guard.test.ts b/src/__tests__/labs-module-ai-gate-guard.test.ts index d798e69ef..70ca38c36 100644 --- a/src/__tests__/labs-module-ai-gate-guard.test.ts +++ b/src/__tests__/labs-module-ai-gate-guard.test.ts @@ -65,6 +65,8 @@ const NOT_MODEL_INPUT: Record = { "links stored rows to their biomarker; no model is involved", "src/lib/jobs/lab-biomarker-backfill.ts#enqueueBootTimeLabBiomarkerBackfill": "finds the accounts whose rows need linking; no model is involved", + "src/lib/jobs/lab-document-date-repair.ts#candidateQuery": + "finds document lab values filed at their confirmation moment to re-date them; no model is involved", "src/lib/jobs/soft-delete-retention.ts#SOFT_DELETE_RETENTION": "purges rows past the trash window; no model is involved", "src/lib/insights/correlation-channel-series.ts#fetchLabDraws": diff --git a/src/lib/documents/commit.ts b/src/lib/documents/commit.ts index 17bae2712..97633752b 100644 --- a/src/lib/documents/commit.ts +++ b/src/lib/documents/commit.ts @@ -66,7 +66,7 @@ export interface CommittedRecordRef { } /** A stated day, or null when it is not a real calendar date. */ -function statedDay(date: string | null): string | null { +export function statedDay(date: string | null): string | null { return date && isCalendarDateKey(date) ? date : null; } @@ -97,7 +97,7 @@ export interface FactDocumentDates { * no date beside each value, and every reading used to be filed under the day * it was confirmed. */ -function documentDay(dates: FactDocumentDates | null): string | null { +export function documentDay(dates: FactDocumentDates | null): string | null { const day = dates?.reportDate ?? dates?.documentDate ?? null; return day ? dateOnlyKey(day) : null; } diff --git a/src/lib/jobs/__tests__/queue-policy-tables.test.ts b/src/lib/jobs/__tests__/queue-policy-tables.test.ts index 7a728b6fb..41c5b2f3d 100644 --- a/src/lib/jobs/__tests__/queue-policy-tables.test.ts +++ b/src/lib/jobs/__tests__/queue-policy-tables.test.ts @@ -112,6 +112,15 @@ describe("maintenance group queue policies", () => { expectPolicy("maintenance", queue, policy); }); + // One-shot repairs on a fixed singleton key: a boot during a run must be + // able to queue the next pass, so only queued duplicates collapse. + it.each([["lab-document-date-repair", "short"]])( + "%s is %s", + (queue, policy) => { + expectPolicy("maintenance", queue, policy); + }, + ); + it("environment-fetch is deliberately left on standard", () => { // Its explicit-range backfill sends with no key at all, on purpose, so any // policy would collapse two different requested date ranges onto the shared diff --git a/src/lib/jobs/lab-document-date-repair-queue.ts b/src/lib/jobs/lab-document-date-repair-queue.ts new file mode 100644 index 000000000..34ccea6e6 --- /dev/null +++ b/src/lib/jobs/lab-document-date-repair-queue.ts @@ -0,0 +1,10 @@ +/** + * The queue facts of `lab-document-date-repair.ts`: its name and its one + * singleton key. They live apart from the job so the worker registry and the + * boot discovery can read them without loading it. + */ +export const LAB_DOCUMENT_DATE_REPAIR_QUEUE = "lab-document-date-repair"; + +/** One pass covers every account, so there is only ever one job. */ +export const LAB_DOCUMENT_DATE_REPAIR_SINGLETON_KEY = + "lab-document-date-repair"; diff --git a/src/lib/jobs/lab-document-date-repair.ts b/src/lib/jobs/lab-document-date-repair.ts new file mode 100644 index 000000000..57edb24e5 --- /dev/null +++ b/src/lib/jobs/lab-document-date-repair.ts @@ -0,0 +1,234 @@ +/** + * One-shot repair: lab values confirmed from a document before the confirm + * step learned to date an undated value by its document. + * + * A lab report usually prints the collection date once, in its header. The + * extraction stores that date on the document (`reportDate`) and leaves each + * value's own date empty, and the confirm step used to file such a value + * under the moment it was confirmed. The confirm step now takes the + * document's day (`documentDay` in `src/lib/documents/commit.ts`); this pass + * moves the values it filed wrongly before, by the same rule. + * + * A value is moved only when every link in the chain is certain: + * - the lab result was written by the document confirm path + * (`source = 'DOCUMENT'`, which no other writer sets) and is live; + * - exactly one APPROVED observation fact names it as its committed record, + * so the document it came from is known; + * - its `takenAt` still sits within a few seconds of its own `createdAt`, + * which is what a value filed under its confirmation moment looks like + * (both were stamped in the same create). A value the person re-dated + * since, or one that carried its own date, sits elsewhere; + * - the staged fact states no calendar date of its own (decrypted here; + * SQL cannot see it); + * - the document has a report date, else a filing date, and that day is + * not the day the value is filed under in the owner's time zone. A value + * really taken on the day it was confirmed stays where it is. + * + * The moved value lands at noon UTC on the document's day, the anchor every + * date-only lab value uses. Idempotent: a moved value no longer sits at its + * creation moment, so a second run finds nothing, and a boot with nothing to + * repair costs one query. + */ +import type { Job } from "pg-boss"; + +import { Prisma, type PrismaClient } from "@/generated/prisma/client"; +import { activeAccountSql } from "@/lib/account-deletion/active-account"; +import { invalidateUserHealthScore } from "@/lib/cache/invalidate"; +import { prisma as defaultPrisma } from "@/lib/db"; +import { documentDay, statedDay } from "@/lib/documents/commit"; +import { decryptFactData } from "@/lib/documents/store"; +import { getGlobalBoss } from "@/lib/jobs/boss-instance"; +import { jobDone, jobFailed, type JobOutcome } from "@/lib/jobs/job-outcome"; +import { workerLog } from "@/lib/jobs/reminder/shared"; +import { annotate } from "@/lib/logging/context"; +import { withBackgroundEvent } from "@/lib/logging/background"; +import { dateOnlyAtNoonUtc } from "@/lib/tz/date-only"; +import { userDayKey } from "@/lib/tz/format"; +import { resolveUserTimezone } from "@/lib/tz/resolver"; +import type { ObservationFactData } from "@/lib/validations/inbound-documents"; + +import { + LAB_DOCUMENT_DATE_REPAIR_QUEUE, + LAB_DOCUMENT_DATE_REPAIR_SINGLETON_KEY, +} from "./lab-document-date-repair-queue"; + +/** + * How far `takenAt` may sit from `createdAt` and still count as the moment of + * confirmation. The two are stamped a few milliseconds apart in one create; a + * few seconds leaves room for a clock step and nothing a person would type. + */ +const CONFIRM_MOMENT_TOLERANCE_SECONDS = 5; + +interface CandidateRow { + id: string; + userId: string; + takenAt: Date; + dataEncrypted: Uint8Array; + reportDate: Date | null; + documentDate: Date | null; +} + +/** + * The SQL half of the rule. `limit` turns it into the boot's existence + * check. `.planning/v1.46.2-fragments/lab-repair-count.sql` is the same + * predicate as a read-only count. + */ +function candidateQuery(limit: number | null): Prisma.Sql { + // The day comparison runs here as well as in the repair, so a value that + // really was taken on its confirmation day drops out of the boot check and + // the pass converges to an empty set. A zone Postgres does not know is read + // as UTC rather than failing the query; the repair re-checks every row in + // the zone the app resolves. + return Prisma.sql` + WITH "known_zones" AS (SELECT "name" FROM pg_timezone_names) + SELECT r."id", + r."user_id" AS "userId", + r."taken_at" AS "takenAt", + f."data_encrypted" AS "dataEncrypted", + d."report_date" AS "reportDate", + d."document_date" AS "documentDate" + FROM "lab_results" r + JOIN "extracted_facts" f + ON f."committed_record_id" = r."id" + AND f."committed_record_type" = 'labResult' + AND f."status" = 'APPROVED' + AND f."fact_type" = 'OBSERVATION' + AND f."user_id" = r."user_id" + JOIN "inbound_documents" d + ON d."id" = f."document_id" + AND d."user_id" = r."user_id" + JOIN "users" u ON u."id" = r."user_id" + WHERE r."source" = 'DOCUMENT' + AND r."deleted_at" IS NULL + AND abs(extract(epoch FROM (r."taken_at" - r."created_at"))) <= ${CONFIRM_MOMENT_TOLERANCE_SECONDS} + AND COALESCE(d."report_date", d."document_date") IS NOT NULL + AND to_char(COALESCE(d."report_date", d."document_date"), 'YYYY-MM-DD') + <> to_char( + (r."taken_at" AT TIME ZONE 'UTC') AT TIME ZONE ( + CASE WHEN u."timezone" IN (SELECT "name" FROM "known_zones") + THEN u."timezone" ELSE 'UTC' END + ), + 'YYYY-MM-DD' + ) + AND ( + SELECT count(*) FROM "extracted_facts" other + WHERE other."committed_record_id" = r."id" + AND other."committed_record_type" = 'labResult' + ) = 1 + AND ${activeAccountSql('r."user_id"')} + ORDER BY r."id" + ${limit === null ? Prisma.empty : Prisma.sql`LIMIT ${limit}`} + `; +} + +export interface LabDocumentDateRepairResult { + /** Rows the SQL predicate returned. */ + scanned: number; + /** Rows moved to their document's day. */ + updated: number; + /** Rows left alone: an own date, or already on the document's day. */ + unchanged: number; +} + +/** The repair itself, across every account. */ +export async function repairUndatedDocumentLabValues( + prisma: PrismaClient = defaultPrisma, +): Promise { + const rows = await prisma.$queryRaw(candidateQuery(null)); + const touchedUsers = new Set(); + let updated = 0; + let unchanged = 0; + + for (const row of rows) { + const data = decryptFactData(row.dataEncrypted) as ObservationFactData; + // A value with a date of its own is never touched, whatever its row says. + if (statedDay(data.effectiveDate ?? null) !== null) { + unchanged += 1; + continue; + } + const day = documentDay({ + reportDate: row.reportDate, + documentDate: row.documentDate, + }); + const tz = await resolveUserTimezone(row.userId); + if (!day || userDayKey(row.takenAt, tz) === day) { + unchanged += 1; + continue; + } + // Conditional on the value read: a re-date that landed since the read + // wins over the repair. + const moved = await prisma.labResult.updateMany({ + where: { + id: row.id, + userId: row.userId, + takenAt: row.takenAt, + deletedAt: null, + }, + data: { takenAt: dateOnlyAtNoonUtc(day) }, + }); + if (moved.count === 0) { + unchanged += 1; + continue; + } + updated += 1; + touchedUsers.add(row.userId); + } + + for (const userId of touchedUsers) invalidateUserHealthScore(userId); + return { scanned: rows.length, updated, unchanged }; +} + +/** The queue handler: one run over every account. */ +export async function handleLabDocumentDateRepair( + _jobs: Job>[], + prisma: PrismaClient = defaultPrisma, +): Promise { + return withBackgroundEvent("job.lab_document_date_repair", async () => { + let result: LabDocumentDateRepairResult; + try { + result = await repairUndatedDocumentLabValues(prisma); + } catch (err) { + workerLog("error", "[lab-document-date-repair] failed", err); + return jobFailed("lab document date repair failed", err); + } + workerLog( + "info", + `[lab-document-date-repair] scanned=${result.scanned} updated=${result.updated} unchanged=${result.unchanged}`, + ); + annotate({ + action: { name: "labs.documentDate.repair" }, + meta: { ...result }, + }); + return jobDone({ + candidates_scanned: result.scanned, + updated: result.updated, + unchanged: result.unchanged, + }); + }); +} + +/** + * Boot discovery: one existence query, and the one job only when there is + * something to repair. + */ +export async function enqueueBootTimeLabDocumentDateRepair( + prisma: PrismaClient = defaultPrisma, +): Promise<{ enqueued: boolean; error: string | null }> { + const boss = getGlobalBoss(); + if (!boss) return { enqueued: false, error: null }; + try { + const any = await prisma.$queryRaw(candidateQuery(1)); + if (any.length === 0) return { enqueued: false, error: null }; + const id = await boss.send( + LAB_DOCUMENT_DATE_REPAIR_QUEUE, + {}, + { singletonKey: LAB_DOCUMENT_DATE_REPAIR_SINGLETON_KEY, retryLimit: 2 }, + ); + return { enqueued: id !== null, error: null }; + } catch (err) { + return { + enqueued: false, + error: err instanceof Error ? err.message : String(err), + }; + } +} diff --git a/src/lib/jobs/queue-runtime.ts b/src/lib/jobs/queue-runtime.ts index 49fc6d154..7d6a3ae9d 100644 --- a/src/lib/jobs/queue-runtime.ts +++ b/src/lib/jobs/queue-runtime.ts @@ -485,6 +485,9 @@ export const QUEUE_RUNTIME: Readonly> = { }, why: "One account's folded means, a day at a time; stops at three quarters of the default expiry and sends a follow-up that resumes at the next day.", }, + "lab-document-date-repair": short( + "Reads the document lab values still filed at their confirmation moment, a handful per account, and moves each with one update.", + ), "measurement-maintenance": { runtime: "long", expireInSeconds: MEASUREMENT_MAINTENANCE_SEND_OPTIONS.expireInSeconds, diff --git a/src/lib/jobs/registry/maintenance-discovery.ts b/src/lib/jobs/registry/maintenance-discovery.ts index c0f2c41e2..2de44262a 100644 --- a/src/lib/jobs/registry/maintenance-discovery.ts +++ b/src/lib/jobs/registry/maintenance-discovery.ts @@ -77,6 +77,15 @@ export const MAINTENANCE_DISCOVERY: readonly BootDiscovery[] = [ await import("@/lib/jobs/measurement-fold-repair") ).enqueueBootTimeMeasurementFoldRepair(), }, + // Converges to nothing once the misdated values are moved; a boot with + // none costs one query. + { + name: "lab-document-date-repair", + run: async () => + ( + await import("@/lib/jobs/lab-document-date-repair") + ).enqueueBootTimeLabDocumentDateRepair(), + }, // Self-limiting and idempotent; a boot with no backlog costs one query. { name: "compaction-tombstone-purge", diff --git a/src/lib/jobs/registry/maintenance-repairs.ts b/src/lib/jobs/registry/maintenance-repairs.ts new file mode 100644 index 000000000..2d3224e86 --- /dev/null +++ b/src/lib/jobs/registry/maintenance-repairs.ts @@ -0,0 +1,25 @@ +/** + * One-shot data repairs of the maintenance group, kept beside + * `maintenance.ts` so that file stays under the size ratchet. Each is sent + * by its boot discovery in `maintenance-discovery.ts` only while it has + * something to repair. + */ +import { LAB_DOCUMENT_DATE_REPAIR_QUEUE } from "@/lib/jobs/lab-document-date-repair-queue"; + +import { defineJob, type JobDefinition } from "./define-job"; + +export const MAINTENANCE_REPAIR_JOBS: readonly JobDefinition[] = [ + // Document lab values filed under their confirmation moment. + defineJob({ + queue: LAB_DOCUMENT_DATE_REPAIR_QUEUE, + policy: { + policy: "short", + reason: + "Fixed singleton key; collapse queued duplicates only, so a boot during a run can queue the next one.", + }, + work: { localConcurrency: 1 }, + load: async () => + (await import("@/lib/jobs/lab-document-date-repair")) + .handleLabDocumentDateRepair, + }), +]; diff --git a/src/lib/jobs/registry/maintenance.ts b/src/lib/jobs/registry/maintenance.ts index 587123022..574d48352 100644 --- a/src/lib/jobs/registry/maintenance.ts +++ b/src/lib/jobs/registry/maintenance.ts @@ -202,6 +202,7 @@ import { } from "@/lib/medications/intake-slot-dedup"; import { MAINTENANCE_DISCOVERY } from "./maintenance-discovery"; +import { MAINTENANCE_REPAIR_JOBS } from "./maintenance-repairs"; import { defineJob, type JobDefinition, @@ -524,6 +525,7 @@ const jobs: readonly JobDefinition[] = [ (await import("@/lib/jobs/measurement-fold-repair")) .handleMeasurementFoldRepair, }), + ...MAINTENANCE_REPAIR_JOBS, // The nightly check vacuums when dead rows pile up; the same singleton key // as the operator's run, so the two collapse. defineJob({ diff --git a/src/lib/jobs/subsystem-surface.ts b/src/lib/jobs/subsystem-surface.ts index ea14eac89..419541cca 100644 --- a/src/lib/jobs/subsystem-surface.ts +++ b/src/lib/jobs/subsystem-surface.ts @@ -120,6 +120,7 @@ const subsystemSurface = { "health-connect-import": { audience: "account" }, "compaction-tombstone-purge": { audience: "system" }, "measurement-fold-repair": { audience: "system" }, + "lab-document-date-repair": { audience: "system" }, "measurement-maintenance": { audience: "system" }, "pulse-day-mean": { audience: "system" }, "medication-intake-import": { audience: "system" }, diff --git a/tests/integration/lab-document-date-repair.test.ts b/tests/integration/lab-document-date-repair.test.ts new file mode 100644 index 000000000..82c6aa086 --- /dev/null +++ b/tests/integration/lab-document-date-repair.test.ts @@ -0,0 +1,232 @@ +/** + * The one-shot repair of lab values a document confirm filed under the + * moment of confirmation (real Postgres). + * + * Before the confirm step dated an undated value by its document, such a + * value got `takenAt = now`, the same instant as its `createdAt`. The repair + * moves exactly those to the document's report date (else its filing date) + * at noon UTC, and leaves alone a value with a date of its own, a value that + * really was taken on its confirmation day in the owner's zone, one the + * person re-dated since, and one whose document has no date. A second run + * changes nothing. + */ +import { beforeEach, describe, expect, it } from "vitest"; + +import { encryptFactData, encryptFactProvenance } from "@/lib/documents/store"; +import { repairUndatedDocumentLabValues } from "@/lib/jobs/lab-document-date-repair"; +import { getPrismaClient, truncateAllTables } from "./setup"; + +const USER = "user-lab-document-date-repair"; +const noon = (day: string) => new Date(`${day}T12:00:00.000Z`); + +beforeEach(async () => { + const prisma = getPrismaClient(); + await truncateAllTables(prisma); + await prisma.user.create({ + data: { + id: USER, + username: "lab-date-repair", + email: "lab-date-repair@example.test", + timezone: "Europe/Berlin", + }, + }); +}); + +/** + * A document, one confirmed value from it and the fact that links the two, + * as the confirm route left them. + */ +async function confirmedValue(opts: { + id: string; + reportDate: string | null; + documentDate: string | null; + effectiveDate: string | null; + createdAt: Date; + takenAt: Date; +}) { + const prisma = getPrismaClient(); + const doc = await prisma.inboundDocument.create({ + data: { + userId: USER, + filename: `${opts.id}.pdf`, + mimeType: "application/pdf", + byteSize: 1024, + contentEncrypted: Buffer.from("stand-in for the stored bytes"), + status: "CONFIRMED", + reportDate: opts.reportDate ? noon(opts.reportDate) : null, + documentDate: opts.documentDate ? noon(opts.documentDate) : null, + }, + }); + await prisma.labResult.create({ + data: { + id: opts.id, + userId: USER, + analyte: "Ferritin", + value: 80, + unit: "ng/mL", + source: "DOCUMENT", + takenAt: opts.takenAt, + createdAt: opts.createdAt, + }, + }); + await prisma.extractedFact.create({ + data: { + documentId: doc.id, + userId: USER, + factType: "OBSERVATION", + status: "APPROVED", + confidence: 0.95, + needsReview: false, + dataEncrypted: encryptFactData({ + label: "Ferritin", + code: null, + codeSystem: null, + value: 80, + valueText: null, + unit: "ng/mL", + referenceLow: null, + referenceHigh: null, + referenceText: null, + effectiveDate: opts.effectiveDate, + }), + provenanceEncrypted: encryptFactProvenance({ + sourceText: "Ferritin 80 ng/mL", + anchored: true, + sourceOffset: 0, + page: 0, + confidence: 0.95, + }), + committedRecordId: opts.id, + committedRecordType: "labResult", + }, + }); +} + +async function takenAt(): Promise> { + const rows = await getPrismaClient().labResult.findMany({ + where: { userId: USER }, + select: { id: true, takenAt: true }, + }); + return Object.fromEntries(rows.map((r) => [r.id, r.takenAt.toISOString()])); +} + +describe("lab document date repair", () => { + it("moves only values filed under their confirmation moment, and converges", async () => { + const confirmed = new Date("2026-10-20T08:00:00.123Z"); + + // Collection date in the header, value undated, filed at confirmation. + await confirmedValue({ + id: "misdated", + reportDate: "2021-03-01", + documentDate: "2026-10-18", + effectiveDate: null, + createdAt: confirmed, + takenAt: confirmed, + }); + // No report date: the filing date is the document's day. + await confirmedValue({ + id: "misdated-filing-date", + reportDate: null, + documentDate: "2021-02-15", + effectiveDate: null, + createdAt: confirmed, + takenAt: new Date(confirmed.getTime() + 2), + }); + // Its own date, even where the row looks like a confirmation moment. + await confirmedValue({ + id: "own-date", + reportDate: "2021-03-01", + documentDate: null, + effectiveDate: "2021-02-20", + createdAt: confirmed, + takenAt: confirmed, + }); + // Confirmed at 00:30 in Berlin on the report day itself: 22:30 UTC the + // evening before, which a UTC comparison would have moved. + const justAfterMidnight = new Date("2026-10-19T22:30:00.000Z"); + await confirmedValue({ + id: "taken-on-confirm-day", + reportDate: "2026-10-20", + documentDate: "2026-10-20", + effectiveDate: null, + createdAt: justAfterMidnight, + takenAt: justAfterMidnight, + }); + // Re-dated by the person after the confirm. + await confirmedValue({ + id: "re-dated", + reportDate: "2021-03-01", + documentDate: null, + effectiveDate: null, + createdAt: confirmed, + takenAt: new Date("2026-10-01T07:45:00.000Z"), + }); + // A document with no date at all leaves nothing to move to. + await confirmedValue({ + id: "undated-document", + reportDate: null, + documentDate: null, + effectiveDate: null, + createdAt: confirmed, + takenAt: confirmed, + }); + + const before = await takenAt(); + const first = await repairUndatedDocumentLabValues(getPrismaClient()); + + expect(first.updated).toBe(2); + expect(await takenAt()).toEqual({ + ...before, + misdated: "2021-03-01T12:00:00.000Z", + "misdated-filing-date": "2021-02-15T12:00:00.000Z", + }); + + const afterFirst = await takenAt(); + const second = await repairUndatedDocumentLabValues(getPrismaClient()); + expect(second.updated).toBe(0); + expect(await takenAt()).toEqual(afterFirst); + }); + + it("leaves a lab value alone when a second fact also names it", async () => { + const confirmed = new Date("2026-10-20T08:00:00.000Z"); + await confirmedValue({ + id: "linked-twice", + reportDate: "2021-03-01", + documentDate: null, + effectiveDate: null, + createdAt: confirmed, + takenAt: confirmed, + }); + const prisma = getPrismaClient(); + const second = await prisma.inboundDocument.create({ + data: { + userId: USER, + filename: "other.pdf", + mimeType: "application/pdf", + byteSize: 1, + contentEncrypted: Buffer.from("x"), + reportDate: noon("2020-01-01"), + }, + }); + const fact = await prisma.extractedFact.findFirstOrThrow({ + where: { committedRecordId: "linked-twice" }, + }); + await prisma.extractedFact.create({ + data: { + documentId: second.id, + userId: USER, + factType: "OBSERVATION", + status: "APPROVED", + dataEncrypted: fact.dataEncrypted, + provenanceEncrypted: fact.provenanceEncrypted, + committedRecordId: "linked-twice", + committedRecordType: "labResult", + }, + }); + + const result = await repairUndatedDocumentLabValues(prisma); + + expect(result.updated).toBe(0); + expect((await takenAt())["linked-twice"]).toBe(confirmed.toISOString()); + }); +});