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