diff --git a/src/lib/medications/compliance-rollups.ts b/src/lib/medications/compliance-rollups.ts new file mode 100644 index 000000000..bb72b9604 --- /dev/null +++ b/src/lib/medications/compliance-rollups.ts @@ -0,0 +1,551 @@ +/** + * v1.4.39 W-MED — persistent medication-compliance rollup tier. + * + * Replaces the unbounded `MedicationIntakeEvent.findMany` walk on + * `/api/medications/intake?scope=compliance` with a ledger keyed + * `(userId, medicationId, day)` where `day` is `YYYY-MM-DD` anchored + * to the user's IANA timezone. + * + * Write hooks fire from every `MedicationIntakeEvent` mutation site + * (create / update / delete / bulk import) plus the reminder-worker's + * mint path that creates `takenAt:null` rows on RED-phase reminders. + * Reads collapse the trailing N-day window into a single bounded + * `findMany` against `medication_compliance_rollups`; the route-level + * coverage probe falls back to live + fires a boot backfill when a + * user has intake events but zero rollup rows. + * + * Audit anchor: `.planning/round-v1438-perf-analysis.md` §2.5 + §5 P4. + */ +import type { Prisma } from "@/generated/prisma/client"; + +import { prisma } from "@/lib/db"; +import { getGlobalBoss } from "@/lib/jobs/boss-instance"; +import { annotate } from "@/lib/logging/context"; +import { + DEFAULT_TIMEZONE, + isValidTimezone, + userDayKey, +} from "@/lib/tz/format"; + +/** + * pg-boss queue name for the boot-time medication compliance backfill. + * Mirrors the `rollup-full-backfill` v1.4.35.1 pattern: discovery is + * idempotent across reboots and the worker upserts every + * `(user, medication, day)` row for the trailing window. + */ +export const MEDICATION_COMPLIANCE_BACKFILL_QUEUE = + "medication-compliance-full-backfill"; + +/** Worker concurrency cap — backfill stays serial to spare the pool. */ +export const MEDICATION_COMPLIANCE_BACKFILL_CONCURRENCY = 1; + +/** Trailing window the boot backfill folds per uncovered user. */ +export const MEDICATION_COMPLIANCE_BACKFILL_DAYS = 90; + +/** Payload `boss.send` carries onto the backfill queue. */ +export interface MedicationComplianceBackfillPayload { + userId: string; + /** Wall-clock kick-off for debugging only. */ + enqueuedAt: string; +} + +/** Reader-result row shape — matches `buildComplianceBuckets` output. */ +export interface ComplianceBucket { + /** YYYY-MM-DD in the user's tz. */ + date: string; + scheduled: number; + taken: number; +} + +/** + * Prisma client surface the helpers accept. Either the regular + * `prisma` singleton or a `$transaction(tx)` argument — both expose + * the `medicationIntakeEvent.findMany` + `medicationComplianceRollup.*` + * methods we touch. + */ +type PrismaLike = typeof prisma | Prisma.TransactionClient; + +/** + * Compute the wall-clock components of `date` interpreted in `tz`. + * Same shape as the cadence helper's `wallClockInTz`, lifted here so + * the compliance module doesn't depend on a sibling-folder helper. + */ +function wallClockInTz( + date: Date, + tz: string, +): { year: number; month: number; day: number } { + const parts = new Intl.DateTimeFormat("en-US", { + timeZone: tz, + year: "numeric", + month: "2-digit", + day: "2-digit", + }).formatToParts(date); + const get = (type: Intl.DateTimeFormatPartTypes): string => + parts.find((p) => p.type === type)?.value ?? "0"; + return { + year: Number(get("year")), + month: Number(get("month")), + day: Number(get("day")), + }; +} + +/** + * UTC offset (in minutes) of `tz` at the given instant. Positive east + * of UTC. Honours DST via `Intl.DateTimeFormat`. + */ +function tzOffsetMinutes(date: Date, tz: string): number { + const parts = new Intl.DateTimeFormat("en-US", { + timeZone: tz, + year: "numeric", + month: "2-digit", + day: "2-digit", + hour: "2-digit", + minute: "2-digit", + second: "2-digit", + hour12: false, + }).formatToParts(date); + const get = (type: Intl.DateTimeFormatPartTypes): string => + parts.find((p) => p.type === type)?.value ?? "0"; + let hour = Number(get("hour")); + if (hour === 24) hour = 0; + const asIfUtc = Date.UTC( + Number(get("year")), + Number(get("month")) - 1, + Number(get("day")), + hour, + Number(get("minute")), + Number(get("second")), + ); + return Math.round((asIfUtc - date.getTime()) / 60000); +} + +/** + * The UTC instant corresponding to local-midnight on `dayKey` in `tz`. + * Two-pass convergence handles DST transitions correctly. + */ +function startOfDayUtcInTz(dayKey: string, tz: string): Date { + const [yearStr, monthStr, dayStr] = dayKey.split("-"); + const year = Number(yearStr); + const month = Number(monthStr); + const day = Number(dayStr); + let guess = new Date(Date.UTC(year, month - 1, day, 0, 0, 0, 0)); + for (let i = 0; i < 2; i++) { + const offsetMin = tzOffsetMinutes(guess, tz); + guess = new Date( + Date.UTC(year, month - 1, day, 0, 0, 0, 0) - offsetMin * 60_000, + ); + } + return guess; +} + +/** + * Resolve a safe tz string — fall back to the server default when the + * supplied value is missing or not an IANA zone the host recognises. + */ +function safeTimezone(tz: string | null | undefined): string { + if (tz && isValidTimezone(tz)) return tz; + return DEFAULT_TIMEZONE; +} + +/** + * Convert a `scheduledFor` UTC instant to the user's `YYYY-MM-DD` + * day-key. Re-exposed so the write-hook caller can derive the same + * `dayKey` the rollup row is anchored on. + */ +export function dayKeyForScheduledFor( + scheduledFor: Date, + tz: string | null | undefined, +): string { + return userDayKey(scheduledFor, safeTimezone(tz)); +} + +/** + * Recompute the rollup row for one `(userId, medicationId, dayKey)` + * tuple. Reads every `MedicationIntakeEvent` whose `scheduledFor` lands + * in `[startOfDayInTz, startOfNextDayInTz)` then upserts the + * `medication_compliance_rollups` row. + * + * The hook is idempotent: re-running for the same key produces the + * same row. When the day window holds zero events the row is deleted + * so the read path returns the trailing-window zero-default rather + * than a stale stub. + * + * `tx` accepts the `prisma.$transaction(tx)` client argument when the + * parent write is transactional — passing the transaction client keeps + * the rollup write inside the same transactional boundary. + */ +export async function recomputeMedicationComplianceForDay( + userId: string, + medicationId: string, + dayKey: string, + tz: string | null | undefined, + tx?: PrismaLike, +): Promise { + const client = (tx ?? prisma) as PrismaLike; + const safeTz = safeTimezone(tz); + const start = startOfDayUtcInTz(dayKey, safeTz); + const end = new Date(start.getTime() + 86_400_000); + // Re-derive end via the day-after boundary to honour DST 23 / 25-hour + // days. `start + 86_400_000` is correct for non-transition days; on + // a fall-back day we add one calendar day in the local zone. + const nextDayKey = (() => { + const [y, m, d] = dayKey.split("-").map(Number); + const probe = new Date(Date.UTC(y, m - 1, d, 12, 0, 0)); + const probeNext = new Date(probe.getTime() + 86_400_000); + const parts = wallClockInTz(probeNext, safeTz); + return `${String(parts.year).padStart(4, "0")}-${String(parts.month).padStart(2, "0")}-${String(parts.day).padStart(2, "0")}`; + })(); + const dstSafeEnd = startOfDayUtcInTz(nextDayKey, safeTz); + const windowEnd = dstSafeEnd.getTime() > start.getTime() ? dstSafeEnd : end; + + const events = await client.medicationIntakeEvent.findMany({ + where: { + userId, + medicationId, + scheduledFor: { gte: start, lt: windowEnd }, + }, + select: { takenAt: true, skipped: true }, + }); + + let scheduled = 0; + let taken = 0; + let skipped = 0; + for (const evt of events) { + scheduled += 1; + if (evt.skipped) { + skipped += 1; + } else if (evt.takenAt) { + taken += 1; + } + } + + if (scheduled === 0) { + // No events landed inside the user-day window after the mutation — + // the row would otherwise stay around with a stale (1, 1, 0) tuple + // from a now-deleted intake event. Delete the rollup row so the + // reader returns the trailing-window zero-default for this day. + await client.medicationComplianceRollup.deleteMany({ + where: { userId, medicationId, day: dayKey }, + }); + return; + } + + await client.medicationComplianceRollup.upsert({ + where: { + userId_medicationId_day: { userId, medicationId, day: dayKey }, + }, + update: { + scheduled, + taken, + skipped, + computedAt: new Date(), + }, + create: { + userId, + medicationId, + day: dayKey, + scheduled, + taken, + skipped, + }, + }); +} + +/** + * Convenience hook for write-paths that have the `scheduledFor` + * instant in hand. Derives `dayKey` in the user's tz then dispatches + * to `recomputeMedicationComplianceForDay`. Best-effort: surfaces + * failures through `annotate` + `console.error` so a silent populator + * regression shows up in ops without blocking the parent write. + */ +export async function recomputeMedicationComplianceForEvent(input: { + userId: string; + medicationId: string; + scheduledFor: Date; + tz: string | null | undefined; + tx?: PrismaLike; +}): Promise { + const dayKey = dayKeyForScheduledFor(input.scheduledFor, input.tz); + try { + await recomputeMedicationComplianceForDay( + input.userId, + input.medicationId, + dayKey, + input.tz, + input.tx, + ); + } catch (err) { + const message = err instanceof Error ? err.message : String(err); + annotate({ + meta: { + medication_compliance_rollup_failed: true, + medication_compliance_rollup_error: message, + }, + }); + console.error( + "[medication-compliance-rollups] recompute failed:", + message, + ); + } +} + +/** + * Read trailing `days` of compliance buckets for the user, oldest day + * first. Returns the same `{date, scheduled, taken}` shape the legacy + * `buildComplianceBuckets` helper produced so the route swap is + * byte-identical. + * + * Days without a rollup row land as zero-filled buckets. The caller's + * cache wrapper carries the user-tz in its key so a tz change forces + * a fresh read. + */ +export async function readMedicationCompliance( + userId: string, + days: number, + tz: string | null | undefined, + now: Date = new Date(), +): Promise { + const safeTz = safeTimezone(tz); + + // Build the expected day-key window oldest → newest so the response + // is sorted and zero-filled regardless of which days hold rollups. + const todayKey = userDayKey(now, safeTz); + const expectedKeys: string[] = []; + for (let i = days - 1; i >= 0; i--) { + const probe = new Date(now.getTime() - i * 86_400_000); + expectedKeys.push(userDayKey(probe, safeTz)); + } + // Guard against a DST day-of-the-year landing the loop on the same + // dayKey twice — dedup by Map insertion order. + const dedup = new Set(); + const orderedKeys: string[] = []; + for (const key of expectedKeys) { + if (!dedup.has(key)) { + dedup.add(key); + orderedKeys.push(key); + } + } + // Always include today even if the trailing window dropped it (e.g. + // days===1 on a DST boundary). Cheap defensive insert. + if (!dedup.has(todayKey)) { + orderedKeys.push(todayKey); + } + + const oldestKey = orderedKeys[0]; + const newestKey = orderedKeys[orderedKeys.length - 1]; + + const rows = await prisma.medicationComplianceRollup.findMany({ + where: { + userId, + day: { gte: oldestKey, lte: newestKey }, + }, + select: { day: true, scheduled: true, taken: true, skipped: true }, + }); + + // Fold per-medication rows into one per-day total. The legacy route + // shape was aggregated across all the user's medications; preserving + // that surface keeps the front-end byte-stable. + const totals = new Map(); + for (const key of orderedKeys) { + totals.set(key, { scheduled: 0, taken: 0 }); + } + for (const row of rows) { + const bucket = totals.get(row.day); + if (!bucket) continue; + bucket.scheduled += row.scheduled; + bucket.taken += row.taken; + } + + return orderedKeys.map((date) => { + const bucket = totals.get(date) ?? { scheduled: 0, taken: 0 }; + return { date, scheduled: bucket.scheduled, taken: bucket.taken }; + }); +} + +/** + * Returns true when the user has at least one + * `medication_compliance_rollups` row inside the trailing `days` + * window. The route uses this as the coverage probe — when the + * trailing window is empty but the live `medication_intake_events` + * table has rows, the read path falls through to the legacy live + * aggregator + fires a boot backfill in the background. + */ +export async function hasMedicationComplianceCoverage( + userId: string, + days: number, + tz: string | null | undefined, + now: Date = new Date(), +): Promise { + const safeTz = safeTimezone(tz); + const oldestKey = userDayKey( + new Date(now.getTime() - (days - 1) * 86_400_000), + safeTz, + ); + const row = await prisma.medicationComplianceRollup.findFirst({ + where: { userId, day: { gte: oldestKey } }, + select: { day: true }, + }); + return row !== null; +} + +/** + * Full-fold helper. Walks every `(medicationId, day)` pair the user + * has logged across the trailing `days` window and upserts each + * rollup row. The boot-time backfill worker calls this once per + * uncovered user; the route's lazy fallback enqueues it when the + * coverage probe returns false. + * + * Bounded SQL: one `DISTINCT (medicationId, dayKey)` $queryRaw scan + * over the trailing window, then one upsert per row. The dayKey is + * computed in Postgres via `to_char(... AT TIME ZONE $tz)` so we + * don't have to ship millions of rows back to Node. + */ +export async function recomputeUserMedicationCompliance( + userId: string, + days: number = MEDICATION_COMPLIANCE_BACKFILL_DAYS, + tz: string | null | undefined = null, +): Promise<{ rowsUpserted: number; durationMs: number }> { + const startedAt = Date.now(); + const safeTz = safeTimezone(tz); + + const now = new Date(); + const oldestKey = userDayKey( + new Date(now.getTime() - (days - 1) * 86_400_000), + safeTz, + ); + const oldestStart = startOfDayUtcInTz(oldestKey, safeTz); + + // Distinct (medication_id, dayKey) pairs the user has any intake + // event for inside the trailing window. `to_char(... AT TIME ZONE + // $tz)` keeps the day-bucket on the user's wall-clock. + const pairs = await prisma.$queryRaw< + Array<{ medication_id: string; day: string }> + >` + SELECT DISTINCT + "medication_id", + to_char("scheduled_for" AT TIME ZONE ${safeTz}, 'YYYY-MM-DD') AS "day" + FROM "medication_intake_events" + WHERE "user_id" = ${userId} + AND "scheduled_for" >= ${oldestStart} + `; + + let rowsUpserted = 0; + for (const pair of pairs) { + await recomputeMedicationComplianceForDay( + userId, + pair.medication_id, + pair.day, + safeTz, + ); + rowsUpserted += 1; + } + + return { rowsUpserted, durationMs: Date.now() - startedAt }; +} + +/** + * Fire-and-forget warm-up — kept around for symmetry with + * `ensureUserRollupsFresh` even though the on-write hook fully owns + * coverage in steady state. The route currently doesn't call this + * directly; the coverage-probe + boot-backfill cascade handles the + * cold-mount case. + */ +export async function ensureUserMedicationComplianceFresh( + userId: string, + days: number = MEDICATION_COMPLIANCE_BACKFILL_DAYS, + tz: string | null | undefined = null, +): Promise<{ recomputed: boolean }> { + try { + if (await hasMedicationComplianceCoverage(userId, days, tz)) { + return { recomputed: false }; + } + await recomputeUserMedicationCompliance(userId, days, tz); + return { recomputed: true }; + } catch (err) { + const message = err instanceof Error ? err.message : String(err); + annotate({ + meta: { + medication_compliance_rollup_refresh_failed: true, + medication_compliance_rollup_refresh_error: message, + }, + }); + console.error( + "[medication-compliance-rollups] ensure-fresh failed:", + message, + ); + return { recomputed: false }; + } +} + +/** + * Boot-time enqueue helper — mirrors v1.4.35.1 + * `enqueueBootTimeRollupBackfill`. Finds users with intake events but + * zero compliance-rollup coverage for the trailing window and enqueues + * one full-fold per account onto `MEDICATION_COMPLIANCE_BACKFILL_QUEUE`. + * + * Idempotent across reboots: once an account has at least one rollup + * row inside the trailing window the discovery query drops them. + * pg-boss `singletonKey` coalesces duplicate sends. + */ +export async function enqueueBootTimeMedicationComplianceBackfill(): Promise<{ + enqueued: number; + skipped: number; + error: string | null; +}> { + const boss = getGlobalBoss(); + if (!boss) { + return { enqueued: 0, skipped: 0, error: null }; + } + + try { + // Trailing window cutoff for the existence probe — `scheduled_for` + // older than this is not what the read path consults. + const cutoff = new Date( + Date.now() - MEDICATION_COMPLIANCE_BACKFILL_DAYS * 86_400_000, + ); + const users = await prisma.$queryRaw>` + SELECT DISTINCT mie."user_id" AS id + FROM "medication_intake_events" mie + LEFT JOIN "medication_compliance_rollups" r + ON r."user_id" = mie."user_id" + AND r."medication_id" = mie."medication_id" + WHERE mie."scheduled_for" >= ${cutoff} + AND r."day" IS NULL + `; + + if (users.length === 0) { + return { enqueued: 0, skipped: 0, error: null }; + } + + let enqueued = 0; + let skipped = 0; + for (const { id } of users) { + const payload: MedicationComplianceBackfillPayload = { + userId: id, + enqueuedAt: new Date().toISOString(), + }; + const jobId = await boss.send( + MEDICATION_COMPLIANCE_BACKFILL_QUEUE, + payload, + { + retryLimit: 3, + retryDelay: 60, + retryBackoff: true, + singletonKey: `medication-compliance-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), + }; + } +}