diff --git a/src/app/api/ingest/medication/route.ts b/src/app/api/ingest/medication/route.ts index 4b01a61c1..30e6c77b7 100644 --- a/src/app/api/ingest/medication/route.ts +++ b/src/app/api/ingest/medication/route.ts @@ -191,6 +191,12 @@ export const POST = apiHandler(async (request: NextRequest) => { }); let event; + // v1.16.10 — only the genuine pending→taken transition consumes + // inventory units (the slot upsert's `consumedTransition` flag). A + // re-post converging onto an already-taken slot row — including a + // pre-v1.16.10 row that predates the consumption stamp — must not + // decrement stock again. A fresh ad-hoc create is always a take. + let consumed = true; if (attribution.slotInstant) { // Scheduled dose — converge onto the one canonical slot row (the // pending REMINDER row when one exists) through the shared upsert: @@ -209,6 +215,7 @@ export const POST = apiHandler(async (request: NextRequest) => { attributionSource: "AUTO", }); event = applied.row; + consumed = applied.consumedTransition; } else { // Ad-hoc / PRN / off-window — standalone row under the documented // contract (`scheduledFor = takenAt`). @@ -227,14 +234,18 @@ export const POST = apiHandler(async (request: NextRequest) => { // v1.16.10 — the external ingest records a take; consume inventory // units. The stamp on the row keeps a retried POST exactly-once (the - // idempotency probe above already short-circuits the common replay). - await consumeForIntake({ - client: prisma, - userId: apiToken.userId, - medicationId: medication.id, - eventId: event.id, - intakeAt: effectiveTakenAt, - }); + // idempotency probe above already short-circuits the common replay) + // and the transition gate keeps a slot-converged re-post from + // double-decrementing through a pre-stamp row. + if (consumed) { + await consumeForIntake({ + client: prisma, + userId: apiToken.userId, + medicationId: medication.id, + eventId: event.id, + intakeAt: effectiveTakenAt, + }); + } await auditLog("medication.ingest.external", { userId: apiToken.userId, diff --git a/src/app/api/medications/[id]/intake/[eventId]/__tests__/route.test.ts b/src/app/api/medications/[id]/intake/[eventId]/__tests__/route.test.ts index b832029b5..e5f6bab07 100644 --- a/src/app/api/medications/[id]/intake/[eventId]/__tests__/route.test.ts +++ b/src/app/api/medications/[id]/intake/[eventId]/__tests__/route.test.ts @@ -623,6 +623,40 @@ describe("PUT / DELETE — v1.16.10 inventory consumption", () => { expect(restoreForIntake).not.toHaveBeenCalled(); }); + it("an in-place time correction on an already-taken row never consumes (pre-stamp legacy rows must not retro-charge)", async () => { + const { consumeForIntake, restoreForIntake } = await import( + "@/lib/medications/inventory/consumption" + ); + vi.mocked(prisma.medicationIntakeEvent.findFirst) + // PUT lookup — a row already taken BEFORE the edit, stamp NULL + // (pre-v1.16.10: its stock moved through the legacy hook at take + // time). The edit only corrects the takenAt minute. + .mockResolvedValueOnce({ + id: "e1", + userId: "user-1", + medicationId: "m1", + scheduledFor: at(7, 0), + takenAt: at(7, 5), + skipped: false, + inventoryConsumption: null, + } as never) + .mockResolvedValue(null as never); + vi.mocked(prisma.medicationIntakeEvent.update).mockResolvedValue({ + id: "e1", + scheduledFor: at(7, 0), + takenAt: at(7, 10), + skipped: false, + } as never); + + const res = await PUT( + putReq({ takenAt: at(7, 10).toISOString() }), + ROUTE_CTX, + ); + expect(res.status).toBe(200); + expect(consumeForIntake).not.toHaveBeenCalled(); + expect(restoreForIntake).not.toHaveBeenCalled(); + }); + it("a taken → skipped edit restores the row's stamp and never consumes", async () => { const { consumeForIntake, restoreForIntake } = await import( "@/lib/medications/inventory/consumption" diff --git a/src/app/api/medications/[id]/intake/[eventId]/route.ts b/src/app/api/medications/[id]/intake/[eventId]/route.ts index c6440447f..c3c2a857e 100644 --- a/src/app/api/medications/[id]/intake/[eventId]/route.ts +++ b/src/app/api/medications/[id]/intake/[eventId]/route.ts @@ -225,6 +225,14 @@ export const PUT = apiHandler( const slotMoved = targetScheduledFor.getTime() !== event.scheduledFor.getTime(); + // v1.16.10 — transition gate for the inventory hooks below. Only an + // edit moving the row INTO taken may consume; a row that was + // already taken keeps its stamp frozen (a stamped row would no-op + // anyway, and a pre-v1.16.10 taken row — NULL stamp, stock already + // moved by the legacy hook at take time — must never retro-consume + // on a time correction). + const wasTaken = event.takenAt !== null && !event.skipped; + let updated; if (!slotMoved) { // The slot is unchanged — a bare in-place update can never collide with @@ -245,19 +253,22 @@ export const PUT = apiHandler( syncVersion: { increment: 1 }, }, }); - // v1.16.10 — inventory follows the post-edit state: a row that is - // (now) taken consumes (the stamp keeps an already-consumed row - // untouched — the stamp freezes what the original take pulled, - // even when `unitsPerDose` changed since); a row edited out of - // taken (skip flip, takenAt cleared) refunds its stamp. + // v1.16.10 — inventory follows the post-edit state: an edit that + // moves the row INTO taken consumes; a row that stays taken keeps + // its stamp untouched (the stamp freezes what the original take + // pulled, even when `unitsPerDose` changed since, and a pre-stamp + // legacy row must not retro-consume); a row edited out of taken + // (skip flip, takenAt cleared) refunds its stamp. if (nextTakenAt !== null && !nextSkipped) { - await consumeForIntake({ - client: prisma, - userId: user.id, - medicationId: id, - eventId, - intakeAt: nextTakenAt, - }); + if (!wasTaken) { + await consumeForIntake({ + client: prisma, + userId: user.id, + medicationId: id, + eventId, + intakeAt: nextTakenAt, + }); + } } else { await restoreForIntake({ client: prisma, @@ -304,7 +315,16 @@ export const PUT = apiHandler( doseTaken: event.doseTaken, }); updated = applied.row; - if (nextTakenAt !== null && !nextSkipped) { + // A taken source WITHOUT a stamp is a pre-v1.16.10 row whose + // stock moved through the legacy hook: the refund above was a + // no-op, so consuming on the converged row would double-charge — + // skip it. Every other taken outcome nets the move to exactly one + // consumption. + if ( + nextTakenAt !== null && + !nextSkipped && + (!wasTaken || event.inventoryConsumption !== null) + ) { await consumeForIntake({ client: prisma, 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 61c827638..f35456c20 100644 --- a/src/app/api/medications/intake/bulk/__tests__/route.test.ts +++ b/src/app/api/medications/intake/bulk/__tests__/route.test.ts @@ -926,6 +926,43 @@ describe("POST /api/medications/intake/bulk — v1.16.10 inventory consumption", expect(restoreForIntake).not.toHaveBeenCalled(); }); + it("a taken re-post onto an already-taken slot (journal replay) never reaches consume", async () => { + // The replayed row pre-dates the consumption stamp (NULL) — exactly + // the shape a full client journal re-sync re-posts for historical + // doses. The upsert reports no pending→taken transition, so the + // consume hook must not run: the stamp cannot gate a pre-v1.16.10 + // row and the replay must not drain today's stock. + 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", + } 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); + expect(consumeForIntake).not.toHaveBeenCalled(); + expect(restoreForIntake).not.toHaveBeenCalled(); + }); + it("an explicit skip entry restores the slot row's stamp instead of consuming", async () => { vi.mocked(prisma.medicationIntakeEvent.findMany).mockResolvedValueOnce([ { diff --git a/src/app/api/medications/intake/bulk/route.ts b/src/app/api/medications/intake/bulk/route.ts index aa231b9ea..b923db783 100644 --- a/src/app/api/medications/intake/bulk/route.ts +++ b/src/app/api/medications/intake/bulk/route.ts @@ -422,11 +422,16 @@ async function postBulk(request: NextRequest): Promise { inserted += 1; results.push({ index: i, status: "inserted", id: applied.row.id }); } - // v1.16.10 — inventory: a taken write consumes (the stamp on the - // row makes a replayed sync batch exactly-once), an explicit skip - // refunds a previously-taken row's stamp. Pending echoes carry - // neither flag and never touch stock. - if (!applied.noDowngradeNoOp && isExplicitTaken) { + // v1.16.10 — inventory: only the genuine pending→taken + // transition consumes (the upsert's `consumedTransition` flag — + // the ONLY transition that may move stock). The stamp keeps a + // replayed batch exactly-once for post-v1.16.10 rows; the + // transition gate additionally protects pre-stamp rows: a full + // journal replay re-posting historical taken doses must not + // drain today's stock through their NULL stamps. An explicit + // skip refunds a previously-taken row's stamp. Pending echoes + // carry neither flag and never touch stock. + if (applied.consumedTransition && isExplicitTaken) { await consumeForIntake({ client: prisma, userId: user.id, @@ -509,8 +514,9 @@ async function postBulk(request: NextRequest): Promise { inserted += 1; results.push({ index: i, status: "inserted", id: applied.row.id }); } - // v1.16.10 — same inventory rule as the canonical-slot branch. - if (!applied.noDowngradeNoOp && isExplicitTaken) { + // v1.16.10 — same inventory rule as the canonical-slot branch: + // gate on the upsert's pending→taken transition flag. + if (applied.consumedTransition && isExplicitTaken) { await consumeForIntake({ client: prisma, userId: user.id, diff --git a/src/app/api/medications/intake/route.ts b/src/app/api/medications/intake/route.ts index 2ffc077f3..002c91589 100644 --- a/src/app/api/medications/intake/route.ts +++ b/src/app/api/medications/intake/route.ts @@ -523,6 +523,12 @@ export const POST = apiHandler(async (request: NextRequest) => { const slotMoved = targetScheduledFor.getTime() !== existing.scheduledFor.getTime(); + // v1.16.10 — transition gate for the inventory consume below. Only + // a row moving INTO taken may consume; a re-take of an already- + // taken row (e.g. a time correction) leaves its stamp frozen, and a + // pre-v1.16.10 taken row (NULL stamp, stock already moved by the + // legacy hook at take time) must never retro-consume. + const wasTaken = existing.takenAt !== null && !existing.skipped; if (!slotMoved) { [updated] = await prisma.$transaction([ prisma.medicationIntakeEvent.update({ @@ -544,17 +550,18 @@ export const POST = apiHandler(async (request: NextRequest) => { data: { snoozedUntil: null }, }), ]); - // v1.16.10 — the toggle recorded a take; consume inventory units. - // The stamp on the row makes a re-toggle of an already-taken slot - // a no-op, so the count only moves on the genuine pending→taken - // (or skipped→taken) transition. - await consumeForIntake({ - client: prisma, - userId: user.id, - medicationId: existing.medicationId, - eventId: intakeId, - intakeAt: resolvedTakenAt, - }); + // v1.16.10 — the toggle recorded a take; consume inventory units + // on the genuine pending→taken (or skipped→taken) transition. The + // stamp additionally keeps a racing replay exactly-once. + if (!wasTaken) { + await consumeForIntake({ + client: prisma, + userId: user.id, + medicationId: existing.medicationId, + eventId: intakeId, + intakeAt: resolvedTakenAt, + }); + } } else { // The take re-attributed to a different slot. Tombstone the source row // and route the dose through the shared canonical-slot upsert, which @@ -600,14 +607,19 @@ export const POST = apiHandler(async (request: NextRequest) => { data: { snoozedUntil: null }, }); // v1.16.10 — consume on the converged row (the old row's stamp - // was refunded above, so the move nets one consumption). - await consumeForIntake({ - client: prisma, - userId: user.id, - medicationId: existing.medicationId, - eventId: applied.row.id, - intakeAt: resolvedTakenAt, - }); + // was refunded above, so the move nets one consumption). A taken + // source WITHOUT a stamp is a pre-v1.16.10 row whose stock moved + // through the legacy hook: the refund above was a no-op, so + // consuming on the target would double-charge — skip it. + if (!wasTaken || existing.inventoryConsumption !== null) { + await consumeForIntake({ + client: prisma, + userId: user.id, + medicationId: existing.medicationId, + eventId: applied.row.id, + intakeAt: resolvedTakenAt, + }); + } } } else if (status === "skipped") { updated = await prisma.medicationIntakeEvent.update({ diff --git a/src/app/api/telegram/webhook/route.ts b/src/app/api/telegram/webhook/route.ts index e5e5620ef..9d480846d 100644 --- a/src/app/api/telegram/webhook/route.ts +++ b/src/app/api/telegram/webhook/route.ts @@ -275,6 +275,10 @@ async function markMedicationTaken( const written = await prisma.$transaction(async (tx) => { let slotScheduledFor: Date; let eventId: string; + // v1.16.10 — only a genuine pending→taken transition may consume + // inventory units; a redelivered button press converging onto an + // already-taken slot row (including a pre-stamp row) must not. + let consumedTransition = true; if (canonicalSlot) { const applied = await applyCanonicalSlotWrite({ client: tx, @@ -291,6 +295,7 @@ async function markMedicationTaken( }); slotScheduledFor = applied.row.scheduledFor; eventId = applied.row.id; + consumedTransition = applied.consumedTransition; } else { // Ad-hoc / PRN — standalone row under the documented contract // (`scheduledFor = takenAt`). @@ -312,20 +317,24 @@ async function markMedicationTaken( where: { id: medication.id }, data: { snoozedUntil: null }, }); - return { slotScheduledFor, eventId }; + return { slotScheduledFor, eventId, consumedTransition }; }); scheduledFor = written.slotScheduledFor; // v1.16.10 — the button confirmed a take; consume inventory units. // Runs after the intake transaction committed (the dose record must // never hinge on the stock write); the stamp keeps a Telegram - // redelivery exactly-once. - await consumeForIntake({ - client: prisma, - userId, - medicationId: medication.id, - eventId: written.eventId, - intakeAt: takenAt, - }); + // redelivery exactly-once and the transition gate keeps a re-press + // on an already-taken slot from decrementing through a pre-stamp + // row. + if (written.consumedTransition) { + await consumeForIntake({ + client: prisma, + userId, + medicationId: medication.id, + eventId: written.eventId, + intakeAt: takenAt, + }); + } // The confirmation must reflect on the next read (today feed, card // pill, compliance) rather than wait out the cache TTL. invalidateUserMedications(userId, { evict: true }); diff --git a/src/lib/medications/inventory/__tests__/consumption.test.ts b/src/lib/medications/inventory/__tests__/consumption.test.ts index 8929d7bf6..d61163d4a 100644 --- a/src/lib/medications/inventory/__tests__/consumption.test.ts +++ b/src/lib/medications/inventory/__tests__/consumption.test.ts @@ -90,6 +90,55 @@ function makeClient(state: FakeState) { (e) => e.id === args.where.id && e.userId === args.where.userId, ) ?? null, ), + // The hooks' atomic claim — evaluate the conditional WHERE the way + // Postgres would (the gate fields plus the stamp predicate: + // `Prisma.AnyNull` matches a NULL stamp, a JSON value matches by + // structural equality) and apply the data to every matching row. + updateMany: vi.fn( + async (args: { + where: { + id?: string; + userId?: string; + medicationId?: string; + deletedAt?: null; + takenAt?: { not: null }; + skipped?: boolean; + inventoryConsumption?: { equals: unknown }; + }; + data: Record; + }) => { + const w = args.where; + const matches = state.events.filter((e) => { + if (w.id !== undefined && e.id !== w.id) return false; + if (w.userId !== undefined && e.userId !== w.userId) return false; + if (w.medicationId !== undefined && e.medicationId !== w.medicationId) + return false; + if ("deletedAt" in w && e.deletedAt !== null) return false; + if (w.takenAt !== undefined && e.takenAt === null) return false; + if (w.skipped !== undefined && e.skipped !== w.skipped) return false; + if (w.inventoryConsumption !== undefined) { + const eq = w.inventoryConsumption.equals; + if (eq === Prisma.AnyNull) { + if (e.inventoryConsumption !== null) return false; + } else if ( + JSON.stringify(e.inventoryConsumption) !== JSON.stringify(eq) + ) { + return false; + } + } + return true; + }); + for (const event of matches) { + if ("inventoryConsumption" in args.data) { + event.inventoryConsumption = + args.data.inventoryConsumption === Prisma.DbNull + ? null + : args.data.inventoryConsumption; + } + } + return { count: matches.length }; + }, + ), update: vi.fn( async (args: { where: { id: string }; @@ -653,3 +702,66 @@ describe("restoreForIntake", () => { ]); }); }); + +describe("claim atomicity (concurrent duplicate calls)", () => { + it("two interleaved consumes decrement exactly once — the claim, not a read, is the gate", async () => { + const state: FakeState = { + unitsPerDose: 1, + events: [takenEvent()], + items: [ + item({ + id: "open", + state: "IN_USE", + unitsRemaining: 4, + firstUseAt: new Date(NOW.getTime() - MS_PER_DAY), + expiresAt: new Date(NOW.getTime() + 29 * MS_PER_DAY), + }), + ], + }; + const client = makeClient(state); + + // Both calls pass any plain read of the (still NULL) stamp; only the + // conditional claim UPDATE — atomic per statement, like Postgres + // under the row lock — lets exactly one through. + await Promise.all([ + consumeForIntake(consumeArgs(client)), + consumeForIntake(consumeArgs(client)), + ]); + + expect(state.items[0].unitsRemaining).toBe(3); + expect(state.events[0].inventoryConsumption).toEqual([ + { itemId: "open", units: 1 }, + ]); + }); + + it("two interleaved restores refund exactly once", async () => { + const state: FakeState = { + unitsPerDose: 1, + events: [ + takenEvent({ + inventoryConsumption: [{ itemId: "open", units: 1 }], + }), + ], + items: [ + item({ + id: "open", + state: "IN_USE", + unitsRemaining: 3, + firstUseAt: new Date(NOW.getTime() - MS_PER_DAY), + expiresAt: new Date(NOW.getTime() + 29 * MS_PER_DAY), + }), + ], + }; + const client = makeClient(state); + + const args = { + client: client as never, + userId: "user-1", + eventId: "evt-1", + }; + await Promise.all([restoreForIntake(args), restoreForIntake(args)]); + + expect(state.items[0].unitsRemaining).toBe(4); + expect(state.events[0].inventoryConsumption).toBeNull(); + }); +}); diff --git a/src/lib/medications/inventory/consumption.ts b/src/lib/medications/inventory/consumption.ts index c80a38d2b..8ba4e8d57 100644 --- a/src/lib/medications/inventory/consumption.ts +++ b/src/lib/medications/inventory/consumption.ts @@ -135,26 +135,30 @@ export async function consumeForIntake(input: { const { client, userId, medicationId, eventId, intakeAt } = input; try { await runAtomically(client, async (tx) => { - const event = await tx.medicationIntakeEvent.findFirst({ - where: { id: eventId, userId }, - select: { - id: true, - medicationId: true, - takenAt: true, - skipped: true, - deletedAt: true, - inventoryConsumption: true, + // Atomic claim — gate check and stamp reservation in ONE + // conditional UPDATE. A plain read-then-write gate is not safe + // under READ COMMITTED: two concurrent consume calls for the same + // event (a double tap racing an iOS sync replay) both read the + // NULL stamp and both decrement. The claim takes the event's row + // lock and re-evaluates the predicate after a blocking peer + // commits, so exactly one caller wins; the loser sees count 0 and + // returns. The placeholder `[]` is overwritten with the real + // stamp below inside the same transaction — and rolls back with + // it if the consumption fails, so a failed consume never leaves a + // row stamped. + const claimed = await tx.medicationIntakeEvent.updateMany({ + where: { + id: eventId, + userId, + medicationId, + deletedAt: null, + takenAt: { not: null }, + skipped: false, + inventoryConsumption: { equals: Prisma.AnyNull }, }, + data: { inventoryConsumption: [] }, }); - if ( - !event || - event.deletedAt !== null || - event.takenAt === null || - event.skipped || - event.inventoryConsumption !== null - ) { - return; - } + if (claimed.count === 0) return; const medication = await tx.medication.findFirst({ where: { id: medicationId, userId }, @@ -249,7 +253,7 @@ export async function consumeForIntake(input: { // with no tracked stock must not retro-consume containers // registered later when a sync replay revisits the row. await tx.medicationIntakeEvent.update({ - where: { id: event.id }, + where: { id: eventId }, data: { inventoryConsumption: stamp as unknown as Prisma.InputJsonValue }, }); @@ -302,17 +306,32 @@ export async function restoreForIntake(input: { inventoryConsumption: true, }, }); - if (!event) return; - const stamp = parseStamp(event.inventoryConsumption); + if (!event || event.inventoryConsumption === null) return; + const stampValue = event.inventoryConsumption; + + // Atomic claim — clear the stamp ONLY while it still holds the + // exact value just read (jsonb structural equality). The claim + // takes the event's row lock, so two concurrent restores cannot + // both refund: the loser blocks, re-evaluates against the + // now-NULL stamp, sees count 0 and returns. The refund below then + // applies the captured stamp inside the same transaction; a + // mid-refund failure rolls the clear back with it. + const claimed = await tx.medicationIntakeEvent.updateMany({ + where: { + id: eventId, + userId, + inventoryConsumption: { + equals: stampValue as Prisma.InputJsonValue, + }, + }, + data: { inventoryConsumption: Prisma.DbNull }, + }); + if (claimed.count === 0) return; + + const stamp = parseStamp(stampValue); if (stamp === null || stamp.length === 0) { - // Clear an empty/malformed stamp so a later re-take can - // consume afresh; a NULL stamp is already clear. - if (event.inventoryConsumption !== null) { - await tx.medicationIntakeEvent.update({ - where: { id: event.id }, - data: { inventoryConsumption: Prisma.DbNull }, - }); - } + // Empty / malformed stamp — the claim already cleared it so a + // later re-take can consume afresh; nothing to refund. return; } @@ -351,11 +370,6 @@ export async function restoreForIntake(input: { itemIds.push(item.id); } - await tx.medicationIntakeEvent.update({ - where: { id: event.id }, - data: { inventoryConsumption: Prisma.DbNull }, - }); - annotate({ action: { name: "medication.inventory.restored" }, meta: {