fix(medications): harden the slot upsert against races, downgrades, and double-decrements

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.
This commit is contained in:
Marc-André Bombeck committed 2026-06-01 08:32:42 +02:00
1 parent 5047eb7a20
commit 710902a43c
11 files changed
+1385 -116

No files matched your search

@@ -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" } }),
);
});
});
+50 -56
View File
@@ -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<ReturnType<typeof consumeOneDose>> = null;
if (!skipped) {
if (!skipped && consumedTransition) {
try {
inventoryOutcome = await consumeOneDose({
userId: user.id,
@@ -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" } }),
);
});
});
+50 -32
View File
@@ -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<Response> {
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<Response> {
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 &&
@@ -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);
});
});
+75 -26
View File
@@ -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()}`);
}
}
@@ -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> = {}): 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<SlotIntakeRow>;
updateImpl?: (id: string) => Promise<SlotIntakeRow>;
}) {
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<typeof applyCanonicalSlotWrite>[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");
});
});
@@ -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(<the day>, 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());
});
});
@@ -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
@@ -22,6 +22,251 @@ import type { WorkerScheduleRow } from "@/lib/medications/scheduling/worker-help
type PrismaLike = Pick<PrismaClient, "medication" | "medicationIntakeEvent">;
/** 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<SlotIntakeRow[]> {
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<ApplyCanonicalSlotWriteResult> {
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<ApplyCanonicalSlotWriteResult> {
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,
@@ -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({