fix(medications): stock claims are atomic and only a dose becoming taken consumes

Two hardening passes on the consumption stamp:

- consumeForIntake claims the stamp with a single conditional UPDATE
  (NULL-stamp predicate + row lock) instead of read-then-write, so a
  double tap racing a sync replay decrements exactly once; the loser
  sees count 0. restoreForIntake clears the stamp the same way, so
  concurrent restores cannot refund twice.
- Every seam consumes only on the pending-to-taken transition. A sync
  replay that re-posts an already-taken history row, or an edit that
  corrects the timestamp of an old dose, never touches today's stock.
This commit is contained in:
Marc-André Bombeck committed 2026-06-12 18:48:30 +02:00
1 parent e21dfd040e
commit 7eeab2ae98
9 files changed
+345 -90

No files matched your search

+19 -8
View File
@@ -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,
@@ -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"
@@ -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,
@@ -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([
{
+13 -7
View File
@@ -422,11 +422,16 @@ async function postBulk(request: NextRequest): Promise<Response> {
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<Response> {
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,
+31 -19
View File
@@ -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({
+18 -9
View File
@@ -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 });
@@ -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<string, unknown>;
}) => {
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();
});
});
+48 -34
View File
@@ -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: {