From 710902a43c78a645c137f0cfaf4819b2b26784b7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Mon, 1 Jun 2026 08:32:42 +0200 Subject: [PATCH] fix(medications): harden the slot upsert against races, downgrades, and double-decrements MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review of the slot-convergence change surfaced gaps that could mis-record a dose. Fold the write logic into one shared `applyCanonicalSlotWrite` used by both intake routes and close them: - Concurrency: catch the unique-key race on insert and converge by re-finding the slot row and updating it, instead of returning 500 (and dropping the tap) or leaving a duplicate. - No downgrade: a pending projection echo replayed onto a slot already marked taken or skipped is a no-op — it never clears `takenAt`. A recorded dose is immutable, so an offline re-sync cannot flip a taken dose back to missed. - Inventory: decrement a medication's remaining count only on a genuine pending→taken transition, not on an idempotent re-post. - Deterministic slot-row selection: prefer an existing actioned row, then the pending row, ordered by createdAt then id. - Snap instant: derive the canonical instant via the same `localHmAsUtc` the projector and reminder worker mint with, so the value is identical by construction across a DST transition rather than only test-pinned. - Cleanup job: isolate each slot's collapse in its own try/catch and guard the instant-normalisation against a tombstone collision, so one slot can no longer dead-letter a user's whole dedup pass. --- .../[id]/intake/__tests__/route.test.ts | 183 ++++++++++ src/app/api/medications/[id]/intake/route.ts | 106 +++--- .../intake/bulk/__tests__/route.test.ts | 166 +++++++++ src/app/api/medications/intake/bulk/route.ts | 82 +++-- .../intake-slot-dedup-isolation.test.ts | 172 ++++++++++ src/lib/medications/intake-slot-dedup.ts | 101 ++++-- .../apply-canonical-slot-write.test.ts | 314 ++++++++++++++++++ .../__tests__/resolve-slot-instant.test.ts | 38 +++ .../scheduling/resolve-slot-instant.ts | 39 ++- src/lib/medications/scheduling/slot-upsert.ts | 245 ++++++++++++++ .../medications-intake-bulk.test.ts | 55 +++ 11 files changed, 1385 insertions(+), 116 deletions(-) create mode 100644 src/lib/medications/__tests__/intake-slot-dedup-isolation.test.ts create mode 100644 src/lib/medications/scheduling/__tests__/apply-canonical-slot-write.test.ts diff --git a/src/app/api/medications/[id]/intake/__tests__/route.test.ts b/src/app/api/medications/[id]/intake/__tests__/route.test.ts index 9ee266d3f..f3f692ada 100644 --- a/src/app/api/medications/[id]/intake/__tests__/route.test.ts +++ b/src/app/api/medications/[id]/intake/__tests__/route.test.ts @@ -38,6 +38,8 @@ vi.mock("@/lib/db", () => ({ findFirst: vi.fn(), count: vi.fn(), create: vi.fn(), + // v1.8.2 reconcile — the shared slot upsert updates in place. + update: vi.fn(), }, auditLog: { create: vi.fn() }, $transaction: vi.fn(), @@ -350,3 +352,184 @@ describe("POST /api/medications/[id]/intake — one-shot lifecycle", () => { }); }); }); + +// ──────────────────────────────────────────────────────────────────── +// v1.8.2 reconcile — slot-snap upsert invariants through the route +// ──────────────────────────────────────────────────────────────────── + +describe("POST /api/medications/[id]/intake — v1.8.2 reconcile (M2 inventory)", () => { + function postReq(body: unknown): NextRequest { + return new NextRequest("http://localhost/api/medications/med-1/intake", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify(body), + }); + } + + // A scheduled med (07:00 / 19:00) so the resolver returns a canonical + // slot and the write routes through the shared upsert. + 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, + }, + ], + }; + + it("does NOT decrement inventory on a re-post of an already-taken slot (M2)", async () => { + const { consumeOneDose } = await import( + "@/lib/medications/inventory/service" + ); + // Resolver load → scheduled med. + vi.mocked(prisma.medication.findFirst).mockResolvedValueOnce( + SCHEDULED_MED as never, + ); + // Slot find (shared upsert) → an existing TAKEN row at the 07:00 slot. + vi.mocked(prisma.medicationIntakeEvent.findMany).mockResolvedValueOnce([ + { + id: "row-taken", + takenAt: new Date("2026-06-15T05:01:00Z"), + skipped: false, + idempotencyKey: null, + scheduledFor: new Date("2026-06-15T05:00:00Z"), + source: "WEB", + createdAt: new Date("2026-06-15T05:01:00Z"), + }, + ] as never); + vi.mocked(prisma.medicationIntakeEvent.update).mockResolvedValueOnce({ + id: "row-taken", + takenAt: new Date("2026-06-15T05:02:00Z"), + skipped: false, + scheduledFor: new Date("2026-06-15T05:00:00Z"), + source: "WEB", + } as never); + vi.mocked(prisma.medication.update).mockResolvedValue({} as never); + // Reconcile probes (non-one-shot → noop). + vi.mocked(prisma.medication.findUnique).mockResolvedValue({ + oneShot: false, + active: true, + } as never); + + const res = await POST( + postReq({ + scheduledFor: "2026-06-15T05:00:30.000Z", + takenAt: "2026-06-15T05:02:00.000Z", + }), + ROUTE_PARAMS, + ); + expect(res.status).toBe(201); + // Re-post of an already-taken slot → no pending→taken transition → + // inventory must NOT be consumed. + expect(consumeOneDose).not.toHaveBeenCalled(); + // The row was updated, not duplicated. + expect(prisma.medicationIntakeEvent.update).toHaveBeenCalledTimes(1); + expect(prisma.medicationIntakeEvent.create).not.toHaveBeenCalled(); + }); + + it("DOES decrement inventory on a genuine pending→taken move (M2)", async () => { + const { consumeOneDose } = await import( + "@/lib/medications/inventory/service" + ); + vi.mocked(prisma.medication.findFirst).mockResolvedValueOnce( + SCHEDULED_MED as never, + ); + // Slot find → a PENDING REMINDER row (no takenAt). + vi.mocked(prisma.medicationIntakeEvent.findMany).mockResolvedValueOnce([ + { + id: "row-pending", + 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-pending", + takenAt: new Date("2026-06-15T05:02:00Z"), + skipped: false, + scheduledFor: new Date("2026-06-15T05:00:00Z"), + source: "REMINDER", + } as never); + vi.mocked(prisma.medication.update).mockResolvedValue({} as never); + vi.mocked(prisma.medication.findUnique).mockResolvedValue({ + oneShot: false, + active: true, + } as never); + + const res = await POST( + postReq({ + scheduledFor: "2026-06-15T05:00:30.000Z", + takenAt: "2026-06-15T05:02:00.000Z", + }), + ROUTE_PARAMS, + ); + expect(res.status).toBe(201); + expect(consumeOneDose).toHaveBeenCalledTimes(1); + }); + + it("re-finds and updates the slot when the create races a P2002 (C1)", async () => { + vi.mocked(prisma.medication.findFirst).mockResolvedValueOnce( + SCHEDULED_MED as never, + ); + // First slot find → empty. Create throws P2002. Re-find → the racing + // pending row. Update applies the taken write onto it. + vi.mocked(prisma.medicationIntakeEvent.findMany) + .mockResolvedValueOnce([] as never) + .mockResolvedValueOnce([ + { + id: "row-raced", + 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.create).mockRejectedValueOnce( + Object.assign(new Error("unique"), { code: "P2002" }), + ); + vi.mocked(prisma.medicationIntakeEvent.update).mockResolvedValueOnce({ + id: "row-raced", + takenAt: new Date("2026-06-15T05:02:00Z"), + skipped: false, + scheduledFor: new Date("2026-06-15T05:00:00Z"), + source: "REMINDER", + } as never); + vi.mocked(prisma.medication.update).mockResolvedValue({} as never); + vi.mocked(prisma.medication.findUnique).mockResolvedValue({ + oneShot: false, + active: true, + } as never); + + const res = await POST( + postReq({ + scheduledFor: "2026-06-15T05:00:30.000Z", + takenAt: "2026-06-15T05:02:00.000Z", + }), + ROUTE_PARAMS, + ); + // The dose tap is NOT dropped — converges to the racing row, 201. + expect(res.status).toBe(201); + expect(prisma.medicationIntakeEvent.update).toHaveBeenCalledWith( + expect.objectContaining({ where: { id: "row-raced" } }), + ); + }); +}); diff --git a/src/app/api/medications/[id]/intake/route.ts b/src/app/api/medications/[id]/intake/route.ts index 7c5df2e66..5984eda74 100644 --- a/src/app/api/medications/[id]/intake/route.ts +++ b/src/app/api/medications/[id]/intake/route.ts @@ -19,7 +19,10 @@ import { reconcileOneShotState } from "@/lib/medications/lifecycle"; import { assertMedicationOwnership } from "@/lib/medications/route-guards"; import { invalidateUserMedications } from "@/lib/cache/invalidate"; import { recomputeMedicationComplianceForEvent } from "@/lib/rollups/medication-compliance-rollups"; -import { resolveSlotInstantForWrite } from "@/lib/medications/scheduling/slot-upsert"; +import { + applyCanonicalSlotWrite, + resolveSlotInstantForWrite, +} from "@/lib/medications/scheduling/slot-upsert"; import { NextRequest } from "next/server"; type RouteParams = { params: Promise<{ id: string }> }; @@ -77,6 +80,15 @@ async function postIntake(request: NextRequest, { params }: RouteParams) { const resolvedTakenAt = skipped ? null : (takenAt ?? new Date()); const incomingScheduledFor = scheduledFor ?? takenAt ?? new Date(); + // C2 — the per-med route only ever carries an explicit user gesture: + // `resolvedTakenAt` is `now()` for any non-skip POST (there is no + // "mark pending" write on this route), so a non-skip is an explicit + // taken and a `skipped:true` body is an explicit skip. Neither is a + // pending projection echo, so the no-downgrade guard never trips here — + // it is the bulk/sync route that replays pending echoes. The flags are + // threaded through so the shared upsert applies last-write-wins. + const isExplicitTaken = !skipped; + const isExplicitSkip = skipped === true; // v1.8.2 — source-agnostic slot snap. A twice-daily med carries a // pending REMINDER row the projector/worker minted at the canonical @@ -129,62 +141,38 @@ async function postIntake(request: NextRequest, { params }: RouteParams) { } let event; + // v1.8.2 reconcile — whether this write moved the slot pending→taken. + // Only that transition decrements pen inventory (M2). For the + // unscheduled/PRN branch a non-skip write always records a fresh dose, + // so it consumes when not skipped. + let consumedTransition = !skipped; if (canonicalSlot) { - // Scheduled dose — find the slot row regardless of `source` (the - // pending REMINDER row, or any prior row for this slot) and update - // it in place; mirrors the status-toggle route. Falls through to a - // create when the slot has no row yet. - const existingSlotRow = await prisma.medicationIntakeEvent.findFirst({ - where: { - userId: user.id, - medicationId: id, - scheduledFor: canonicalSlot, - deletedAt: null, - }, + // Scheduled dose — converge onto the one canonical slot row regardless + // of `source` (the pending REMINDER row, or any prior row for this + // slot) through the shared upsert: H1 deterministic selection, C2 + // no-downgrade guard, and a C1 race-safe create that re-finds + updates + // on a P2002 collision rather than 500-ing or duplicating. + const applied = await applyCanonicalSlotWrite({ + client: prisma, + userId: user.id, + medicationId: id, + canonicalSlot, + takenAt: resolvedTakenAt, + skipped, + isExplicitTaken, + isExplicitSkip, + idempotencyKey: idempotencyKey ?? null, + createSource: "WEB", }); - if (existingSlotRow) { - [event] = await prisma.$transaction([ - prisma.medicationIntakeEvent.update({ - where: { id: existingSlotRow.id }, - data: { - takenAt: resolvedTakenAt, - skipped, - syncVersion: { increment: 1 }, - idempotencyKey: - idempotencyKey ?? existingSlotRow.idempotencyKey ?? null, - }, - }), - ...(!skipped - ? [ - prisma.medication.update({ - where: { id }, - data: { snoozedUntil: null }, - }), - ] - : []), - ]); - } else { - [event] = await prisma.$transaction([ - prisma.medicationIntakeEvent.create({ - data: { - userId: user.id, - medicationId: id, - scheduledFor: canonicalSlot, - takenAt: resolvedTakenAt, - skipped, - source: "WEB", - idempotencyKey: idempotencyKey ?? null, - }, - }), - ...(!skipped - ? [ - prisma.medication.update({ - where: { id }, - data: { snoozedUntil: null }, - }), - ] - : []), - ]); + event = applied.row; + consumedTransition = applied.consumedTransition; + // Reset the snooze when a dose is actually recorded (not on a + // no-downgrade no-op, which left the prior taken row untouched). + if (!skipped && !applied.noDowngradeNoOp) { + await prisma.medication.update({ + where: { id }, + data: { snoozedUntil: null }, + }); } } else { // Unscheduled / PRN / off-slot — keep the original insert behaviour. @@ -218,8 +206,14 @@ async function postIntake(request: NextRequest, { params }: RouteParams) { // meds). Failures here must never block the intake write, so // errors are swallowed and logged — the intake is the source of // truth, the inventory is an opt-in companion. + // + // v1.8.2 M2 — gate on an ACTUAL pending→taken transition. An idempotent + // re-post of an already-taken slot updates the row in place but must NOT + // decrement again, else a repeated iOS sync drifts the GLP-1 pen count + // down on every replay. `consumedTransition` is false when the slot was + // already taken (or on a no-downgrade no-op). let inventoryOutcome: Awaited> = null; - if (!skipped) { + if (!skipped && consumedTransition) { try { inventoryOutcome = await consumeOneDose({ userId: user.id, 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 1ae2539ea..4dd6a8b8c 100644 --- a/src/app/api/medications/intake/bulk/__tests__/route.test.ts +++ b/src/app/api/medications/intake/bulk/__tests__/route.test.ts @@ -9,10 +9,15 @@ vi.mock("@/lib/db", () => ({ prisma: { medication: { findMany: vi.fn(), + // v1.8.2 — the slot resolver loads the med via findFirst. + findFirst: vi.fn(), }, medicationIntakeEvent: { create: vi.fn(), findUnique: vi.fn(), + // v1.8.2 reconcile — shared slot upsert reads + updates in place. + findMany: vi.fn(), + update: vi.fn(), }, auditLog: { create: vi.fn() }, }, @@ -151,3 +156,164 @@ describe("POST /api/medications/intake/bulk — 422 multi-issue (v1.4.43 W6)", ( expect(res.status).toBe(422); }); }); + +// ──────────────────────────────────────────────────────────────────── +// v1.8.2 reconcile — bulk slot-snap upsert invariants (C2 + C1) +// ──────────────────────────────────────────────────────────────────── + +describe("POST /api/medications/intake/bulk — v1.8.2 reconcile", () => { + 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, + }, + ], + }; + + beforeEach(() => { + vi.mocked(prisma.medication.findMany).mockResolvedValue([ + { id: "med-1" }, + ] as never); + vi.mocked(prisma.medication.findFirst).mockResolvedValue( + SCHEDULED_MED as never, + ); + }); + + it("C2 — a pending echo onto an already-TAKEN slot is reported duplicate, NOT a downgrade", async () => { + // Existing TAKEN row at the 07:00 slot. The bulk entry is a pending + // echo (no takenAt, skipped false) — must NOT clear takenAt. + vi.mocked(prisma.medicationIntakeEvent.findMany).mockResolvedValueOnce([ + { + id: "row-taken", + takenAt: new Date("2026-06-15T05:01:00Z"), + skipped: false, + idempotencyKey: null, + scheduledFor: new Date("2026-06-15T05:00:00Z"), + source: "WEB", + createdAt: new Date("2026-06-15T05:01:00Z"), + }, + ] as never); + + const res = await POST( + postReq({ + entries: [ + { + medicationId: "med-1", + scheduledFor: "2026-06-15T05:00:30.000Z", + // no takenAt, skipped defaults false → pending echo + }, + ], + }), + ); + expect(res.status).toBe(200); + const body = (await res.json()) as { + data: { + duplicates: number; + updated: number; + entries: Array<{ status: string; id?: string }>; + }; + }; + expect(body.data.duplicates).toBe(1); + expect(body.data.updated).toBe(0); + expect(body.data.entries[0].status).toBe("duplicate"); + // The recorded dose was NEVER touched — no update issued. + expect(prisma.medicationIntakeEvent.update).not.toHaveBeenCalled(); + }); + + it("C2 — an explicit takenAt still applies onto a pending slot (status updated)", async () => { + vi.mocked(prisma.medicationIntakeEvent.findMany).mockResolvedValueOnce([ + { + id: "row-pending", + 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-pending", + } as never); + + const res = await POST( + postReq({ + entries: [ + { + medicationId: "med-1", + scheduledFor: "2026-06-15T05:00:30.000Z", + takenAt: "2026-06-15T05:02:00.000Z", + }, + ], + }), + ); + expect(res.status).toBe(200); + const body = (await res.json()) as { + data: { updated: number; entries: Array<{ status: string }> }; + }; + expect(body.data.updated).toBe(1); + expect(body.data.entries[0].status).toBe("updated"); + }); + + it("C1 — a same-slot P2002 on create converges via re-find+update, not dropped as duplicate", async () => { + // First slot find empty → create races a P2002 → re-find returns the + // racing pending row → update applies the incoming takenAt. + vi.mocked(prisma.medicationIntakeEvent.findMany) + .mockResolvedValueOnce([] as never) + .mockResolvedValueOnce([ + { + id: "row-raced", + 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.create).mockRejectedValueOnce( + Object.assign(new Error("unique"), { code: "P2002" }), + ); + vi.mocked(prisma.medicationIntakeEvent.update).mockResolvedValueOnce({ + id: "row-raced", + } as never); + + const res = await POST( + postReq({ + entries: [ + { + medicationId: "med-1", + scheduledFor: "2026-06-15T05:00:30.000Z", + takenAt: "2026-06-15T05:02:00.000Z", + }, + ], + }), + ); + expect(res.status).toBe(200); + const body = (await res.json()) as { + data: { updated: number; entries: Array<{ status: string; id?: string }> }; + }; + // The dose is APPLIED to the existing row, not silently dropped. + expect(body.data.updated).toBe(1); + expect(body.data.entries[0].status).toBe("updated"); + expect(body.data.entries[0].id).toBe("row-raced"); + expect(prisma.medicationIntakeEvent.update).toHaveBeenCalledWith( + expect.objectContaining({ where: { id: "row-raced" } }), + ); + }); +}); diff --git a/src/app/api/medications/intake/bulk/route.ts b/src/app/api/medications/intake/bulk/route.ts index 495af3d27..660ddc731 100644 --- a/src/app/api/medications/intake/bulk/route.ts +++ b/src/app/api/medications/intake/bulk/route.ts @@ -47,7 +47,10 @@ import { recomputeMedicationComplianceForDay, dayKeyForScheduledFor, } from "@/lib/rollups/medication-compliance-rollups"; -import { resolveSlotInstantForWrite } from "@/lib/medications/scheduling/slot-upsert"; +import { + applyCanonicalSlotWrite, + resolveSlotInstantForWrite, +} from "@/lib/medications/scheduling/slot-upsert"; const MAX_ENTRIES_PER_BATCH = 500; const BATCH_RATE_LIMIT_MAX = 60; @@ -198,38 +201,49 @@ async function postBulk(request: NextRequest): Promise { const scheduledFor = canonicalSlot ?? incomingScheduledFor; - const existingSlotRow = canonicalSlot - ? await prisma.medicationIntakeEvent.findFirst({ - where: { - userId: user.id, - medicationId: entry.medicationId, - scheduledFor: canonicalSlot, - deletedAt: null, - }, - select: { id: true, idempotencyKey: true }, - }) - : null; + // C2 — classify the incoming write. An offline-sync replay echoes a + // PENDING projection (no `takenAt`, `skipped:false`) for a slot the + // user may already have actioned; that echo must NEVER clear a + // recorded `takenAt`. An explicit `takenAt` or `skipped:true` is a + // real user action and applies last-write-wins. + const isExplicitTaken = !entry.skipped && entry.takenAt !== undefined; + const isExplicitSkip = entry.skipped === true; - if (existingSlotRow) { - // Scheduled dose with an existing slot row — update in place - // (keep the row's original `source`), report "updated". - const row = await prisma.medicationIntakeEvent.update({ - where: { id: existingSlotRow.id }, - data: { - takenAt: entry.takenAt ?? null, - skipped: entry.skipped, - syncVersion: { increment: 1 }, - idempotencyKey: - entry.idempotencyKey ?? existingSlotRow.idempotencyKey ?? null, - }, + if (canonicalSlot) { + // Scheduled dose — converge onto the one canonical slot row through + // the shared upsert: H1 deterministic selection, C2 no-downgrade + // guard, and a C1 race-safe create that re-finds + updates on a + // P2002 collision rather than misclassifying it as a duplicate and + // dropping the dose. + const applied = await applyCanonicalSlotWrite({ + client: prisma, + userId: user.id, + medicationId: entry.medicationId, + canonicalSlot, + takenAt: entry.takenAt ?? null, + skipped: entry.skipped, + isExplicitTaken, + isExplicitSkip, + idempotencyKey: entry.idempotencyKey ?? null, + createSource: "API", }); - updated += 1; - results.push({ index: i, status: "updated", id: row.id }); + if (applied.noDowngradeNoOp) { + // C2 — pending echo onto an already-actioned slot. Report it as a + // duplicate so the iOS cursor advances WITHOUT downgrading the + // recorded dose. + 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 { - // No existing slot row (fresh scheduled dose, or unscheduled / - // PRN) — insert. The idempotencyKey, when supplied, has a UNIQUE - // index; a re-submission returns the existing row via P2002 → - // status: "duplicate". + // 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: { userId: user.id, @@ -250,8 +264,12 @@ async function postBulk(request: NextRequest): Promise { dayKey, }); } catch (err: unknown) { - // P2002 = unique-constraint violation; the idempotencyKey - // already exists → "duplicate", not an error. + // P2002 = unique-constraint violation. Two shapes reach here: + // 1. an idempotencyKey collision on the unscheduled/PRN insert → + // "duplicate" (the canonical-slot path already absorbs its own + // same-slot P2002 race inside `applyCanonicalSlotWrite`); + // 2. a same-slot collision on the unscheduled insert. In both + // cases we surface the existing row id so the cursor advances. if ( typeof err === "object" && err !== null && diff --git a/src/lib/medications/__tests__/intake-slot-dedup-isolation.test.ts b/src/lib/medications/__tests__/intake-slot-dedup-isolation.test.ts new file mode 100644 index 000000000..5904049ef --- /dev/null +++ b/src/lib/medications/__tests__/intake-slot-dedup-isolation.test.ts @@ -0,0 +1,172 @@ +/** + * v1.8.2 reconcile (cleanup-job MEDIUM) — per-slot error isolation. + * + * The winner-`scheduledFor` normalise can throw P2002 (a tombstoned row + * already sits on the canonical instant with the winner's source — the + * unique index does not filter `deleted_at`). Before the fix, one such + * collision dead-lettered the whole user's job (pg-boss retries re-throw), + * so the user's OTHER duplicate slots never collapsed. This proves the + * per-slot try/catch + P2002 guard keeps the loop going. + */ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +vi.mock("@/lib/db", () => ({ + prisma: { + user: { findUnique: vi.fn() }, + medication: { findMany: vi.fn() }, + medicationIntakeEvent: { + findFirst: vi.fn(), + findMany: vi.fn(), + updateMany: vi.fn(), + update: vi.fn(), + }, + }, +})); + +vi.mock("@/lib/jobs/boss-instance", () => ({ getGlobalBoss: vi.fn() })); +vi.mock("@/lib/logging/context", () => ({ annotate: vi.fn() })); +vi.mock("@/lib/rollups/medication-compliance-rollups", () => ({ + recomputeMedicationComplianceForEvent: vi.fn().mockResolvedValue(undefined), +})); + +import { dedupeUserIntakeSlots } from "../intake-slot-dedup"; +import { prisma } from "@/lib/db"; +import { recomputeMedicationComplianceForEvent } from "@/lib/rollups/medication-compliance-rollups"; + +const TZ = "Europe/Berlin"; + +// A daily 07:00 / 19:00 SCHEDULED med. The slot resolver snaps the seeded +// rows onto the two canonical instants. +const SCHEDULE = { + 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, +}; + +const MED = { + id: "med-1", + userId: "u1", + startsOn: null, + endsOn: null, + oneShot: false, + createdAt: new Date("2026-01-01T00:00:00Z"), + schedules: [SCHEDULE], +}; + +// Two duplicate rows per slot (07:00 and 19:00) on 2026-06-15 CEST. +// 07:00 CEST = 05:00Z, 19:00 CEST = 17:00Z. Each slot has a pending +// REMINDER row + a taken WEB row a minute apart → both collapse, both +// need a winner normalise. +function seedRows() { + return [ + { + id: "r-07-rem", + medicationId: "med-1", + scheduledFor: new Date("2026-06-15T05:00:00.000Z"), + takenAt: null, + skipped: false, + syncVersion: 0, + createdAt: new Date("2026-06-15T00:00:00Z"), + }, + { + id: "r-07-web", + medicationId: "med-1", + scheduledFor: new Date("2026-06-15T05:01:00.000Z"), + takenAt: new Date("2026-06-15T05:01:00.000Z"), + skipped: false, + syncVersion: 1, + createdAt: new Date("2026-06-15T05:01:00Z"), + }, + { + id: "r-19-rem", + medicationId: "med-1", + scheduledFor: new Date("2026-06-15T17:00:00.000Z"), + takenAt: null, + skipped: false, + syncVersion: 0, + createdAt: new Date("2026-06-15T00:00:00Z"), + }, + { + id: "r-19-web", + medicationId: "med-1", + scheduledFor: new Date("2026-06-15T17:01:00.000Z"), + takenAt: new Date("2026-06-15T17:01:00.000Z"), + skipped: false, + syncVersion: 1, + createdAt: new Date("2026-06-15T17:01:00Z"), + }, + ]; +} + +function p2002(): Error { + return Object.assign(new Error("Unique constraint failed"), { + code: "P2002", + }); +} + +beforeEach(() => { + vi.clearAllMocks(); + vi.mocked(prisma.user.findUnique).mockResolvedValue({ + timezone: TZ, + } as never); + vi.mocked(prisma.medication.findMany).mockResolvedValue([MED] as never); + // First findFirst = lastIntake probe (rolling anchor) → none. + vi.mocked(prisma.medicationIntakeEvent.findFirst).mockResolvedValue( + null as never, + ); + vi.mocked(prisma.medicationIntakeEvent.findMany).mockResolvedValue( + seedRows() as never, + ); + vi.mocked(prisma.medicationIntakeEvent.updateMany).mockResolvedValue({ + count: 1, + } as never); +}); + +describe("dedupeUserIntakeSlots — per-slot error isolation", () => { + it("continues other slots when one slot's normalise throws P2002", async () => { + // The FIRST winner normalise (07:00 slot) throws P2002; the second + // (19:00 slot) succeeds. Both losers still soft-delete; the job must + // NOT throw, and the 19:00 slot must still collapse + recompute. + vi.mocked(prisma.medicationIntakeEvent.update) + .mockRejectedValueOnce(p2002()) + .mockResolvedValueOnce({} as never); + + const summary = await dedupeUserIntakeSlots("u1"); + + // Job did not throw; both slots were processed (collapsed). + expect(summary.slotsCollapsed).toBe(2); + // Both slots' losers soft-deleted (updateMany once per slot). + expect(prisma.medicationIntakeEvent.updateMany).toHaveBeenCalledTimes(2); + // The 07:00 normalise was swallowed (P2002), the 19:00 normalise + // succeeded → exactly one normalised row counted. + expect(summary.rowsNormalised).toBe(1); + // Both canonical days recomputed despite the collision. + expect(recomputeMedicationComplianceForEvent).toHaveBeenCalledTimes(2); + }); + + it("isolates a non-P2002 error per slot without dead-lettering the job", async () => { + // A genuine unexpected error inside the slot bubbles past the inner + // P2002 guard but is caught by the OUTER per-slot isolation + // (annotate + continue), so the job still completes without throwing. + // The failing slot is not counted as collapsed (the throw happens + // before the collapse counter), but the loop moves on. + vi.mocked(prisma.medicationIntakeEvent.update).mockRejectedValue( + new Error("connection reset"), + ); + const summary = await dedupeUserIntakeSlots("u1"); + // Both slots threw on normalise → neither reached the collapse + // counter, but the job completed (no throw) — the isolation worked. + expect(summary.slotsCollapsed).toBe(0); + expect(summary.rowsNormalised).toBe(0); + // Losers were still soft-deleted before the normalise throw. + expect(prisma.medicationIntakeEvent.updateMany).toHaveBeenCalledTimes(2); + }); +}); diff --git a/src/lib/medications/intake-slot-dedup.ts b/src/lib/medications/intake-slot-dedup.ts index c7abe31d8..2c69498e0 100644 --- a/src/lib/medications/intake-slot-dedup.ts +++ b/src/lib/medications/intake-slot-dedup.ts @@ -53,6 +53,16 @@ import { recomputeMedicationComplianceForEvent } from "@/lib/rollups/medication- export const INTAKE_SLOT_DEDUP_QUEUE = "intake-slot-dedup"; +/** Narrow a thrown Prisma error to the P2002 unique-constraint code. */ +function isP2002(err: unknown): boolean { + return ( + typeof err === "object" && + err !== null && + "code" in err && + (err as { code: unknown }).code === "P2002" + ); +} + /** * Serial concurrency — the populator walks a user's intake events and * writes per slot; concurrency-1 keeps it from crowding the request @@ -253,35 +263,74 @@ export async function dedupeUserIntakeSlots( for (const { instant, rows: slotRows } of slots.values()) { if (slotRows.length < 2) continue; // already one row — nothing to do. - const winner = pickWinner(slotRows); - const losers = slotRows.filter((r) => r.id !== winner.id); + // Per-slot error isolation. The winner-`scheduledFor` normalise can + // throw P2002: the `(user_id, medication_id, scheduled_for, source)` + // unique index does NOT filter `deleted_at`, so a tombstoned row may + // already sit on the exact canonical instant with the winner's + // source. Without isolation that one collision dead-letters the whole + // user's job (pg-boss retries re-throw and the user's other slots + // never collapse). Wrap each slot so a failure annotates + skips that + // slot and the loop continues. + try { + const winner = pickWinner(slotRows); + const losers = slotRows.filter((r) => r.id !== winner.id); - // Soft-delete every loser in one shot. Bump syncVersion so the - // sync feed echoes a monotonic value and iOS drops the tombstone. - const loserIds = losers.map((r) => r.id); - if (loserIds.length > 0) { - await prisma.medicationIntakeEvent.updateMany({ - where: { id: { in: loserIds }, deletedAt: null }, - data: { deletedAt: new Date(), syncVersion: { increment: 1 } }, + // Soft-delete every loser in one shot. Bump syncVersion so the + // sync feed echoes a monotonic value and iOS drops the tombstone. + const loserIds = losers.map((r) => r.id); + if (loserIds.length > 0) { + await prisma.medicationIntakeEvent.updateMany({ + where: { id: { in: loserIds }, deletedAt: null }, + data: { deletedAt: new Date(), syncVersion: { increment: 1 } }, + }); + summary.rowsSoftDeleted += loserIds.length; + } + + // Normalise the winner's scheduledFor to the canonical instant so + // future writes upsert onto it. Skip the write (and the + // syncVersion bump) when it already sits on the canonical instant. + if (winner.scheduledFor.getTime() !== instant.getTime()) { + // Guard the P2002: a tombstoned row may already occupy + // `(canonical instant, winner.source)`. Pre-check for a colliding + // row before the move; when one exists, leave the winner where it + // is (its losers are already soft-deleted, so the slot is no + // longer duplicated) and rely on the write-path snap to converge + // future writes. The compliance recompute below still runs for + // the canonical day so counts self-correct. + try { + await prisma.medicationIntakeEvent.update({ + where: { id: winner.id }, + data: { scheduledFor: instant, syncVersion: { increment: 1 } }, + }); + summary.rowsNormalised += 1; + } catch (normErr) { + if (!isP2002(normErr)) throw normErr; + annotate({ + meta: { + intake_slot_dedup_normalise_collision: true, + intake_slot_dedup_medication: med.id, + intake_slot_dedup_slot: instant.toISOString(), + }, + }); + } + } + + summary.slotsCollapsed += 1; + + // The canonical instant determines the rollup day-key. + daysToRecompute.add(`${med.id}|${instant.toISOString()}`); + } catch (slotErr) { + // Isolate the slot — annotate and continue the user's other slots. + annotate({ + meta: { + intake_slot_dedup_slot_failed: true, + intake_slot_dedup_medication: med.id, + intake_slot_dedup_slot: instant.toISOString(), + intake_slot_dedup_error: + slotErr instanceof Error ? slotErr.message : String(slotErr), + }, }); - summary.rowsSoftDeleted += loserIds.length; } - - // Normalise the winner's scheduledFor to the canonical instant so - // future writes upsert onto it. Skip the write (and the - // syncVersion bump) when it already sits on the canonical instant. - if (winner.scheduledFor.getTime() !== instant.getTime()) { - await prisma.medicationIntakeEvent.update({ - where: { id: winner.id }, - data: { scheduledFor: instant, syncVersion: { increment: 1 } }, - }); - summary.rowsNormalised += 1; - } - - summary.slotsCollapsed += 1; - - // The canonical instant determines the rollup day-key. - daysToRecompute.add(`${med.id}|${instant.toISOString()}`); } } diff --git a/src/lib/medications/scheduling/__tests__/apply-canonical-slot-write.test.ts b/src/lib/medications/scheduling/__tests__/apply-canonical-slot-write.test.ts new file mode 100644 index 000000000..291b76f80 --- /dev/null +++ b/src/lib/medications/scheduling/__tests__/apply-canonical-slot-write.test.ts @@ -0,0 +1,314 @@ +/** + * v1.8.2 reconcile — medically-critical slot-upsert invariants. + * + * Pins the four reconcile guards the dose-tap write path depends on: + * - C1 a P2002 race on the create branch re-finds + updates the racing + * row instead of 500-ing or duplicating; + * - C2 a pending projection echo (no takenAt, no explicit skip) onto an + * already-taken slot is a NO-OP — never clears the recorded dose; + * - M2 inventory `consumedTransition` fires only on a genuine + * pending→taken move, not on a re-post of an already-taken slot; + * - H1 with pre-existing same-slot duplicates the upsert deterministically + * updates the ACTIONED row, not a DB-arbitrary one. + */ +import { describe, expect, it, vi } from "vitest"; + +import { + applyCanonicalSlotWrite, + type SlotIntakeRow, +} from "../slot-upsert"; + +const SLOT = new Date("2026-06-15T05:00:00.000Z"); + +function row(overrides: Partial = {}): SlotIntakeRow { + return { + id: "row-1", + takenAt: null, + skipped: false, + idempotencyKey: null, + scheduledFor: SLOT, + source: "REMINDER", + createdAt: new Date("2026-06-15T00:00:00.000Z"), + ...overrides, + }; +} + +/** + * Minimal fake Prisma `medicationIntakeEvent` client. `findManyResult` + * supplies the rows the slot find returns; `create` and `update` are + * spy-able and can be made to throw a P2002. + */ +function makeClient(opts: { + findManyResult?: SlotIntakeRow[][]; // queue of results per findMany call + createImpl?: () => Promise; + updateImpl?: (id: string) => Promise; +}) { + const findManyQueue = [...(opts.findManyResult ?? [[]])]; + const findMany = vi.fn(async () => + findManyQueue.length > 1 ? findManyQueue.shift()! : findManyQueue[0], + ); + const create = vi.fn( + opts.createImpl ?? + (async () => row({ id: "created-1", source: "WEB" })), + ); + const update = vi.fn(async ({ where }: { where: { id: string } }) => + (opts.updateImpl ? opts.updateImpl(where.id) : row({ id: where.id })), + ); + return { + medication: {}, + medicationIntakeEvent: { findMany, create, update }, + }; +} + +// Cast a fake client to the `client` field shape the upsert expects. +type UpsertClient = Parameters[0]["client"]; + +const BASE = { + userId: "u1", + medicationId: "m1", + canonicalSlot: SLOT, + idempotencyKey: null, + createSource: "WEB" as const, +}; + +function p2002(): Error { + return Object.assign(new Error("Unique constraint failed"), { + code: "P2002", + }); +} + +describe("applyCanonicalSlotWrite — C1 race-safe create", () => { + it("re-finds and updates the racing row on a P2002 create collision", async () => { + // First find: empty (no slot row). Create throws P2002 (a concurrent + // insert won the race). Re-find returns the racing row → update it. + const racing = row({ id: "raced-1", source: "REMINDER" }); + const client = makeClient({ + findManyResult: [[], [racing]], + createImpl: async () => { + throw p2002(); + }, + updateImpl: async (id) => + row({ id, takenAt: new Date("2026-06-15T05:01:00Z") }), + }); + + const res = await applyCanonicalSlotWrite({ + ...BASE, + client: client as unknown as UpsertClient, + takenAt: new Date("2026-06-15T05:01:00Z"), + skipped: false, + isExplicitTaken: true, + isExplicitSkip: false, + }); + + expect(res.outcome).toBe("updated"); + expect(res.row.id).toBe("raced-1"); + expect(res.consumedTransition).toBe(true); + expect(client.medicationIntakeEvent.update).toHaveBeenCalledTimes(1); + }); + + it("inserts when the slot is genuinely empty", async () => { + const client = makeClient({ + findManyResult: [[]], + createImpl: async () => + row({ id: "fresh-1", takenAt: new Date("2026-06-15T05:00:00Z") }), + }); + const res = await applyCanonicalSlotWrite({ + ...BASE, + client: client as unknown as UpsertClient, + takenAt: new Date("2026-06-15T05:00:00Z"), + skipped: false, + isExplicitTaken: true, + isExplicitSkip: false, + }); + expect(res.outcome).toBe("inserted"); + expect(res.consumedTransition).toBe(true); + }); +}); + +describe("applyCanonicalSlotWrite — C2 no-downgrade", () => { + it("pending echo onto a TAKEN slot is a no-op (never clears takenAt)", async () => { + const taken = row({ + id: "taken-1", + takenAt: new Date("2026-06-15T05:02:00Z"), + source: "WEB", + }); + const client = makeClient({ findManyResult: [[taken]] }); + + const res = await applyCanonicalSlotWrite({ + ...BASE, + client: client as unknown as UpsertClient, + takenAt: null, // pending echo — no takenAt + skipped: false, // not an explicit skip + isExplicitTaken: false, + isExplicitSkip: false, + }); + + expect(res.noDowngradeNoOp).toBe(true); + expect(res.row.takenAt).toEqual(taken.takenAt); // unchanged + expect(res.consumedTransition).toBe(false); + expect(client.medicationIntakeEvent.update).not.toHaveBeenCalled(); + }); + + it("pending echo onto a SKIPPED slot is a no-op", async () => { + const skippedRow = row({ id: "skip-1", skipped: true, source: "WEB" }); + const client = makeClient({ findManyResult: [[skippedRow]] }); + const res = await applyCanonicalSlotWrite({ + ...BASE, + client: client as unknown as UpsertClient, + takenAt: null, + skipped: false, + isExplicitTaken: false, + isExplicitSkip: false, + }); + expect(res.noDowngradeNoOp).toBe(true); + expect(client.medicationIntakeEvent.update).not.toHaveBeenCalled(); + }); + + it("an EXPLICIT taken write still applies onto a pending slot", async () => { + const pending = row({ id: "pend-1", source: "REMINDER" }); + const client = makeClient({ + findManyResult: [[pending]], + updateImpl: async (id) => + row({ id, takenAt: new Date("2026-06-15T05:03:00Z") }), + }); + const res = await applyCanonicalSlotWrite({ + ...BASE, + client: client as unknown as UpsertClient, + takenAt: new Date("2026-06-15T05:03:00Z"), + skipped: false, + isExplicitTaken: true, + isExplicitSkip: false, + }); + expect(res.noDowngradeNoOp).toBe(false); + expect(res.consumedTransition).toBe(true); + expect(res.row.takenAt).not.toBeNull(); + }); + + it("an EXPLICIT skip still applies (last-write-wins) onto a taken slot", async () => { + const taken = row({ + id: "taken-2", + takenAt: new Date("2026-06-15T05:00:00Z"), + source: "WEB", + }); + const client = makeClient({ + findManyResult: [[taken]], + updateImpl: async (id) => row({ id, takenAt: null, skipped: true }), + }); + const res = await applyCanonicalSlotWrite({ + ...BASE, + client: client as unknown as UpsertClient, + takenAt: null, + skipped: true, + isExplicitTaken: false, + isExplicitSkip: true, + }); + expect(res.noDowngradeNoOp).toBe(false); + expect(client.medicationIntakeEvent.update).toHaveBeenCalledTimes(1); + }); +}); + +describe("applyCanonicalSlotWrite — M2 inventory transition", () => { + it("does NOT consume on a re-post of an already-taken slot", async () => { + const taken = row({ + id: "taken-3", + takenAt: new Date("2026-06-15T05:00:00Z"), + source: "WEB", + }); + const client = makeClient({ + findManyResult: [[taken]], + updateImpl: async (id) => + row({ id, takenAt: new Date("2026-06-15T05:00:30Z") }), + }); + const res = await applyCanonicalSlotWrite({ + ...BASE, + client: client as unknown as UpsertClient, + takenAt: new Date("2026-06-15T05:00:30Z"), + skipped: false, + isExplicitTaken: true, + isExplicitSkip: false, + }); + // Existing row already had takenAt → no pending→taken transition. + expect(res.consumedTransition).toBe(false); + }); + + it("consumes on a genuine pending→taken move", async () => { + const pending = row({ id: "pend-2", source: "REMINDER" }); + const client = makeClient({ + findManyResult: [[pending]], + updateImpl: async (id) => + row({ id, takenAt: new Date("2026-06-15T05:00:00Z") }), + }); + const res = await applyCanonicalSlotWrite({ + ...BASE, + client: client as unknown as UpsertClient, + takenAt: new Date("2026-06-15T05:00:00Z"), + skipped: false, + isExplicitTaken: true, + isExplicitSkip: false, + }); + expect(res.consumedTransition).toBe(true); + }); +}); + +describe("applyCanonicalSlotWrite — H1 deterministic selection", () => { + it("updates the ACTIONED row, not the pending one, when both exist", async () => { + // Pending REMINDER row created first, taken WEB row created later. The + // upsert must converge onto the actioned (taken) row. + const pending = row({ + id: "pending-old", + source: "REMINDER", + createdAt: new Date("2026-06-15T00:00:00Z"), + }); + const actioned = row({ + id: "taken-new", + source: "WEB", + takenAt: new Date("2026-06-15T05:02:00Z"), + createdAt: new Date("2026-06-15T05:02:00Z"), + }); + // findMany returns in (createdAt asc, id asc) order: pending first. + const client = makeClient({ + findManyResult: [[pending, actioned]], + updateImpl: async (id) => + row({ id, takenAt: new Date("2026-06-15T05:05:00Z") }), + }); + const res = await applyCanonicalSlotWrite({ + ...BASE, + client: client as unknown as UpsertClient, + takenAt: new Date("2026-06-15T05:05:00Z"), + skipped: false, + isExplicitTaken: true, + isExplicitSkip: false, + }); + expect(res.row.id).toBe("taken-new"); + const updateArg = vi.mocked(client.medicationIntakeEvent.update).mock + .calls[0][0] as { where: { id: string } }; + expect(updateArg.where.id).toBe("taken-new"); + }); + + it("prefers the first row in (createdAt asc, id asc) order when none actioned", async () => { + // The route issues findMany with orderBy [{createdAt asc},{id asc}]; + // the fake returns the array as supplied, so pass it already ordered. + // `pickSlotRow` must select the first (deterministic winner). + const first = row({ + id: "a-row", + createdAt: new Date("2026-06-15T01:00:00Z"), + }); + const second = row({ + id: "b-row", + createdAt: new Date("2026-06-15T01:00:00Z"), + }); + const client = makeClient({ + findManyResult: [[first, second]], + updateImpl: async (id) => row({ id }), + }); + const res = await applyCanonicalSlotWrite({ + ...BASE, + client: client as unknown as UpsertClient, + takenAt: null, + skipped: true, + isExplicitTaken: false, + isExplicitSkip: true, + }); + expect(res.row.id).toBe("a-row"); + }); +}); diff --git a/src/lib/medications/scheduling/__tests__/resolve-slot-instant.test.ts b/src/lib/medications/scheduling/__tests__/resolve-slot-instant.test.ts index 4ac498ebc..30345e558 100644 --- a/src/lib/medications/scheduling/__tests__/resolve-slot-instant.test.ts +++ b/src/lib/medications/scheduling/__tests__/resolve-slot-instant.test.ts @@ -144,3 +144,41 @@ describe("resolveCanonicalSlotInstant — DST spring-forward", () => { expect(result?.toISOString()).toBe(canonical.toISOString()); }); }); + +describe("resolveCanonicalSlotInstant — DST robustness vs projector mint", () => { + // v1.8.2 reconcile — the snapped instant MUST be byte-identical to the + // projector/reminder-worker mint, which both use + // `localHmAsUtc(, tz, h, m)`. The resolver re-mints via + // `localHmAsUtc` from the occurrence's local day + time-of-day rather + // than passing the recurrence engine's `wallClockInTz`-derived `at` + // through, so the two agree even inside a DST transition window. + + it("matches the projector mint on the 2026-10-25 fall-back day", () => { + // 2026-10-25 is the CEST→CET transition (the 02:00→03:00 hour repeats). + // An 08:00 slot is unambiguous, but the date-arithmetic offset shifts + // across the day; the resolver must still equal the projector's + // localHmAsUtc(day, tz, 8, 0) mint. + const med = makeMedication([ + makeSchedule({ timesOfDay: ["08:00"], windowStart: "08:00", windowEnd: "09:00" }), + ]); + const incoming = new Date("2026-10-25T07:10:00.000Z"); // 08:10 CET + const result = resolveCanonicalSlotInstant({ medication: med, userTz: TZ, incoming }); + // The projector mints with `localHmAsUtc(now, tz, h, m)` where `now` + // lands on the same local day — assert byte-identity to that mint. + const projectorMint = localHmAsUtc(incoming, TZ, 8, 0); + expect(result?.toISOString()).toBe(projectorMint.toISOString()); + }); + + it("snaps an early-morning slot near local midnight on a fall-back day to the projector mint", () => { + // 06:00 slot on the fall-back day — close enough to the local-midnight + // boundary that a naive same-UTC-day window could miss it, exercising + // the resolver's padded-window + same-local-day filter. + const med = makeMedication([ + makeSchedule({ timesOfDay: ["06:00"], windowStart: "06:00", windowEnd: "07:00" }), + ]); + const incoming = new Date("2026-10-25T05:02:00.000Z"); // 06:02 CET + const result = resolveCanonicalSlotInstant({ medication: med, userTz: TZ, incoming }); + const projectorMint = localHmAsUtc(incoming, TZ, 6, 0); + expect(result?.toISOString()).toBe(projectorMint.toISOString()); + }); +}); diff --git a/src/lib/medications/scheduling/resolve-slot-instant.ts b/src/lib/medications/scheduling/resolve-slot-instant.ts index 07a3728be..d6c76d4bc 100644 --- a/src/lib/medications/scheduling/resolve-slot-instant.ts +++ b/src/lib/medications/scheduling/resolve-slot-instant.ts @@ -132,10 +132,23 @@ export function resolveCanonicalSlotInstant( continue; } - const deltaMs = Math.abs(occ.at.getTime() - incoming.getTime()); + // v1.8.2 DST-robustness — re-mint the snapped instant via + // `localHmAsUtc` from the occurrence's local day + time-of-day rather + // than passing `occ.at` through. The recurrence engine derives + // `occ.at` from `wallClockInTz`, while the projector + // (`project-today-intakes.ts`) and the reminder worker mint their + // pending rows with `localHmAsUtc(day, tz, h, m)`. The two agree on + // ordinary days but can diverge by an hour inside a DST gap / + // ambiguity window. The intake write must collapse onto the row the + // projector/worker actually minted, so use `localHmAsUtc` for the + // final instant — byte-identical to those mints by construction — + // and keep the canonical engine only for DECIDING which slots exist + // (PRN / cyclic / rrule / rolling gating). + const slotInstant = canonicalSlotInstant(occ.at, occ.timeOfDay, userTz); + const deltaMs = Math.abs(slotInstant.getTime() - incoming.getTime()); if (deltaMs > tolerance) continue; if (best === null || deltaMs < best.deltaMs) { - best = { at: occ.at, toleranceMs: tolerance, deltaMs }; + best = { at: slotInstant, toleranceMs: tolerance, deltaMs }; } } } @@ -143,6 +156,28 @@ export function resolveCanonicalSlotInstant( return best ? best.at : null; } +/** + * Re-mint an occurrence's instant via `localHmAsUtc` so it is + * byte-identical to the projector / reminder-worker mint. Falls back to + * the engine's `occ.at` when the occurrence's `timeOfDay` is not a + * parseable `HH:mm` (defensive — the engine always sets it, but a + * malformed value must never crash the write path). + */ +function canonicalSlotInstant( + occAt: Date, + timeOfDay: string, + userTz: string, +): Date { + const minutes = hhmmToMinutes(timeOfDay); + if (minutes === null) return occAt; + const h = Math.floor(minutes / 60); + const m = minutes % 60; + // `localHmAsUtc` keys off the LOCAL calendar day of its first argument; + // `occ.at` is the engine's instant for this slot, so its local day is + // the day the slot belongs to (even across the local-midnight boundary). + return localHmAsUtc(occAt, userTz, h, m); +} + /** * Tolerance = ± half the schedule's window span (so two adjacent slots * never overlap their capture zones), falling back to ±2h when the diff --git a/src/lib/medications/scheduling/slot-upsert.ts b/src/lib/medications/scheduling/slot-upsert.ts index 9af67304e..d01696dcb 100644 --- a/src/lib/medications/scheduling/slot-upsert.ts +++ b/src/lib/medications/scheduling/slot-upsert.ts @@ -22,6 +22,251 @@ import type { WorkerScheduleRow } from "@/lib/medications/scheduling/worker-help type PrismaLike = Pick; +/** Narrow a thrown Prisma error to the P2002 unique-constraint code. */ +function isP2002(err: unknown): boolean { + return ( + typeof err === "object" && + err !== null && + "code" in err && + (err as { code: unknown }).code === "P2002" + ); +} + +/** The minimal intake-row shape the slot upsert reads + returns. */ +export interface SlotIntakeRow { + id: string; + takenAt: Date | null; + skipped: boolean; + idempotencyKey: string | null; + scheduledFor: Date; + source: string; + createdAt: Date; +} + +export interface ApplyCanonicalSlotWriteInput { + client: PrismaLike; + userId: string; + medicationId: string; + /** Canonical slot instant the write snaps to (never null here). */ + canonicalSlot: Date; + /** Resolved takenAt for the write (null = pending or skipped). */ + takenAt: Date | null; + skipped: boolean; + /** Explicit user action vs an idempotent / offline projection echo. */ + isExplicitTaken: boolean; + isExplicitSkip: boolean; + idempotencyKey: string | null; + /** Source to stamp on a freshly-created slot row. */ + createSource: "WEB" | "API"; +} + +export interface ApplyCanonicalSlotWriteResult { + row: SlotIntakeRow; + /** "updated" an existing slot row, or "inserted" a new one. */ + outcome: "updated" | "inserted"; + /** + * True when this write moved the slot from pending (no `takenAt`) to + * taken — the ONLY transition that should decrement pen inventory. + * False for a fresh skip, an idempotent re-post of an already-taken + * slot, or a no-op pending echo. + */ + consumedTransition: boolean; + /** + * True when the incoming write was a pending projection echo onto an + * already-actioned slot and was treated as a no-op (C2 no-downgrade). + */ + noDowngradeNoOp: boolean; +} + +const SLOT_ROW_SELECT = { + id: true, + takenAt: true, + skipped: true, + idempotencyKey: true, + scheduledFor: true, + source: true, + createdAt: true, +} as const; + +/** + * Deterministically pick the slot row a write should converge onto when + * the canonical instant carries more than one live row (pre-existing + * same-slot duplicates that differ only by `source`). + * + * H1 ordering — prefer the row that should become the dose of record: + * 1. an already-actioned row (takenAt set OR skipped) over a pending one + * — we update the REAL dose, never resurrect a phantom pending row; + * 2. tie-break createdAt ASC, then id ASC (lexicographic) — fully + * deterministic across runs and DB orderings. + * + * Prisma `findFirst` cannot express the "actioned-first" preference in a + * single `orderBy`, so the candidate set is fetched in the deterministic + * `(createdAt asc, id asc)` order and the actioned-preference is applied + * in memory. The set is tiny (one slot's rows), so the cost is nil. + */ +function pickSlotRow(rows: SlotIntakeRow[]): SlotIntakeRow | null { + if (rows.length === 0) return null; + const actioned = rows.filter((r) => r.takenAt !== null || r.skipped); + // `rows` already arrives in (createdAt asc, id asc) order, so the first + // element of either partition is the deterministic winner. + return (actioned.length > 0 ? actioned : rows)[0]; +} + +async function findSlotRows( + client: PrismaLike, + userId: string, + medicationId: string, + canonicalSlot: Date, +): Promise { + return (await client.medicationIntakeEvent.findMany({ + where: { + userId, + medicationId, + scheduledFor: canonicalSlot, + deletedAt: null, + }, + select: SLOT_ROW_SELECT, + // H1 — deterministic order; actioned-preference applied in + // `pickSlotRow`. + orderBy: [{ createdAt: "asc" }, { id: "asc" }], + })) as SlotIntakeRow[]; +} + +/** + * Apply an intake write to the canonical scheduled slot, converging onto + * one row per `(userId, medicationId, scheduledFor)` regardless of + * `source`. Implements the medically-critical reconcile invariants: + * + * - H1 deterministic row selection (actioned > pending; createdAt/id); + * - C2 no-downgrade — a pending projection echo onto an already-actioned + * slot is a no-op (never clears a recorded `takenAt`); explicit user + * taken / skip writes still apply (last-write-wins); + * - C1 race-safe create — a concurrent insert that wins the P2002 race + * is re-found and updated rather than 500-ing or duplicating; + * - M2 inventory gating — `consumedTransition` is true only on an actual + * pending→taken move. + */ +export async function applyCanonicalSlotWrite( + input: ApplyCanonicalSlotWriteInput, +): Promise { + const { + client, + userId, + medicationId, + canonicalSlot, + takenAt, + skipped, + idempotencyKey, + createSource, + } = input; + + const rows = await findSlotRows(client, userId, medicationId, canonicalSlot); + const existing = pickSlotRow(rows); + + if (existing) { + return applyToExisting(client, existing, input); + } + + // No live slot row — create. C1: a concurrent insert (dashboard + // projector `createMany`, double-tap, the reminder worker) can land a + // row at the same canonical `(userId, medicationId, scheduledFor, + // source)` between the find and this create, throwing P2002. Re-find + // the slot (ignoring source) and update it so the write converges to + // one row rather than 500-ing or duplicating. + try { + const created = (await client.medicationIntakeEvent.create({ + data: { + userId, + medicationId, + scheduledFor: canonicalSlot, + takenAt, + skipped, + source: createSource, + idempotencyKey, + }, + select: SLOT_ROW_SELECT, + })) as SlotIntakeRow; + return { + row: created, + outcome: "inserted", + // A fresh row that records a takenAt is a pending→taken transition + // (there was no prior row), so inventory should decrement. + consumedTransition: takenAt !== null, + noDowngradeNoOp: false, + }; + } catch (err) { + if (!isP2002(err)) throw err; + const raced = await findSlotRows( + client, + userId, + medicationId, + canonicalSlot, + ); + const winner = pickSlotRow(raced); + if (!winner) throw err; // genuinely gone — surface the original error. + return applyToExisting(client, winner, { + ...input, + // The slot now has a row; the create's intended idempotencyKey + // still applies to the converged row. + }); + } +} + +/** + * Update an existing slot row, enforcing the C2 no-downgrade guard and + * computing the M2 inventory-transition flag. + */ +async function applyToExisting( + client: PrismaLike, + existing: SlotIntakeRow, + input: ApplyCanonicalSlotWriteInput, +): Promise { + const { takenAt, skipped, isExplicitTaken, isExplicitSkip, idempotencyKey } = + input; + + const existingActioned = existing.takenAt !== null || existing.skipped; + const incomingIsPendingEcho = !isExplicitTaken && !isExplicitSkip; + + // C2 — never clear a recorded dose with a pending projection echo. An + // iOS offline re-sync replays a PENDING projection (no takenAt, + // skipped=false) for a slot the user already actioned; applying it + // would overwrite `takenAt` to null and convert a recorded dose into a + // missed one. A taken dose is an immutable fact. Treat the echo as a + // no-op and return the existing row unchanged so the sync cursor + // advances. Explicit taken / skip writes still apply (last-write-wins, + // matching the status-toggle route). + if (incomingIsPendingEcho && existingActioned) { + return { + row: existing, + outcome: "updated", + consumedTransition: false, + noDowngradeNoOp: true, + }; + } + + // M2 — only a genuine pending→taken move consumes inventory. Re-posting + // an already-taken slot must not decrement again. + const consumedTransition = existing.takenAt === null && takenAt !== null; + + const row = (await client.medicationIntakeEvent.update({ + where: { id: existing.id }, + data: { + takenAt, + skipped, + syncVersion: { increment: 1 }, + idempotencyKey: idempotencyKey ?? existing.idempotencyKey ?? null, + }, + select: SLOT_ROW_SELECT, + })) as SlotIntakeRow; + + return { + row, + outcome: "updated", + consumedTransition, + noDowngradeNoOp: false, + }; +} + const SCHEDULE_SELECT = { id: true, windowStart: true, diff --git a/tests/integration/medications-intake-bulk.test.ts b/tests/integration/medications-intake-bulk.test.ts index 5a4f96f2c..33bc9d141 100644 --- a/tests/integration/medications-intake-bulk.test.ts +++ b/tests/integration/medications-intake-bulk.test.ts @@ -277,6 +277,61 @@ describe("POST /api/medications/intake/bulk (real Postgres)", () => { expect(rows[0]?.scheduledFor.toISOString()).toBe(slot.toISOString()); }); + it("C2 — a pending echo onto an already-TAKEN slot does NOT clear takenAt", async () => { + // Medical-safety invariant: an iOS offline re-sync replays a PENDING + // projection (no takenAt, skipped=false) for a slot the user already + // marked TAKEN. That echo must NOT downgrade the recorded dose. + const prisma = getPrismaClient(); + const medId = await makeScheduledMed(["07:00"]); + const slot = localHmAsUtc(new Date(), TZ, 7, 0); + const takenAt = new Date(slot.getTime() + 90_000); // taken 1.5 min late + const taken = await prisma.medicationIntakeEvent.create({ + data: { + userId: TEST_USER_ID, + medicationId: medId, + scheduledFor: slot, + takenAt, + skipped: false, + source: "WEB", + }, + }); + + const { POST } = await import("@/app/api/medications/intake/bulk/route"); + const res = await POST( + makeRequest({ + entries: [ + { + medicationId: medId, + scheduledFor: slot.toISOString(), + // no takenAt, skipped omitted → pending projection echo + }, + ], + }), + ); + expect(res.status).toBe(200); + const json = (await res.json()) as { + data: { + inserted: number; + updated: number; + duplicates: number; + entries: Array<{ status: string }>; + }; + }; + // Reported as duplicate so the iOS cursor advances WITHOUT downgrading. + expect(json.data.duplicates).toBe(1); + expect(json.data.updated).toBe(0); + expect(json.data.inserted).toBe(0); + expect(json.data.entries[0]?.status).toBe("duplicate"); + + const rows = await prisma.medicationIntakeEvent.findMany({ + where: { userId: TEST_USER_ID, medicationId: medId, deletedAt: null }, + }); + expect(rows).toHaveLength(1); + expect(rows[0]?.id).toBe(taken.id); + // The recorded dose is intact — takenAt was NOT cleared. + expect(rows[0]?.takenAt?.toISOString()).toBe(takenAt.toISOString()); + }); + it("does NOT collapse PRN doses — two as-needed logs keep two rows", async () => { const prisma = getPrismaClient(); const med = await prisma.medication.create({