412 lines
19 KiB
TypeScript
412 lines
19 KiB
TypeScript
/**
|
|
* Postgres reservation store — production twin of engine/capacity.ts MemoryReservationStore.
|
|
*
|
|
* Concurrency model
|
|
* - Every (shop, location, method, date, slot) has a CapacityCounter row; slot "*" is the day total.
|
|
* - Counters = confirmed bookings + holds that have not been swept yet.
|
|
* - Any mutation runs in one transaction that locks the counter rows it touches
|
|
* with SELECT … FOR UPDATE, always in sorted key order (no deadlocks; "*" sorts
|
|
* before "HH:MM" so the day row is always locked first).
|
|
* - Expired holds for a locked date are swept inside the same transaction, so a stale
|
|
* hold can never block a live customer.
|
|
*
|
|
* Guarantees (covered by reservations.server.test.ts against real Postgres):
|
|
* - N concurrent customers for the last K places → exactly K succeed.
|
|
* - Duplicate order webhooks → one booking (unique shopId+orderId).
|
|
* - Expired hold at order time → re-reserved if space, otherwise flagged needs_review.
|
|
* - Reschedule releases old + reserves new atomically; failure leaves the booking untouched.
|
|
*/
|
|
import type { Prisma, PrismaClient } from "@prisma/client";
|
|
import type { Caps, Selection } from "../engine/capacity";
|
|
import type { Method, UsageSnapshot } from "../engine/types";
|
|
import { usageKey } from "../engine/availability";
|
|
|
|
type Tx = Prisma.TransactionClient;
|
|
interface Key {
|
|
locationId: string;
|
|
method: string;
|
|
date: string;
|
|
slot: string;
|
|
}
|
|
|
|
const keyStr = (k: Key) => `${k.locationId}|${k.method}|${k.date}|${k.slot}`;
|
|
const keysFor = (s: Selection): Key[] => [
|
|
{ locationId: s.locationId, method: s.method, date: s.date, slot: "*" },
|
|
{ locationId: s.locationId, method: s.method, date: s.date, slot: s.slot },
|
|
];
|
|
const sameSel = (a: Selection, b: Selection) =>
|
|
a.locationId === b.locationId && a.method === b.method && a.date === b.date && a.slot === b.slot;
|
|
|
|
export type HoldResult =
|
|
| { ok: true; token: string; expiresAt: Date }
|
|
| { ok: false; code: "DAY_FULL" | "SLOT_FULL"; message: string };
|
|
|
|
export interface BookingDetails {
|
|
orderName?: string;
|
|
timezone: string;
|
|
startUtc?: Date;
|
|
endUtc?: Date;
|
|
customerName?: string;
|
|
customerEmail?: string;
|
|
customerPhone?: string;
|
|
address?: Prisma.InputJsonValue;
|
|
items?: Prisma.InputJsonValue;
|
|
fee?: number;
|
|
peakDayId?: string | null;
|
|
configVersion: number;
|
|
snapshot: Prisma.InputJsonValue;
|
|
}
|
|
|
|
export type ConfirmResult =
|
|
| { ok: true; bookingId: string; duplicate: boolean; reReserved: boolean }
|
|
| { ok: false; bookingId: string; code: "NEEDS_REVIEW"; message: string; duplicate: boolean };
|
|
|
|
export class ReservationService {
|
|
constructor(
|
|
private db: PrismaClient,
|
|
private opts: { txTimeoutMs?: number } = {},
|
|
) {}
|
|
|
|
private tx<T>(fn: (tx: Tx) => Promise<T>): Promise<T> {
|
|
return this.db.$transaction(fn, { timeout: this.opts.txTimeoutMs ?? 10000, maxWait: 10000 });
|
|
}
|
|
|
|
/** Ensure + lock counter rows in deterministic order; returns counts by key. */
|
|
private async lock(tx: Tx, shopId: string, keys: Key[]): Promise<Map<string, number>> {
|
|
const uniq = [...new Map(keys.map((k) => [keyStr(k), k])).values()].sort((a, b) =>
|
|
keyStr(a) < keyStr(b) ? -1 : 1,
|
|
);
|
|
const out = new Map<string, number>();
|
|
for (const k of uniq) {
|
|
await tx.$executeRaw`
|
|
INSERT INTO "CapacityCounter" ("shopId","locationId","method","date","slot","count")
|
|
VALUES (${shopId}, ${k.locationId}, ${k.method}, ${k.date}, ${k.slot}, 0)
|
|
ON CONFLICT DO NOTHING`;
|
|
const rows = await tx.$queryRaw<{ count: number }[]>`
|
|
SELECT "count" FROM "CapacityCounter"
|
|
WHERE "shopId"=${shopId} AND "locationId"=${k.locationId} AND "method"=${k.method}
|
|
AND "date"=${k.date} AND "slot"=${k.slot}
|
|
FOR UPDATE`;
|
|
out.set(keyStr(k), Number(rows[0].count));
|
|
}
|
|
return out;
|
|
}
|
|
|
|
private async bump(tx: Tx, shopId: string, sel: Selection, delta: 1 | -1, counts?: Map<string, number>) {
|
|
for (const k of keysFor(sel)) {
|
|
await tx.$executeRaw`
|
|
UPDATE "CapacityCounter" SET "count" = GREATEST(0, "count" + ${delta})
|
|
WHERE "shopId"=${shopId} AND "locationId"=${k.locationId} AND "method"=${k.method}
|
|
AND "date"=${k.date} AND "slot"=${k.slot}`;
|
|
if (counts) counts.set(keyStr(k), Math.max(0, (counts.get(keyStr(k)) ?? 0) + delta));
|
|
}
|
|
}
|
|
|
|
/** Delete expired holds for the dates whose day rows we hold locks on. */
|
|
private async sweepLocked(tx: Tx, shopId: string, keys: Key[], now: Date, counts: Map<string, number>, keep?: string) {
|
|
const days = [...new Map(keys.filter((k) => k.slot === "*").map((k) => [keyStr(k), k])).values()];
|
|
for (const d of days) {
|
|
const expired = await tx.hold.findMany({
|
|
where: { shopId, locationId: d.locationId, method: d.method, date: d.date, expiresAt: { lte: now }, NOT: keep ? { token: keep } : undefined },
|
|
});
|
|
for (const h of expired) {
|
|
await tx.hold.delete({ where: { token: h.token } });
|
|
await this.ensureLocked(tx, shopId, keysFor(h as Selection), counts);
|
|
await this.bump(tx, shopId, h as Selection, -1, counts);
|
|
}
|
|
}
|
|
}
|
|
|
|
/** Slot rows for expired holds may not have been in the original lock set. */
|
|
private async ensureLocked(tx: Tx, shopId: string, keys: Key[], counts: Map<string, number>) {
|
|
const missing = keys.filter((k) => !counts.has(keyStr(k)));
|
|
if (missing.length) for (const [k, v] of await this.lock(tx, shopId, missing)) counts.set(k, v);
|
|
}
|
|
|
|
private fits(counts: Map<string, number>, sel: Selection, caps: Caps): null | "DAY_FULL" | "SLOT_FULL" {
|
|
const [day, slot] = keysFor(sel).map((k) => counts.get(keyStr(k)) ?? 0);
|
|
if (caps.dayCap != null && day >= caps.dayCap) return "DAY_FULL";
|
|
if (caps.slotCap != null && slot >= caps.slotCap) return "SLOT_FULL";
|
|
return null;
|
|
}
|
|
|
|
async hold(
|
|
shopId: string,
|
|
token: string,
|
|
sel: Selection,
|
|
caps: Caps,
|
|
now: Date,
|
|
holdMinutes: number,
|
|
fee = 0,
|
|
): Promise<HoldResult> {
|
|
const expiresAt = new Date(now.getTime() + holdMinutes * 60000);
|
|
return this.tx(async (tx) => {
|
|
const prev = await tx.hold.findUnique({ where: { token } });
|
|
if (prev && prev.shopId !== shopId) throw new Error("token belongs to another shop");
|
|
const keys = [...keysFor(sel), ...(prev ? keysFor(prev as Selection) : [])];
|
|
const counts = await this.lock(tx, shopId, keys);
|
|
await this.sweepLocked(tx, shopId, keys, now, counts, token);
|
|
|
|
const live = prev && prev.expiresAt > now;
|
|
if (prev && live && sameSel(prev as Selection, sel)) {
|
|
// Re-confirming the same place never shortens an existing (e.g. waitlist-offer) hold.
|
|
const until = prev.expiresAt > expiresAt ? prev.expiresAt : expiresAt;
|
|
await tx.hold.update({ where: { token }, data: { expiresAt: until, fee } });
|
|
return { ok: true as const, token, expiresAt: until };
|
|
}
|
|
if (prev) {
|
|
// Moving (or reviving) the hold: release the previous place first.
|
|
await this.bump(tx, shopId, prev as Selection, -1, counts);
|
|
await tx.hold.delete({ where: { token } });
|
|
}
|
|
const full = this.fits(counts, sel, caps);
|
|
if (full)
|
|
return {
|
|
ok: false as const,
|
|
code: full,
|
|
message: full === "DAY_FULL" ? "Fully booked for this day" : "This time slot was just taken",
|
|
};
|
|
await this.bump(tx, shopId, sel, 1, counts);
|
|
await tx.hold.create({ data: { token, shopId, ...sel, expiresAt, fee } });
|
|
return { ok: true as const, token, expiresAt };
|
|
});
|
|
}
|
|
|
|
async release(shopId: string, token: string): Promise<void> {
|
|
await this.tx(async (tx) => {
|
|
const h = await tx.hold.findUnique({ where: { token } });
|
|
if (!h || h.shopId !== shopId) return;
|
|
await this.lock(tx, shopId, keysFor(h as Selection));
|
|
await tx.hold.delete({ where: { token } });
|
|
await this.bump(tx, shopId, h as Selection, -1);
|
|
});
|
|
}
|
|
|
|
async confirm(
|
|
shopId: string,
|
|
orderId: string,
|
|
token: string | null,
|
|
fallback: (Selection & { method: Method }) | null,
|
|
caps: Caps,
|
|
now: Date,
|
|
details: BookingDetails,
|
|
): Promise<ConfirmResult> {
|
|
const existing = await this.db.booking.findUnique({ where: { shopId_orderId: { shopId, orderId } } });
|
|
if (existing)
|
|
return existing.status === "needs_review"
|
|
? { ok: false, bookingId: existing.id, code: "NEEDS_REVIEW", message: existing.reviewReason ?? "", duplicate: true }
|
|
: { ok: true, bookingId: existing.id, duplicate: true, reReserved: false };
|
|
|
|
try {
|
|
return await this.tx(async (tx) => {
|
|
const hold = token ? await tx.hold.findUnique({ where: { token } }) : null;
|
|
const sel: Selection | null = hold && hold.shopId === shopId ? (hold as Selection) : fallback;
|
|
const base = {
|
|
shopId,
|
|
orderId,
|
|
orderName: details.orderName,
|
|
timezone: details.timezone,
|
|
customerName: details.customerName,
|
|
customerEmail: details.customerEmail,
|
|
customerPhone: details.customerPhone,
|
|
address: details.address,
|
|
items: details.items,
|
|
fee: details.fee ?? 0,
|
|
peakDayId: details.peakDayId ?? null,
|
|
configVersion: details.configVersion,
|
|
snapshot: details.snapshot,
|
|
};
|
|
if (!sel) {
|
|
const b = await tx.booking.create({
|
|
data: { ...base, status: "needs_review", method: "pickup", reviewReason: "Order has no scheduling data" },
|
|
});
|
|
await tx.bookingEvent.create({ data: { shopId, bookingId: b.id, actor: "system", action: "flagged", after: { reason: "missing" } } });
|
|
return { ok: false as const, bookingId: b.id, code: "NEEDS_REVIEW" as const, message: "Order has no scheduling data", duplicate: false };
|
|
}
|
|
const keys = keysFor(sel);
|
|
const counts = await this.lock(tx, shopId, keys);
|
|
// Re-check under the lock: a concurrent duplicate webhook may have committed meanwhile.
|
|
const already = await tx.booking.findUnique({ where: { shopId_orderId: { shopId, orderId } } });
|
|
if (already) throw new DuplicateSignal();
|
|
const lockedHold = token ? await tx.hold.findUnique({ where: { token } }) : null;
|
|
const live = !!lockedHold && lockedHold.shopId === shopId && lockedHold.expiresAt > now && sameSel(lockedHold as Selection, sel);
|
|
// Sweep everything expired, including this order's own stale hold.
|
|
await this.sweepLocked(tx, shopId, keys, now, counts, live ? token! : undefined);
|
|
|
|
const [slotStart, slotEnd] = sel.slot.split("-");
|
|
const slotData = {
|
|
method: sel.method,
|
|
locationId: sel.locationId,
|
|
date: sel.date,
|
|
slotStart,
|
|
slotEnd,
|
|
startUtc: details.startUtc,
|
|
endUtc: details.endUtc,
|
|
};
|
|
let reReserved = false;
|
|
if (live) {
|
|
await tx.hold.delete({ where: { token: token! } }); // place converts from hold to booking; counter unchanged
|
|
} else {
|
|
const full = this.fits(counts, sel, caps);
|
|
if (full) {
|
|
const reason = token ? "Hold expired and the slot filled up" : "Selected slot was full when the order arrived";
|
|
const b = await tx.booking.create({ data: { ...base, ...slotData, status: "needs_review", reviewReason: reason } });
|
|
await tx.bookingEvent.create({ data: { shopId, bookingId: b.id, actor: "system", action: "flagged", after: { reason: full } } });
|
|
return { ok: false as const, bookingId: b.id, code: "NEEDS_REVIEW" as const, message: reason, duplicate: false };
|
|
}
|
|
await this.bump(tx, shopId, sel, 1, counts);
|
|
reReserved = true;
|
|
}
|
|
const b = await tx.booking.create({ data: { ...base, ...slotData, status: "confirmed" } });
|
|
await tx.bookingEvent.create({ data: { shopId, bookingId: b.id, actor: "system", action: "confirmed", after: { ...sel, reReserved } } });
|
|
return { ok: true as const, bookingId: b.id, duplicate: false, reReserved };
|
|
});
|
|
} catch (e: unknown) {
|
|
// Concurrent duplicate webhook lost the race on the unique index: whole tx rolled back.
|
|
if (e instanceof DuplicateSignal || isUniqueViolation(e)) {
|
|
const b = await this.db.booking.findUniqueOrThrow({ where: { shopId_orderId: { shopId, orderId } } });
|
|
return b.status === "needs_review"
|
|
? { ok: false, bookingId: b.id, code: "NEEDS_REVIEW", message: b.reviewReason ?? "", duplicate: true }
|
|
: { ok: true, bookingId: b.id, duplicate: true, reReserved: false };
|
|
}
|
|
throw e;
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Move a booking. Also used to resolve needs_review bookings (they hold no capacity yet).
|
|
* `force` lets a merchant knowingly exceed capacity (recorded in the audit log).
|
|
*/
|
|
async reschedule(
|
|
shopId: string,
|
|
bookingId: string,
|
|
sel: Selection,
|
|
caps: Caps,
|
|
now: Date,
|
|
actor: string,
|
|
extra: { startUtc?: Date; endUtc?: Date; force?: boolean } = {},
|
|
): Promise<{ ok: true; freed: Selection | null } | { ok: false; message: string }> {
|
|
return this.tx(async (tx) => {
|
|
const b = await tx.booking.findFirst({ where: { id: bookingId, shopId } });
|
|
if (!b || !["confirmed", "needs_review"].includes(b.status)) return { ok: false as const, message: "Booking can't be rescheduled" };
|
|
const old: Selection | null =
|
|
b.status === "confirmed" && b.locationId && b.date && b.slotStart
|
|
? { locationId: b.locationId, method: b.method as Method, date: b.date, slot: `${b.slotStart}-${b.slotEnd}` }
|
|
: null;
|
|
if (old && sameSel(old, sel)) return { ok: true as const, freed: null };
|
|
const keys = [...keysFor(sel), ...(old ? keysFor(old) : [])];
|
|
const counts = await this.lock(tx, shopId, keys);
|
|
await this.sweepLocked(tx, shopId, keys, now, counts);
|
|
// Evaluate the new place as if the old one were already released.
|
|
if (old) await this.bump(tx, shopId, old, -1, counts);
|
|
const full = this.fits(counts, sel, caps);
|
|
if (full && !extra.force) throw new RollbackSignal(full === "DAY_FULL" ? "That day is fully booked" : "That slot is full");
|
|
await this.bump(tx, shopId, sel, 1, counts);
|
|
const [slotStart, slotEnd] = sel.slot.split("-");
|
|
await tx.booking.update({
|
|
where: { id: b.id },
|
|
data: {
|
|
locationId: sel.locationId,
|
|
method: sel.method,
|
|
date: sel.date,
|
|
slotStart,
|
|
slotEnd,
|
|
startUtc: extra.startUtc,
|
|
endUtc: extra.endUtc,
|
|
status: "confirmed",
|
|
reviewReason: null,
|
|
rescheduleCount: actor === "customer" ? { increment: 1 } : undefined,
|
|
},
|
|
});
|
|
await tx.bookingEvent.create({
|
|
data: {
|
|
shopId,
|
|
bookingId: b.id,
|
|
actor,
|
|
action: b.status === "needs_review" ? "resolved" : "rescheduled",
|
|
before: old ? { ...old } : { status: b.status },
|
|
after: { ...sel, forced: !!(full && extra.force) },
|
|
},
|
|
});
|
|
return { ok: true as const, freed: old };
|
|
}).catch((e) => {
|
|
if (e instanceof RollbackSignal) return { ok: false as const, message: e.message };
|
|
throw e;
|
|
});
|
|
}
|
|
|
|
async cancel(shopId: string, orderId: string, actor = "system"): Promise<{ freed: Selection | null; bookingId: string | null }> {
|
|
return this.tx(async (tx) => {
|
|
const b = await tx.booking.findUnique({ where: { shopId_orderId: { shopId, orderId } } });
|
|
if (!b || b.status === "cancelled") return { freed: null, bookingId: b?.id ?? null };
|
|
let freed: Selection | null = null;
|
|
if (b.status === "confirmed" && b.locationId && b.date && b.slotStart) {
|
|
freed = { locationId: b.locationId, method: b.method as Method, date: b.date, slot: `${b.slotStart}-${b.slotEnd}` };
|
|
await this.lock(tx, shopId, keysFor(freed));
|
|
await this.bump(tx, shopId, freed, -1);
|
|
}
|
|
await tx.booking.update({ where: { id: b.id }, data: { status: "cancelled" } });
|
|
await tx.bookingEvent.create({ data: { shopId, bookingId: b.id, actor, action: "cancelled", before: { status: b.status } } });
|
|
await tx.reminderJob.updateMany({ where: { bookingId: b.id, status: "pending" }, data: { status: "cancelled" } });
|
|
return { freed, bookingId: b.id };
|
|
});
|
|
}
|
|
|
|
/** Usage snapshot for the engine. Excludes expired-but-unswept holds and, optionally, the caller's own hold. */
|
|
async usage(shopId: string, fromDate: string, toDate: string, now: Date, excludeToken?: string): Promise<UsageSnapshot> {
|
|
const [counters, stale] = await Promise.all([
|
|
this.db.capacityCounter.findMany({ where: { shopId, date: { gte: fromDate, lte: toDate } } }),
|
|
this.db.hold.findMany({
|
|
where: {
|
|
shopId,
|
|
date: { gte: fromDate, lte: toDate },
|
|
OR: [{ expiresAt: { lte: now } }, ...(excludeToken ? [{ token: excludeToken }] : [])],
|
|
},
|
|
}),
|
|
]);
|
|
const out: UsageSnapshot = {};
|
|
for (const c of counters) {
|
|
const k = usageKey(c.locationId, c.method as Method, c.date);
|
|
out[k] ??= { day: 0, slots: {} };
|
|
if (c.slot === "*") out[k].day += c.count;
|
|
else out[k].slots[c.slot] = (out[k].slots[c.slot] ?? 0) + c.count;
|
|
}
|
|
for (const h of stale) {
|
|
const k = usageKey(h.locationId, h.method as Method, h.date);
|
|
if (!out[k]) continue;
|
|
out[k].day = Math.max(0, out[k].day - 1);
|
|
out[k].slots[h.slot] = Math.max(0, (out[k].slots[h.slot] ?? 0) - 1);
|
|
}
|
|
return out;
|
|
}
|
|
|
|
/** Background sweep (worker, every minute). Returns number of holds released. */
|
|
async sweepExpired(now: Date, limit = 500, shopId?: string): Promise<number> {
|
|
const expired = await this.db.hold.findMany({ where: { expiresAt: { lte: now }, ...(shopId ? { shopId } : {}) }, take: limit, orderBy: { expiresAt: "asc" } });
|
|
const groups = new Map<string, { shopId: string; key: Key }>();
|
|
for (const h of expired)
|
|
groups.set(`${h.shopId}|${h.locationId}|${h.method}|${h.date}`, {
|
|
shopId: h.shopId,
|
|
key: { locationId: h.locationId, method: h.method, date: h.date, slot: "*" },
|
|
});
|
|
let n = 0;
|
|
for (const { shopId, key } of groups.values()) {
|
|
n += await this.tx(async (tx) => {
|
|
const counts = await this.lock(tx, shopId, [key]);
|
|
const before = await tx.hold.count({ where: { shopId, locationId: key.locationId, method: key.method, date: key.date, expiresAt: { lte: now } } });
|
|
await this.sweepLocked(tx, shopId, [key], now, counts);
|
|
return before;
|
|
});
|
|
}
|
|
return n;
|
|
}
|
|
}
|
|
|
|
class RollbackSignal extends Error {}
|
|
class DuplicateSignal extends Error {}
|
|
|
|
function isUniqueViolation(e: unknown): boolean {
|
|
const err = e as { code?: string; message?: string; meta?: { code?: string } };
|
|
return err?.code === "P2002" || err?.meta?.code === "23505" || /unique constraint|23505/i.test(err?.message ?? "");
|
|
}
|