From 9e6d0d5acbc6b9c1d28da4897e8ba16e300d0202 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Wed, 10 Jun 2026 11:18:00 +0200 Subject: [PATCH] fix(medications): converge intake writes onto existing slot rows across sources MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A take whose slot the legacy resolver rejected fell back to a standalone insert that kept the client's scheduledFor, parking a second live row on an anchor the reminder projector had already minted — the partial unique index permits that because it spans source. Both intake write routes now probe the slot source-agnostically and converge onto the existing row before inserting, and a standalone insert anchors on the take time per the ad-hoc contract. The compliance rollup counts distinct slots instead of rows, which also corrects historical duplicates on the next recompute, and the slot dedup pass runs nightly instead of only at worker boot. --- src/app/api/medications/[id]/intake/route.ts | 101 ++++++-- .../intake/bulk/__tests__/route.test.ts | 172 +++++++++++++ src/app/api/medications/intake/bulk/route.ts | 110 +++++++-- .../__tests__/intake-slot-dedup-queue.test.ts | 37 ++- src/lib/jobs/reminder-worker.ts | 35 ++- src/lib/medications/intake-slot-dedup.ts | 14 +- .../medication-compliance-rollups.test.ts | 29 +++ .../rollups/medication-compliance-rollups.ts | 37 ++- .../medication-compliance-rollup-slot.test.ts | 225 ++++++++++++++++++ 9 files changed, 703 insertions(+), 57 deletions(-) create mode 100644 tests/integration/medication-compliance-rollup-slot.test.ts diff --git a/src/app/api/medications/[id]/intake/route.ts b/src/app/api/medications/[id]/intake/route.ts index 1373fe569..538d1b66e 100644 --- a/src/app/api/medications/[id]/intake/route.ts +++ b/src/app/api/medications/[id]/intake/route.ts @@ -269,33 +269,82 @@ async function postIntake(request: NextRequest, { params }: RouteParams) { }); } } else { - // Unscheduled / PRN / off-slot — keep the original insert behaviour. - [event] = await prisma.$transaction([ - prisma.medicationIntakeEvent.create({ - data: { - userId: user.id, - medicationId: id, - scheduledFor: incomingScheduledFor, - takenAt: resolvedTakenAt, - skipped, - source: "WEB", - idempotencyKey: idempotencyKey ?? null, - // v1.8.5 — site only on a resolved taken-injection write. - ...(resolvedInjectionSite !== null && { - injectionSite: resolvedInjectionSite, - }), - }, - }), - // Reset snooze when medication is taken - ...(!skipped - ? [ - prisma.medication.update({ - where: { id }, - data: { snoozedUntil: null }, + // Unscheduled / PRN / off-slot. When the client named an explicit + // `scheduledFor`, converge source-agnostically onto any live row that + // already sits on that instant (e.g. the pending REMINDER row the + // worker minted on a slot the band attribution did not claim) before + // inserting. Without the probe the insert lands a second live row for + // the same slot that differs only by `source` — the partial unique + // index carries `source` and cannot catch it — inflating the + // compliance rollup's scheduled count. A defaulted anchor (takenAt / + // now) never names a slot, so the probe is skipped on that hot path. + const existingSlotRow = + scheduledFor !== undefined + ? await prisma.medicationIntakeEvent.findFirst({ + where: { + userId: user.id, + medicationId: id, + scheduledFor: incomingScheduledFor, + deletedAt: null, + }, + select: { id: true }, + }) + : null; + if (existingSlotRow) { + const applied = await applyCanonicalSlotWrite({ + client: prisma, + userId: user.id, + medicationId: id, + canonicalSlot: incomingScheduledFor, + takenAt: resolvedTakenAt, + skipped, + isExplicitTaken, + isExplicitSkip, + idempotencyKey: idempotencyKey ?? null, + createSource: "WEB", + injectionSite: resolvedInjectionSite, + }); + event = applied.row; + consumedTransition = applied.consumedTransition; + if (!skipped && !applied.noDowngradeNoOp) { + await prisma.medication.update({ + where: { id }, + data: { snoozedUntil: null }, + }); + } + } else { + // Genuinely standalone. Anchor a taken write on the intake instant — + // the documented ad-hoc contract (`scheduledFor = takenAt`) — so an + // unresolvable client anchor can never park a live row exactly on a + // slot instant a pending REMINDER row is minted for later. A skip + // without a slot keeps the incoming instant (it has no takenAt). + [event] = await prisma.$transaction([ + prisma.medicationIntakeEvent.create({ + data: { + userId: user.id, + medicationId: id, + scheduledFor: resolvedTakenAt ?? incomingScheduledFor, + takenAt: resolvedTakenAt, + skipped, + source: "WEB", + idempotencyKey: idempotencyKey ?? null, + // v1.8.5 — site only on a resolved taken-injection write. + ...(resolvedInjectionSite !== null && { + injectionSite: resolvedInjectionSite, }), - ] - : []), - ]); + }, + }), + // Reset snooze when medication is taken + ...(!skipped + ? [ + prisma.medication.update({ + where: { id }, + data: { snoozedUntil: null }, + }), + ] + : []), + ]); + } } // v1.4.25 W19b — pen-inventory dose decrement. Only fires for diff --git a/src/app/api/medications/intake/bulk/__tests__/route.test.ts b/src/app/api/medications/intake/bulk/__tests__/route.test.ts index 30c2a2dd4..69f8a9fda 100644 --- a/src/app/api/medications/intake/bulk/__tests__/route.test.ts +++ b/src/app/api/medications/intake/bulk/__tests__/route.test.ts @@ -18,6 +18,8 @@ vi.mock("@/lib/db", () => ({ // v1.8.2 reconcile — shared slot upsert reads + updates in place. findMany: vi.fn(), update: vi.fn(), + // v1.15.19 — resolver-null convergence probe before standalone insert. + findFirst: vi.fn(), }, auditLog: { create: vi.fn() }, }, @@ -328,3 +330,173 @@ describe("POST /api/medications/intake/bulk — v1.8.2 reconcile", () => { ); }); }); + +// ──────────────────────────────────────────────────────────────────── +// v1.15.19 — resolver-null convergence (cross-source duplicate slots) +// ──────────────────────────────────────────────────────────────────── + +describe("POST /api/medications/intake/bulk — v1.15.19 resolver-null convergence", () => { + const SCHEDULED_MED = { + id: "med-1", + startsOn: null, + endsOn: null, + oneShot: false, + createdAt: new Date("2026-01-01T00:00:00Z"), + schedules: [ + { + id: "s1", + windowStart: "07:00", + windowEnd: "07:00", + daysOfWeek: null, + timesOfDay: ["07:00", "19:00"], + reminderGraceMinutes: null, + rrule: null, + rollingIntervalDays: null, + scheduleType: "SCHEDULED", + cyclicOnWeeks: null, + cyclicOffWeeks: null, + doseWindows: null, + }, + ], + }; + + beforeEach(() => { + // Pin the clock HOURS before the fixture slot so the taken-write + // future-slot guard (TAKEN_FORWARD_GRACE_MS) rejects the snap and the + // resolver returns null — the production shape: the reminder worker has + // already pre-minted the pending REMINDER row at the slot instant, the + // client confirms early with an explicit `scheduledFor` on that instant. + vi.useFakeTimers({ toFake: ["Date"] }); + vi.setSystemTime(new Date("2026-06-15T00:30:00.000Z")); + vi.mocked(prisma.medication.findMany).mockResolvedValue([ + { id: "med-1" }, + ] as never); + vi.mocked(prisma.medication.findFirst).mockResolvedValue( + SCHEDULED_MED as never, + ); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + it("converges an early taken-write onto the pre-minted pending REMINDER row instead of inserting a sibling", async () => { + // Source-agnostic probe finds the live REMINDER row on the incoming + // instant… + vi.mocked(prisma.medicationIntakeEvent.findFirst).mockResolvedValueOnce({ + id: "row-reminder", + } as never); + // …and the shared slot upsert re-reads + updates it in place. + vi.mocked(prisma.medicationIntakeEvent.findMany).mockResolvedValueOnce([ + { + id: "row-reminder", + takenAt: null, + skipped: false, + idempotencyKey: null, + scheduledFor: new Date("2026-06-15T05:00:00Z"), + source: "REMINDER", + createdAt: new Date("2026-06-15T00:00:00Z"), + }, + ] as never); + vi.mocked(prisma.medicationIntakeEvent.update).mockResolvedValueOnce({ + id: "row-reminder", + } as never); + + const res = await POST( + postReq({ + entries: [ + { + medicationId: "med-1", + scheduledFor: "2026-06-15T05:00:00.000Z", + takenAt: "2026-06-15T00:30:00.000Z", + }, + ], + }), + ); + expect(res.status).toBe(200); + const body = (await res.json()) as { + data: { + inserted: number; + updated: number; + entries: Array<{ status: string; id?: string }>; + }; + }; + // The slot did NOT fork into a second live event. + expect(body.data.updated).toBe(1); + expect(body.data.inserted).toBe(0); + expect(body.data.entries[0].status).toBe("updated"); + expect(body.data.entries[0].id).toBe("row-reminder"); + expect(prisma.medicationIntakeEvent.create).not.toHaveBeenCalled(); + expect(prisma.medicationIntakeEvent.update).toHaveBeenCalledWith( + expect.objectContaining({ where: { id: "row-reminder" } }), + ); + }); + + it("anchors a genuinely standalone resolver-null take on takenAt, not on the client's slot anchor", async () => { + // No live row on the incoming instant — the write is ad-hoc and must + // anchor on the intake instant so a worker mint at the slot instant + // later cannot pair up with it as a cross-source duplicate. + vi.mocked(prisma.medicationIntakeEvent.findFirst).mockResolvedValueOnce( + null as never, + ); + vi.mocked(prisma.medicationIntakeEvent.create).mockResolvedValueOnce({ + id: "row-standalone", + } as never); + + const res = await POST( + postReq({ + entries: [ + { + medicationId: "med-1", + scheduledFor: "2026-06-15T05:00:00.000Z", + takenAt: "2026-06-15T00:30:00.000Z", + }, + ], + }), + ); + expect(res.status).toBe(200); + const body = (await res.json()) as { + data: { inserted: number; entries: Array<{ status: string }> }; + }; + expect(body.data.inserted).toBe(1); + expect(body.data.entries[0].status).toBe("inserted"); + expect(prisma.medicationIntakeEvent.create).toHaveBeenCalledWith( + expect.objectContaining({ + data: expect.objectContaining({ + scheduledFor: new Date("2026-06-15T00:30:00.000Z"), + takenAt: new Date("2026-06-15T00:30:00.000Z"), + }), + }), + ); + }); + + it("keeps the idempotencyKey replay contract: a re-submission reports duplicate, not updated", async () => { + // The replay pre-check fires BEFORE the convergence probe so a re-sent + // entry keeps its historical `duplicate` status. + vi.mocked(prisma.medicationIntakeEvent.findUnique).mockResolvedValueOnce({ + id: "row-existing", + } as never); + + const res = await POST( + postReq({ + entries: [ + { + medicationId: "med-1", + scheduledFor: "2026-06-15T05:00:00.000Z", + takenAt: "2026-06-15T00:30:00.000Z", + idempotencyKey: "ios-key-1", + }, + ], + }), + ); + expect(res.status).toBe(200); + const body = (await res.json()) as { + data: { duplicates: number; entries: Array<{ status: string; id?: string }> }; + }; + expect(body.data.duplicates).toBe(1); + expect(body.data.entries[0].status).toBe("duplicate"); + expect(body.data.entries[0].id).toBe("row-existing"); + expect(prisma.medicationIntakeEvent.create).not.toHaveBeenCalled(); + expect(prisma.medicationIntakeEvent.findFirst).not.toHaveBeenCalled(); + }); +}); diff --git a/src/app/api/medications/intake/bulk/route.ts b/src/app/api/medications/intake/bulk/route.ts index 7eb9ca0e9..08d2732b0 100644 --- a/src/app/api/medications/intake/bulk/route.ts +++ b/src/app/api/medications/intake/bulk/route.ts @@ -244,7 +244,10 @@ async function postBulk(request: NextRequest): Promise { isTakenWrite: !entry.skipped && entry.takenAt !== undefined, }); - const scheduledFor = canonicalSlot ?? incomingScheduledFor; + // The `scheduledFor` instant the row actually lands on — set by the + // branch that performs the write so the rollup recompute below keys + // the same day the stored row anchors to. + let effectiveScheduledFor = canonicalSlot ?? incomingScheduledFor; // C2 — classify the incoming write. An offline-sync replay echoes a // PENDING projection (no `takenAt`, `skipped:false`) for a slot the @@ -312,28 +315,101 @@ async function postBulk(request: NextRequest): Promise { results.push({ index: i, status: "inserted", id: applied.row.id }); } } else { - // Unscheduled / PRN — insert. The idempotencyKey, when supplied, - // has a UNIQUE index; a re-submission returns the existing row via - // P2002 → status: "duplicate". - const row = await prisma.medicationIntakeEvent.create({ - data: { + // Resolver-null: PRN / off-slot / future-slot-guarded. Before the + // standalone insert, two guards keep the slot from forking into a + // second live row: + // + // 1. idempotencyKey replay pre-check — a re-submission returns the + // existing row as "duplicate" (previously reached via the P2002 + // catch; the explicit probe keeps that contract now that a + // replay could otherwise converge as "updated" below). + // 2. source-agnostic convergence probe — when ANY live row already + // sits on the incoming instant (e.g. the pending REMINDER row + // the worker pre-minted on a slot the taken-write future guard + // refused to snap to), converge onto THAT row through the shared + // slot upsert instead of inserting a sibling that differs only + // by `source`. The partial unique index carries `source`, so it + // cannot catch this duplicate; the probe must. + if (entry.idempotencyKey) { + const replay = await prisma.medicationIntakeEvent.findUnique({ + where: { idempotencyKey: entry.idempotencyKey }, + select: { id: true }, + }); + if (replay) { + duplicates += 1; + results.push({ index: i, status: "duplicate", id: replay.id }); + continue; + } + } + // Only an explicit client `scheduledFor` names a slot a row could + // already sit on; a defaulted anchor (takenAt / now) never does, so + // the probe is skipped on that path. + const existingSlotRow = + entry.scheduledFor !== undefined + ? await prisma.medicationIntakeEvent.findFirst({ + where: { + userId: user.id, + medicationId: entry.medicationId, + scheduledFor: incomingScheduledFor, + deletedAt: null, + }, + select: { id: true }, + }) + : null; + if (existingSlotRow) { + const applied = await applyCanonicalSlotWrite({ + client: prisma, userId: user.id, medicationId: entry.medicationId, - scheduledFor, + canonicalSlot: incomingScheduledFor, takenAt: entry.takenAt ?? null, skipped: entry.skipped, - source: "API", + isExplicitTaken, + isExplicitSkip, idempotencyKey: entry.idempotencyKey ?? null, - // v1.8.5 — site only on a resolved taken-injection entry. - ...(resolvedInjectionSite !== null && { - injectionSite: resolvedInjectionSite, - }), - }, - }); - inserted += 1; - results.push({ index: i, status: "inserted", id: row.id }); + createSource: "API", + injectionSite: resolvedInjectionSite, + }); + if (applied.noDowngradeNoOp) { + duplicates += 1; + results.push({ index: i, status: "duplicate", id: applied.row.id }); + } else if (applied.outcome === "updated") { + updated += 1; + results.push({ index: i, status: "updated", id: applied.row.id }); + } else { + inserted += 1; + results.push({ index: i, status: "inserted", id: applied.row.id }); + } + } else { + // Genuinely standalone. Anchor the row on the intake instant — + // the documented ad-hoc contract (`scheduledFor = takenAt`) — so + // an unresolvable client anchor (a future slot instant the taken + // guard rejected) can never park a live row exactly on a slot the + // worker later mints a pending REMINDER row for. A pending echo + // (no takenAt) keeps the incoming instant. The idempotencyKey, + // when supplied, has a UNIQUE index; a racing re-submission + // returns the existing row via P2002 → status: "duplicate". + effectiveScheduledFor = entry.takenAt ?? incomingScheduledFor; + const row = await prisma.medicationIntakeEvent.create({ + data: { + userId: user.id, + medicationId: entry.medicationId, + scheduledFor: effectiveScheduledFor, + takenAt: entry.takenAt ?? null, + skipped: entry.skipped, + source: "API", + idempotencyKey: entry.idempotencyKey ?? null, + // v1.8.5 — site only on a resolved taken-injection entry. + ...(resolvedInjectionSite !== null && { + injectionSite: resolvedInjectionSite, + }), + }, + }); + inserted += 1; + results.push({ index: i, status: "inserted", id: row.id }); + } } - const dayKey = dayKeyForScheduledFor(scheduledFor, user.timezone); + const dayKey = dayKeyForScheduledFor(effectiveScheduledFor, user.timezone); touchedDays.set(`${entry.medicationId}|${dayKey}`, { medicationId: entry.medicationId, dayKey, diff --git a/src/lib/jobs/__tests__/intake-slot-dedup-queue.test.ts b/src/lib/jobs/__tests__/intake-slot-dedup-queue.test.ts index 68007b647..3a2d99c0c 100644 --- a/src/lib/jobs/__tests__/intake-slot-dedup-queue.test.ts +++ b/src/lib/jobs/__tests__/intake-slot-dedup-queue.test.ts @@ -35,12 +35,47 @@ describe("reminder-worker — intake-slot-dedup wiring", () => { }); it("registers a boss.work handler against the intake-slot-dedup queue", () => { + // v1.15.19 — span widened from {0,400}: the handler gained the daily + // cron-tick dispatch branch (no `userId` → discovery fan-out) between + // the queue name and the per-user dedup call. expect(source).toMatch( - /boss\.work[\s\S]{0,200}INTAKE_SLOT_DEDUP_QUEUE[\s\S]{0,400}dedupeUserIntakeSlots/, + /boss\.work[\s\S]{0,200}INTAKE_SLOT_DEDUP_QUEUE[\s\S]{0,1200}dedupeUserIntakeSlots/, ); }); it("fires the boot-discovery enqueue helper", () => { expect(source).toMatch(/await enqueueBootTimeIntakeSlotDedup\(\)/); }); + + // v1.15.19 — the boot-only pass left duplicates created between deploys + // standing until the next worker restart. The queue now also carries a + // daily cron tick whose empty payload (no `userId`) dispatches the same + // discovery fan-out. + it("declares a daily cron inside the 03:xx maintenance window", () => { + const cronMatch = source.match( + /const INTAKE_SLOT_DEDUP_CRON\s*=\s*["']([^"']+)["']/, + ); + expect(cronMatch).not.toBeNull(); + // Five-field cron, daily (day-of-month/month/day-of-week all `*`), + // inside the 03:xx maintenance window. + expect(cronMatch![1]).toMatch(/^\d{1,2} 3 \* \* \*$/); + }); + + it("registers the cron in the schedules table", () => { + const schedulesMatch = source.match( + /const schedules:\s*\[string, string\]\[\]\s*=\s*\[([\s\S]*?)\n \];/, + ); + expect(schedulesMatch).not.toBeNull(); + expect(schedulesMatch![1]).toMatch( + /\[INTAKE_SLOT_DEDUP_QUEUE,\s*INTAKE_SLOT_DEDUP_CRON\]/, + ); + }); + + it("dispatches a payload without userId to the discovery fan-out", () => { + // The handler must branch on the missing `userId` BEFORE the per-user + // dedup call, and the branch must run the discovery helper. + expect(source).toMatch( + /INTAKE_SLOT_DEDUP_QUEUE[\s\S]{0,800}if \(!userId\) \{[\s\S]{0,400}enqueueBootTimeIntakeSlotDedup\(\)/, + ); + }); }); diff --git a/src/lib/jobs/reminder-worker.ts b/src/lib/jobs/reminder-worker.ts index 06ae8ae4f..c59be013c 100644 --- a/src/lib/jobs/reminder-worker.ts +++ b/src/lib/jobs/reminder-worker.ts @@ -331,6 +331,15 @@ const FITBIT_SYNC_CRON = "8 * * * *"; // every hour at :08 // to the WHOOP sweep (03:22), inside the maintenance window. const FITBIT_OAUTH_STATE_CLEANUP_QUEUE = "fitbit-oauth-state-cleanup"; const FITBIT_OAUTH_STATE_CLEANUP_CRON = "24 3 * * *"; +// v1.15.19 — daily duplicate dose-slot dedup discovery tick. The boot-time +// pass only ran on worker restart, so a cross-source duplicate slot created +// between deploys (a pending REMINDER row plus a standalone API/WEB row on +// the same instant) survived until the next reboot. The cron payload omits +// `userId`; the handler treats that as the discovery tick and fans out one +// per-user job (singletonKey-coalesced) exactly like the boot pass. Slots at +// 03:28 between the mood-reminder cleanup (03:25) and the inventory expire +// (03:30), inside the maintenance window. +const INTAKE_SLOT_DEDUP_CRON = "28 3 * * *"; const OFFHOST_BACKUP_QUEUE = "data-backup-offhost"; // 02:30 Europe/Berlin — runs after audit-log/idempotency cleanups so old // rows are gone before they're snapshotted, but before the existing @@ -2648,6 +2657,12 @@ export async function startReminderWorker() { // v1.7.0 — daily 03:40 Europe/Berlin prune for expired measurement // tombstones. [MEASUREMENT_TOMBSTONE_CLEANUP_QUEUE, MEASUREMENT_TOMBSTONE_CLEANUP_CRON], + // v1.15.19 — daily 03:28 Europe/Berlin duplicate dose-slot dedup + // discovery. The empty cron payload (no `userId`) is the handler's + // signal to run the discovery fan-out instead of a per-user pass, so + // cross-source duplicate slots created between deploys collapse within + // a day instead of waiting for the next worker reboot. + [INTAKE_SLOT_DEDUP_QUEUE, INTAKE_SLOT_DEDUP_CRON], // v1.10.0 — computed scores (WX-C). Nightly 04:45 Europe/Berlin // Recovery-score compute + store, after the rollup-feeding consolidation // + drain so the signals it reads are already folded. @@ -3175,13 +3190,29 @@ export async function startReminderWorker() { // to the same canonical slot; this handler keeps the winner, soft- // deletes the losers, normalises the winner's scheduledFor, and // recomputes the affected compliance rollups. Serial concurrency so the - // one-time pass never crowds the request pool. - await boss.work( + // pass never crowds the request pool. + // + // v1.15.19 — the queue also carries a daily cron tick (03:28, scheduled + // above) whose payload omits `userId`. That tick runs the SAME discovery + // fan-out the boot pass uses, so duplicate slots created between deploys + // collapse within a day instead of waiting for the next worker reboot. + await boss.work>( INTAKE_SLOT_DEDUP_QUEUE, { localConcurrency: INTAKE_SLOT_DEDUP_CONCURRENCY }, async (jobs) => { for (const job of jobs) { const { userId } = job.data; + if (!userId) { + // Daily discovery tick — fan out one per-user job per account + // still holding duplicate-slot candidates (singletonKey-coalesced, + // identical to the boot pass). + const result = await enqueueBootTimeIntakeSlotDedup(); + workerLog( + "info", + `[intake-slot-dedup] daily discovery enqueued=${result.enqueued} skipped=${result.skipped}${result.error ? ` error=${result.error}` : ""}`, + ); + continue; + } try { const summary = await dedupeUserIntakeSlots(userId); workerLog( diff --git a/src/lib/medications/intake-slot-dedup.ts b/src/lib/medications/intake-slot-dedup.ts index fab3673be..b8a01f6d0 100644 --- a/src/lib/medications/intake-slot-dedup.ts +++ b/src/lib/medications/intake-slot-dedup.ts @@ -1,5 +1,5 @@ /** - * v1.8.2 — one-time, idempotent cleanup for duplicate dose-slot rows. + * v1.8.2 — idempotent cleanup for duplicate dose-slot rows. * * The v1.8.2 write-path fix (`resolveCanonicalSlotInstant` + * `shouldMintMissedDoseRow`) converges every NEW intake write onto one @@ -10,9 +10,11 @@ * minute. The duplicate inflates the per-day `scheduled` count, paints a * phantom "taken", and confuses the "due now" prompt. * - * This module is the boot-time backfill that collapses those existing - * duplicates. It mirrors the `rollup-full-backfill` / `step-consolidation` - * boot pattern: + * This module is the backfill that collapses those existing duplicates. + * Discovery runs at worker boot AND on a daily cron tick (v1.15.19 — a + * duplicate created between deploys must not wait for the next reboot). + * It mirrors the `rollup-full-backfill` / `step-consolidation` boot + * pattern: * - a cheap discovery pre-query enqueues one job ONLY for users that * actually hold duplicate slot rows; * - the per-user handler groups live `MedicationIntakeEvent` rows into @@ -357,7 +359,9 @@ export async function dedupeUserIntakeSlots( } /** - * Boot-time discovery. Finds every user that holds more than one live + * Discovery pass — runs at worker boot and on the daily cron tick (the + * cron payload omits `userId`; the worker handler dispatches here). Finds + * every user that holds more than one live * `MedicationIntakeEvent` row that COULD be a duplicate slot — i.e. a * `(medicationId, scheduledFor-truncated-to-the-minute)` cluster with * more than one row, OR two live rows on the same medication within a diff --git a/src/lib/rollups/__tests__/medication-compliance-rollups.test.ts b/src/lib/rollups/__tests__/medication-compliance-rollups.test.ts index 7afc505c5..02d3ffe70 100644 --- a/src/lib/rollups/__tests__/medication-compliance-rollups.test.ts +++ b/src/lib/rollups/__tests__/medication-compliance-rollups.test.ts @@ -139,6 +139,35 @@ describe("recomputeMedicationComplianceForDay", () => { expect(deleteBinds).toContain("2026-05-18"); }); + it("aggregates at SLOT level so cross-source duplicate rows cannot inflate scheduled", async () => { + // v1.15.19 — the partial unique index carries `source`, so a pending + // REMINDER row and a taken API row can share one `scheduled_for` + // instant. A row-level COUNT(*) counted that slot twice. Pin that the + // statement groups by `scheduled_for` and folds each slot with + // taken-beats-skipped priority (BOOL_OR) before counting. Semantics + // (two live rows, one slot, different source → scheduled=1, taken=1) + // are proven on real Postgres in + // `tests/integration/medication-compliance-rollup-slot.test.ts`. + await recomputeMedicationComplianceForDay( + "user-1", + "med-1", + "2026-05-18", + "Europe/Berlin", + ); + + const insertSql = (executeRaw.mock.calls[0][0] as readonly string[]).join( + "?", + ); + expect(insertSql).toMatch(/GROUP BY\s+"scheduled_for"/); + expect(insertSql).toMatch(/BOOL_OR\("taken_at" IS NOT NULL AND NOT "skipped"\)/); + expect(insertSql).toMatch(/BOOL_OR\("skipped"\)/); + // Slot-level taken beats skipped: a slot is `skipped` only when no row + // in it is taken. + expect(insertSql).toMatch(/NOT "slot_taken" AND "slot_skipped"/); + // Scheduled counts slots (the grouped CTE), not raw event rows. + expect(insertSql).toMatch(/FROM slot/); + }); + it("emits a paired DELETE so a fully-cleared day removes its row", async () => { // The DELETE fires unconditionally and gates on NOT EXISTS — when // the day still has events the predicate matches zero rows. The diff --git a/src/lib/rollups/medication-compliance-rollups.ts b/src/lib/rollups/medication-compliance-rollups.ts index 7abbec4cb..d4fcc0cb3 100644 --- a/src/lib/rollups/medication-compliance-rollups.ts +++ b/src/lib/rollups/medication-compliance-rollups.ts @@ -30,8 +30,20 @@ * `calculateCompliance` / `complianceChips`; this tier deliberately remains * the cheaper coverage signal and is documented as such so the two are not * confused. The `scheduled` / `taken` column names reflect this: a day's - * `taken` is "intake rows logged taken", not "doses inside their on-time - * band". + * `taken` is "slots with an intake logged taken", not "doses inside their + * on-time band". + * + * v1.15.19 — the count is SLOT-level, not row-level. The partial unique + * index on `(user, medication, scheduled_for, source)` lets two live rows + * share one slot instant when their `source` differs (e.g. a pending + * REMINDER row plus a taken API row for the same dose); a row-level + * `COUNT(*)` double-counted that slot as two scheduled doses. The + * aggregate now groups by `scheduled_for` first: `scheduled` counts + * distinct slot instants, and each slot folds its rows with + * taken-beats-skipped priority (`BOOL_OR` of taken, then skipped only when + * no row in the slot is taken). This neutralises cross-source duplicate + * rows retroactively — any recompute over pre-existing duplicates yields + * the deduplicated counts without a data migration. * * Audit anchor: `.planning/round-v1438-perf-analysis.md` §2.5 + §5 P4. */ @@ -194,18 +206,31 @@ export async function recomputeMedicationComplianceForDay( // provides. The DELETE runs second so it cannot clobber a row the // INSERT just wrote — when the day still has events the // `WHERE NOT EXISTS` predicate matches zero rows. + // v1.15.19 — slot-level aggregation. The inner CTE folds the rows that + // share one `scheduled_for` instant (cross-source duplicates: pending + // REMINDER + taken API) into one slot with taken-beats-skipped priority; + // the outer aggregate then counts slots, not rows, so a duplicated slot + // can never inflate `scheduled`. Pure SQL — re-running the recompute + // over historic duplicates self-corrects the counts. await client.$executeRaw` - WITH aggregate AS ( + WITH slot AS ( SELECT - COUNT(*)::int AS scheduled, - SUM(CASE WHEN "skipped" THEN 0 WHEN "taken_at" IS NOT NULL THEN 1 ELSE 0 END)::int AS taken, - SUM(CASE WHEN "skipped" THEN 1 ELSE 0 END)::int AS skipped + BOOL_OR("taken_at" IS NOT NULL AND NOT "skipped") AS slot_taken, + BOOL_OR("skipped") AS slot_skipped FROM "medication_intake_events" WHERE "user_id" = ${userId} AND "medication_id" = ${medicationId} AND "deleted_at" IS NULL AND "scheduled_for" >= ${start} AND "scheduled_for" < ${windowEnd} + GROUP BY "scheduled_for" + ), + aggregate AS ( + SELECT + COUNT(*)::int AS scheduled, + COALESCE(SUM(CASE WHEN "slot_taken" THEN 1 ELSE 0 END), 0)::int AS taken, + COALESCE(SUM(CASE WHEN NOT "slot_taken" AND "slot_skipped" THEN 1 ELSE 0 END), 0)::int AS skipped + FROM slot ) INSERT INTO "medication_compliance_rollups" ("user_id", "medication_id", "day", "scheduled", "taken", "skipped", "computed_at") diff --git a/tests/integration/medication-compliance-rollup-slot.test.ts b/tests/integration/medication-compliance-rollup-slot.test.ts new file mode 100644 index 000000000..f80bbe07c --- /dev/null +++ b/tests/integration/medication-compliance-rollup-slot.test.ts @@ -0,0 +1,225 @@ +/** + * v1.15.19 — slot-level medication-compliance rollup aggregation. + * + * The partial unique index on `(user, medication, scheduled_for, source)` + * lets two LIVE intake rows share one slot instant when their `source` + * differs — the production shape: the reminder worker pre-mints a pending + * REMINDER row on the slot, and a client write that the resolver refused + * to snap (future-slot guard) used to land a standalone API row on the + * exact same instant. The rollup previously counted rows (`COUNT(*)`), so + * a two-dose day with one duplicated slot reported `scheduled=3..4`. + * + * These tests pin the corrected semantics on real Postgres: + * 1. Two live rows on one slot, different source, one taken → + * scheduled=1, taken=1. + * 2. Taken beats skipped inside a slot. + * 3. The recompute self-corrects historic duplicates with no data + * migration (pure SQL, retroactive). + */ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +import { getPrismaClient, truncateAllTables } from "./setup"; + +import { + dayKeyForScheduledFor, + recomputeMedicationComplianceForDay, +} from "@/lib/rollups/medication-compliance-rollups"; + +const TEST_USER_ID = "user-compliance-rollup-slot"; +const TZ = "Europe/Berlin"; + +vi.mock("@/lib/db-compat", () => ({ + ensureDbCompatibility: vi.fn().mockResolvedValue(undefined), +})); + +let medId = ""; + +beforeEach(async () => { + const prisma = getPrismaClient(); + await truncateAllTables(prisma); + await prisma.user.create({ + data: { + id: TEST_USER_ID, + username: "compliance-rollup-slot", + email: "compliance-rollup-slot@example.test", + timezone: TZ, + }, + }); + const med = await prisma.medication.create({ + data: { + userId: TEST_USER_ID, + name: "Ramipril", + dose: "5mg", + active: true, + }, + }); + medId = med.id; +}); + +// Fixed non-DST-boundary day so the slot instants are deterministic. +// 2026-03-10 07:00 / 19:00 Europe/Berlin (CET, UTC+1). +const SLOT_0700 = new Date("2026-03-10T06:00:00.000Z"); +const SLOT_1900 = new Date("2026-03-10T18:00:00.000Z"); + +async function readRollup() { + const prisma = getPrismaClient(); + const dayKey = dayKeyForScheduledFor(SLOT_0700, TZ); + await recomputeMedicationComplianceForDay(TEST_USER_ID, medId, dayKey, TZ); + return prisma.medicationComplianceRollup.findFirst({ + where: { userId: TEST_USER_ID, medicationId: medId, day: dayKey }, + }); +} + +describe("medication-compliance rollup — slot-level aggregation (real Postgres)", () => { + it("counts a cross-source duplicate slot once: scheduled=1, taken=1", async () => { + const prisma = getPrismaClient(); + // The pending REMINDER row the worker minted on the slot… + await prisma.medicationIntakeEvent.create({ + data: { + userId: TEST_USER_ID, + medicationId: medId, + scheduledFor: SLOT_0700, + takenAt: null, + skipped: false, + source: "REMINDER", + }, + }); + // …and the standalone API row a pre-fix client write parked on the + // exact same instant (the unique index tolerates it: source differs). + await prisma.medicationIntakeEvent.create({ + data: { + userId: TEST_USER_ID, + medicationId: medId, + scheduledFor: SLOT_0700, + takenAt: new Date("2026-03-10T06:02:00.000Z"), + skipped: false, + source: "API", + }, + }); + + const rollup = await readRollup(); + expect(rollup).not.toBeNull(); + expect(rollup?.scheduled).toBe(1); + expect(rollup?.taken).toBe(1); + expect(rollup?.skipped).toBe(0); + }); + + it("keeps a two-dose day at scheduled=2 when one slot is duplicated", async () => { + const prisma = getPrismaClient(); + await prisma.medicationIntakeEvent.createMany({ + data: [ + // Duplicated morning slot: pending REMINDER + taken API. + { + userId: TEST_USER_ID, + medicationId: medId, + scheduledFor: SLOT_0700, + takenAt: null, + skipped: false, + source: "REMINDER", + }, + { + userId: TEST_USER_ID, + medicationId: medId, + scheduledFor: SLOT_0700, + takenAt: new Date("2026-03-10T06:01:00.000Z"), + skipped: false, + source: "API", + }, + // Clean evening slot, still pending. + { + userId: TEST_USER_ID, + medicationId: medId, + scheduledFor: SLOT_1900, + takenAt: null, + skipped: false, + source: "REMINDER", + }, + ], + }); + + const rollup = await readRollup(); + // Pre-fix this day reported scheduled=3 (row count). Slot-level it is 2. + expect(rollup?.scheduled).toBe(2); + expect(rollup?.taken).toBe(1); + expect(rollup?.skipped).toBe(0); + }); + + it("taken beats skipped inside one slot", async () => { + const prisma = getPrismaClient(); + await prisma.medicationIntakeEvent.createMany({ + data: [ + { + userId: TEST_USER_ID, + medicationId: medId, + scheduledFor: SLOT_0700, + takenAt: null, + skipped: true, + source: "REMINDER", + }, + { + userId: TEST_USER_ID, + medicationId: medId, + scheduledFor: SLOT_0700, + takenAt: new Date("2026-03-10T06:05:00.000Z"), + skipped: false, + source: "API", + }, + ], + }); + + const rollup = await readRollup(); + expect(rollup?.scheduled).toBe(1); + expect(rollup?.taken).toBe(1); + expect(rollup?.skipped).toBe(0); + }); + + it("counts a purely skipped slot as skipped, not taken", async () => { + const prisma = getPrismaClient(); + await prisma.medicationIntakeEvent.create({ + data: { + userId: TEST_USER_ID, + medicationId: medId, + scheduledFor: SLOT_0700, + takenAt: null, + skipped: true, + source: "WEB", + }, + }); + + const rollup = await readRollup(); + expect(rollup?.scheduled).toBe(1); + expect(rollup?.taken).toBe(0); + expect(rollup?.skipped).toBe(1); + }); + + it("ignores soft-deleted rows when folding a slot", async () => { + const prisma = getPrismaClient(); + await prisma.medicationIntakeEvent.createMany({ + data: [ + { + userId: TEST_USER_ID, + medicationId: medId, + scheduledFor: SLOT_0700, + takenAt: new Date("2026-03-10T06:02:00.000Z"), + skipped: false, + source: "API", + deletedAt: new Date(), + }, + { + userId: TEST_USER_ID, + medicationId: medId, + scheduledFor: SLOT_0700, + takenAt: null, + skipped: false, + source: "REMINDER", + }, + ], + }); + + const rollup = await readRollup(); + // The tombstoned taken row must not paint the live pending slot taken. + expect(rollup?.scheduled).toBe(1); + expect(rollup?.taken).toBe(0); + expect(rollup?.skipped).toBe(0); + }); +});