import { and, asc, eq, inArray } from 'drizzle-orm' import type { NodePgDatabase } from 'drizzle-orm/node-postgres' import * as schema from '#/db/schema.ts' import { PaymentDeclinedError } from '#/lib/payment.ts' import type { PaymentProvider } from '#/lib/payment.ts' type Db = NodePgDatabase // The callback type `db.transaction()` hands its work to. Helpers below // accept either a top-level Db or an in-flight transaction so they compose // into one atomic unit of work when called from within `db.transaction()`. type Tx = Parameters[0]>[0] type DbOrTx = Db | Tx // Drizzle types a `SELECT ... LIMIT 1`-shaped destructure's element as always // present, even though zero rows is a real possibility at runtime (without // noUncheckedIndexedAccess). Route those reads through this helper so their // optionality is genuinely reflected in the type. const firstRow = (rows: Array): T | undefined => rows[0] export class OrderItemNotCancellableError extends Error { constructor(reason: string) { super(`This item can no longer be cancelled: ${reason}`) this.name = 'OrderItemNotCancellableError' } } export class OrderItemNotReadyError extends Error { constructor() { super('This item is not ready for pickup yet') this.name = 'OrderItemNotReadyError' } } export class NoStepInProgressError extends Error { constructor() { super('This item has no Preparation Step currently in progress') this.name = 'NoStepInProgressError' } } export class ForbiddenError extends Error { constructor() { super('You do not have access to this Order') this.name = 'ForbiddenError' } } export type MenuItemWithRecipe = { id: number name: string description: string | null priceCents: number steps: Array<{ sequence: number machineTag: string estimatedDurationSeconds: number }> } export const getMenu = async ( db: Db, brandId: number, ): Promise> => { const items = await db .select() .from(schema.menuItems) .where(eq(schema.menuItems.brandId, brandId)) const steps = await db .select({ menuItemId: schema.menuItemSteps.menuItemId, sequence: schema.menuItemSteps.sequence, machineTag: schema.machineTags.name, estimatedDurationSeconds: schema.menuItemSteps.estimatedDurationSeconds, }) .from(schema.menuItemSteps) .innerJoin( schema.machineTags, eq(schema.menuItemSteps.machineTagId, schema.machineTags.id), ) .where( inArray( schema.menuItemSteps.menuItemId, items.map((item) => item.id), ), ) .orderBy(asc(schema.menuItemSteps.sequence)) return items.map((item) => ({ id: item.id, name: item.name, description: item.description, priceCents: item.priceCents, steps: steps .filter((step) => step.menuItemId === item.id) .map(({ sequence, machineTag, estimatedDurationSeconds }) => ({ sequence, machineTag, estimatedDurationSeconds, })), })) } // Assigns queued Preparation Steps for a given Machine Tag to whichever // Machine carrying that Tag is idle, FIFO by queue position. Called // whenever a Step is created (order placed) or freed (a Step finishes) — // see docs/adr/0003-machine-tag-pool-dynamic-assignment.md. // // Locks that Tag's Machine rows for the duration of the caller's // transaction, so two concurrent calls for the same Tag (e.g. two orders // placed at once) can't both read the same idle Machine as free and // double-book it — the second call blocks here until the first commits. export const assignQueuedStepsForTag = async ( db: DbOrTx, machineTagId: number, ) => { const tagMachines = await db .select({ id: schema.machines.id }) .from(schema.machines) .where(eq(schema.machines.machineTagId, machineTagId)) .for('update') if (tagMachines.length === 0) return const busyMachineIds = new Set( ( await db .select({ machineId: schema.preparationStepInstances.machineId }) .from(schema.preparationStepInstances) .where( and( eq(schema.preparationStepInstances.machineTagId, machineTagId), eq(schema.preparationStepInstances.status, 'in_progress'), ), ) ) .map((row) => row.machineId) .filter((id): id is number => id !== null), ) const idleMachineIds = tagMachines .map((machine) => machine.id) .filter((id) => !busyMachineIds.has(id)) if (idleMachineIds.length === 0) return const queuedSteps = await db .select({ id: schema.preparationStepInstances.id }) .from(schema.preparationStepInstances) .where( and( eq(schema.preparationStepInstances.machineTagId, machineTagId), eq(schema.preparationStepInstances.status, 'queued'), ), ) .orderBy(asc(schema.preparationStepInstances.id)) const assignments = queuedSteps.slice(0, idleMachineIds.length) for (const [index, step] of assignments.entries()) { await db .update(schema.preparationStepInstances) .set({ status: 'in_progress', machineId: idleMachineIds[index], startedAt: new Date(), }) .where(eq(schema.preparationStepInstances.id, step.id)) } } // Copies the Menu Item's live Recipe into orderItemSteps for this Order // Item, then creates its first Preparation Step instance from that copy — // so a later edit to the live Recipe never reshapes an Order Item already // in flight (see docs/adr/0005-recipe-snapshot-per-order-item.md). const initializeOrderItemRecipe = async ( db: DbOrTx, orderItemId: number, menuItemId: number, ) => { const liveSteps = await db .select() .from(schema.menuItemSteps) .where(eq(schema.menuItemSteps.menuItemId, menuItemId)) .orderBy(asc(schema.menuItemSteps.sequence)) // A menu item with no Preparation Steps at all is a data integrity bug // (see spec: every menu item's Recipe has at least one Step), not a valid // "nothing to do" case — surfacing it loudly here beats leaving the Order // Item permanently stuck with no Step and no way to progress or cancel. if (liveSteps.length === 0) { throw new Error(`Menu item ${menuItemId} has no Preparation Steps defined`) } const recipeSteps = await db .insert(schema.orderItemSteps) .values( liveSteps.map((step) => ({ orderItemId, sequence: step.sequence, machineTagId: step.machineTagId, estimatedDurationSeconds: step.estimatedDurationSeconds, })), ) .returning() const firstStep = recipeSteps.find((step) => step.sequence === 1) if (!firstStep) { throw new Error( `Menu item ${menuItemId} has no Preparation Step with sequence 1`, ) } const [instance] = await db .insert(schema.preparationStepInstances) .values({ orderItemId, sequence: firstStep.sequence, machineTagId: firstStep.machineTagId, estimatedDurationSeconds: firstStep.estimatedDurationSeconds, status: 'queued', }) .returning() return instance } export type PlaceOrderInput = { items: Array<{ menuItemId: number }> customer: | { type: 'guest'; name: string; contact: string } | { type: 'account'; customerId: string } } export const placeOrder = async ( db: Db, paymentProvider: PaymentProvider, input: PlaceOrderInput, ) => { if (input.items.length === 0) { throw new Error('An Order must contain at least one Order Item') } const menuItemIds = input.items.map((item) => item.menuItemId) const menuItemRows = await db .select() .from(schema.menuItems) .where(inArray(schema.menuItems.id, menuItemIds)) const menuItemById = new Map(menuItemRows.map((row) => [row.id, row])) for (const id of menuItemIds) { if (!menuItemById.has(id)) { throw new Error(`Unknown menu item: ${id}`) } } // Catch the "menu item has no Recipe Steps" data bug before charging // anyone, rather than charging first and refunding once // initializeOrderItemRecipe discovers it mid-transaction. const itemIdsWithSteps = new Set( ( await db .select({ menuItemId: schema.menuItemSteps.menuItemId }) .from(schema.menuItemSteps) .where(inArray(schema.menuItemSteps.menuItemId, menuItemIds)) ).map((row) => row.menuItemId), ) for (const id of menuItemIds) { if (!itemIdsWithSteps.has(id)) { throw new Error(`Menu item ${id} has no Preparation Steps defined`) } } const totalCents = input.items.reduce( (sum, item) => sum + menuItemById.get(item.menuItemId)!.priceCents, 0, ) // Payment happens before anything is queued: a declined charge means no // Order Item is ever created (see spec: "Order rejected if payment fails"). let charge try { charge = await paymentProvider.charge(totalCents) } catch (error) { if (error instanceof PaymentDeclinedError) throw error throw new PaymentDeclinedError() } try { return await db.transaction(async (tx) => { const [order] = await tx .insert(schema.orders) .values({ customerId: input.customer.type === 'account' ? input.customer.customerId : null, guestName: input.customer.type === 'guest' ? input.customer.name : null, guestContact: input.customer.type === 'guest' ? input.customer.contact : null, guestAccessToken: input.customer.type === 'guest' ? crypto.randomUUID() : null, status: 'paid', paymentChargeId: charge.chargeId, }) .returning() const tagsToAssign = new Set() for (const item of input.items) { const menuItem = menuItemById.get(item.menuItemId)! const [orderItem] = await tx .insert(schema.orderItems) .values({ orderId: order.id, menuItemId: menuItem.id, priceCents: menuItem.priceCents, status: 'paid', }) .returning() const step = await initializeOrderItemRecipe( tx, orderItem.id, menuItem.id, ) tagsToAssign.add(step.machineTagId) } for (const tagId of tagsToAssign) { await assignQueuedStepsForTag(tx, tagId) } return order }) } catch (error) { // The charge already happened; a failed transaction must not leave the // customer paying for an Order that doesn't exist. If the compensating // refund itself fails, the transaction failure is still the real cause — // log the refund failure rather than letting it hide that error. try { await paymentProvider.refund(charge.chargeId, totalCents) } catch (refundError) { console.error( `Failed to refund charge ${charge.chargeId} after a failed order transaction:`, refundError, ) } throw error } } export type OrderAccess = { customerId?: string; guestAccessToken?: string } const assertOrderAccess = ( order: Pick< typeof schema.orders.$inferSelect, 'customerId' | 'guestAccessToken' >, access: OrderAccess, ) => { if (order.customerId) { if (access.customerId === order.customerId) return } else if (order.guestAccessToken) { if (access.guestAccessToken === order.guestAccessToken) return } throw new ForbiddenError() } // Steps are created one at a time (see createFirstStepForOrderItem / // advanceOrderItemStep), so at most one non-done instance ever exists for a // given Order Item at once. const getCurrentStepInstance = async (db: DbOrTx, orderItemId: number) => { const instances = await db .select() .from(schema.preparationStepInstances) .where(eq(schema.preparationStepInstances.orderItemId, orderItemId)) .orderBy(asc(schema.preparationStepInstances.sequence)) return instances.find((row) => row.status !== 'done') ?? null } const updateOrderCompleteIfEverySiblingIsDone = async ( db: DbOrTx, orderId: number, ) => { const siblingItems = await db .select({ status: schema.orderItems.status }) .from(schema.orderItems) .where(eq(schema.orderItems.orderId, orderId)) const orderComplete = siblingItems.every( (item) => item.status === 'picked_up' || item.status === 'cancelled', ) if (orderComplete) { await db .update(schema.orders) .set({ status: 'complete' }) .where(eq(schema.orders.id, orderId)) } } export const cancelOrderItem = async ( db: Db, paymentProvider: PaymentProvider, orderItemId: number, access: OrderAccess, ) => { const orderItem = firstRow( await db .select() .from(schema.orderItems) .where(eq(schema.orderItems.id, orderItemId)), ) if (!orderItem) throw new Error('Order item not found') const order = firstRow( await db .select() .from(schema.orders) .where(eq(schema.orders.id, orderItem.orderId)), ) if (!order) throw new Error('Order not found') assertOrderAccess(order, access) if (orderItem.status === 'cancelled') { throw new OrderItemNotCancellableError('it has already been cancelled') } if (orderItem.status === 'ready' || orderItem.status === 'picked_up') { throw new OrderItemNotCancellableError('it is already done cooking') } const currentStep = await getCurrentStepInstance(db, orderItemId) if (!currentStep || currentStep.status !== 'queued') { throw new OrderItemNotCancellableError('it has already started cooking') } // Refund before touching any state, mirroring placeOrder's charge-first // pattern: if the refund itself fails, nothing about this cancellation is // ever recorded, exactly like a declined charge leaves no Order behind. if (order.paymentChargeId) { await paymentProvider.refund(order.paymentChargeId, orderItem.priceCents) } try { await db.transaction(async (tx) => { // Re-check status = 'queued' as part of the delete itself, atomically: // a Machine may have claimed this Step (assignQueuedStepsForTag), or a // concurrent cancelOrderItem call may have already deleted it, in the // time since the read above (which can include the refund call, an // external network round trip). If so, zero rows match and the catch // below re-charges to undo the refund we already issued. const deletedSteps = await tx .delete(schema.preparationStepInstances) .where( and( eq(schema.preparationStepInstances.id, currentStep.id), eq(schema.preparationStepInstances.status, 'queued'), ), ) .returning({ id: schema.preparationStepInstances.id }) if (deletedSteps.length === 0) { throw new OrderItemNotCancellableError( 'it has already started cooking', ) } await tx .update(schema.orderItems) .set({ status: 'cancelled' }) .where(eq(schema.orderItems.id, orderItemId)) await updateOrderCompleteIfEverySiblingIsDone(tx, orderItem.orderId) }) } catch (error) { // The refund already happened; an aborted cancellation must not leave // the customer refunded for an item that's still cooking. If the // compensating re-charge itself fails, the cancellation failure is // still the real cause — log the re-charge failure rather than letting // it hide that error. if (order.paymentChargeId) { try { await paymentProvider.charge(orderItem.priceCents) } catch (rechargeError) { console.error( `Failed to re-charge ${orderItem.priceCents} cents for order item ${orderItemId} after an aborted cancellation:`, rechargeError, ) } } throw error } } export const advanceOrderItemStep = async (db: Db, orderItemId: number) => { await db.transaction(async (tx) => { const currentStep = await getCurrentStepInstance(tx, orderItemId) if (!currentStep || currentStep.status !== 'in_progress') { throw new NoStepInProgressError() } await tx .update(schema.preparationStepInstances) .set({ status: 'done', doneAt: new Date() }) .where(eq(schema.preparationStepInstances.id, currentStep.id)) const nextStepDefinition = firstRow( await tx .select() .from(schema.orderItemSteps) .where( and( eq(schema.orderItemSteps.orderItemId, orderItemId), eq(schema.orderItemSteps.sequence, currentStep.sequence + 1), ), ), ) if (nextStepDefinition) { await tx.insert(schema.preparationStepInstances).values({ orderItemId, sequence: nextStepDefinition.sequence, machineTagId: nextStepDefinition.machineTagId, estimatedDurationSeconds: nextStepDefinition.estimatedDurationSeconds, status: 'queued', }) await assignQueuedStepsForTag(tx, nextStepDefinition.machineTagId) } else { await tx .update(schema.orderItems) .set({ status: 'ready' }) .where(eq(schema.orderItems.id, orderItemId)) } // Freeing this machine may unblock another queued Step for the same Tag. await assignQueuedStepsForTag(tx, currentStep.machineTagId) }) } export const markOrderItemPickedUp = async ( db: Db, orderItemId: number, access: OrderAccess, ) => { await db.transaction(async (tx) => { const orderItem = firstRow( await tx .select() .from(schema.orderItems) .where(eq(schema.orderItems.id, orderItemId)), ) if (!orderItem) throw new Error('Order item not found') const order = firstRow( await tx .select() .from(schema.orders) .where(eq(schema.orders.id, orderItem.orderId)), ) if (!order) throw new Error('Order not found') assertOrderAccess(order, access) if (orderItem.status !== 'ready') throw new OrderItemNotReadyError() await tx .update(schema.orderItems) .set({ status: 'picked_up' }) .where(eq(schema.orderItems.id, orderItemId)) await updateOrderCompleteIfEverySiblingIsDone(tx, orderItem.orderId) }) } export type OrderItemStatusView = { id: number menuItemName: string status: schema.OrderItemStatus cancellable: boolean steps: Array<{ sequence: number machineTag: string estimatedDurationSeconds: number status: 'done' | 'in_progress' | 'queued' | 'pending' queuePosition: number | null }> etaSeconds: number | null } export const getOrderStatus = async ( db: Db, orderId: number, access: OrderAccess, ) => { const order = firstRow( await db.select().from(schema.orders).where(eq(schema.orders.id, orderId)), ) if (!order) throw new Error('Order not found') assertOrderAccess(order, access) const items = await db .select() .from(schema.orderItems) .where(eq(schema.orderItems.orderId, orderId)) const menuItemIds = [...new Set(items.map((item) => item.menuItemId))] const orderItemIds = items.map((item) => item.id) const menuItemsById = new Map( ( await db .select() .from(schema.menuItems) .where(inArray(schema.menuItems.id, menuItemIds)) ).map((row) => [row.id, row]), ) const recipeRows = orderItemIds.length === 0 ? [] : await db .select({ orderItemId: schema.orderItemSteps.orderItemId, sequence: schema.orderItemSteps.sequence, machineTag: schema.machineTags.name, machineTagId: schema.orderItemSteps.machineTagId, estimatedDurationSeconds: schema.orderItemSteps.estimatedDurationSeconds, }) .from(schema.orderItemSteps) .innerJoin( schema.machineTags, eq(schema.orderItemSteps.machineTagId, schema.machineTags.id), ) .where(inArray(schema.orderItemSteps.orderItemId, orderItemIds)) .orderBy(asc(schema.orderItemSteps.sequence)) const allInstances = orderItemIds.length === 0 ? [] : await db .select() .from(schema.preparationStepInstances) .where( inArray(schema.preparationStepInstances.orderItemId, orderItemIds), ) const instancesByOrderItem = new Map< number, Array<(typeof allInstances)[number]> >() for (const instance of allInstances) { const list = instancesByOrderItem.get(instance.orderItemId) ?? [] list.push(instance) instancesByOrderItem.set(instance.orderItemId, list) } const queuedTagIds = [ ...new Set( allInstances .filter((instance) => instance.status === 'queued') .map((instance) => instance.machineTagId), ), ] const queuedStepsByTag = new Map>() const machineCountByTag = new Map() for (const tagId of queuedTagIds) { queuedStepsByTag.set( tagId, await db .select({ id: schema.preparationStepInstances.id }) .from(schema.preparationStepInstances) .where( and( eq(schema.preparationStepInstances.machineTagId, tagId), eq(schema.preparationStepInstances.status, 'queued'), ), ) .orderBy(asc(schema.preparationStepInstances.id)), ) machineCountByTag.set( tagId, ( await db .select({ id: schema.machines.id }) .from(schema.machines) .where(eq(schema.machines.machineTagId, tagId)) ).length, ) } const itemViews: Array = items.map((item) => { const menuItem = menuItemsById.get(item.menuItemId) const recipe = recipeRows.filter((step) => step.orderItemId === item.id) const instances = instancesByOrderItem.get(item.id) ?? [] const instanceBySequence = new Map( instances.map((row) => [row.sequence, row]), ) let etaSeconds: number | null = 0 const steps = recipe.map((step) => { const instance = instanceBySequence.get(step.sequence) if (!instance) { if (etaSeconds !== null) etaSeconds += step.estimatedDurationSeconds return { sequence: step.sequence, machineTag: step.machineTag, estimatedDurationSeconds: step.estimatedDurationSeconds, status: 'pending' as const, queuePosition: null, } } if (instance.status === 'done') { return { sequence: step.sequence, machineTag: step.machineTag, estimatedDurationSeconds: step.estimatedDurationSeconds, status: 'done' as const, queuePosition: null, } } if (instance.status === 'in_progress') { const elapsedSeconds = instance.startedAt ? Math.floor((Date.now() - instance.startedAt.getTime()) / 1000) : 0 if (etaSeconds !== null) { etaSeconds += Math.max( step.estimatedDurationSeconds - elapsedSeconds, 0, ) } return { sequence: step.sequence, machineTag: step.machineTag, estimatedDurationSeconds: step.estimatedDurationSeconds, status: 'in_progress' as const, queuePosition: 0, } } // Queued: rank among queued steps for the same Tag (batched above), // then spread across that Tag's Machines for a rough ETA. const queuedForTag = queuedStepsByTag.get(step.machineTagId) ?? [] const position = queuedForTag.findIndex((row) => row.id === instance.id) + 1 const machineCount = Math.max( machineCountByTag.get(step.machineTagId) ?? 0, 1, ) if (etaSeconds !== null) { const waitSeconds = Math.floor(Math.max(position - 1, 0) / machineCount) * step.estimatedDurationSeconds etaSeconds += waitSeconds + step.estimatedDurationSeconds } return { sequence: step.sequence, machineTag: step.machineTag, estimatedDurationSeconds: step.estimatedDurationSeconds, status: 'queued' as const, queuePosition: position, } }) if ( item.status === 'ready' || item.status === 'picked_up' || item.status === 'cancelled' ) { etaSeconds = null } const currentInstance = instances.find((row) => row.status !== 'done') return { id: item.id, menuItemName: menuItem?.name ?? '', status: item.status, cancellable: item.status !== 'cancelled' && item.status !== 'ready' && item.status !== 'picked_up' && currentInstance?.status === 'queued', steps, etaSeconds, } }) return { id: order.id, status: order.status, items: itemViews, } }