apps/backend/src/modules/kitchen/workers/kitchen-routing.worker.tsTypeScript
import { Injectable } from '@nestjs/common';
import { randomUUID } from 'node:crypto';
import { DatabaseService } from '../../../infrastructure/database/database.service';
@Injectable()
export class KitchenRoutingWorker {
constructor(private readonly database: DatabaseService) {}
async routeSubmittedOrder(orderId: string): Promise<number> {
const rows = await this.database.query<any>(
`select
o.organization_id,
o.location_id,
o.service_session_id,
oi.id as order_item_id,
oi.product_id,
oi.product_name_snapshot,
oi.quantity::text,
oi.notes,
s.seat_number,
oc.course_number,
coalesce(
(
select jsonb_agg(jsonb_build_object(
'name', oim.modifier_name_snapshot,
'priceDeltaMinor', oim.price_delta_minor_snapshot
))
from order_item_modifiers oim
where oim.order_item_id=oi.id
),
'[]'::jsonb
) as modifiers_snapshot,
coalesce(oiss.product_allergens_snapshot, '[]'::jsonb) as allergens_snapshot
from orders o
join order_items oi on oi.order_id=o.id and oi.status in ('active','remake')
left join seats s on s.id=oi.seat_id
left join order_courses oc on oc.id=oi.course_id
left join order_item_safety_snapshots oiss on oiss.order_item_id=oi.id
where o.id=$1 and o.status='submitted'`,
[orderId],
);
let created = 0;
for (const item of rows) {
const routes = await this.database.query<{ kitchen_station_id: string; priority: number }>(
`select pkr.kitchen_station_id, pkr.priority
from product_kitchen_routes pkr
join kitchen_stations ks on ks.id=pkr.kitchen_station_id
where pkr.product_id=$1
and ks.location_id=$2
and ks.organization_id=$3
and ks.status='active'
order by pkr.priority desc, ks.sort_order`,
[item.product_id, item.location_id, item.organization_id],
);
for (const route of routes) {
const createdTicket = await this.database.transaction(async (tx) => {
// The schema intentionally keeps ticket identity flexible for future split-ticket flows.
// Serialize the read/create pair so concurrent route commands cannot create two active
// tickets for the same order, station and course.
await tx.query(
`select pg_advisory_xact_lock(hashtextextended($1,0))`,
[`kitchen-ticket:${orderId}:${route.kitchen_station_id}:${item.course_number ?? 'null'}`],
);
const ticket = await tx.query<{ id: string }>(
`select id
from kitchen_tickets
where order_id=$1
and station_id=$2
and course_number is not distinct from $3
and status in ('queued','held','fired','in_progress')
limit 1`,
[orderId, route.kitchen_station_id, item.course_number],
);
let ticketId = ticket[0]?.id;
let created = false;
if (!ticketId) {
ticketId = randomUUID();
await tx.query(
`insert into kitchen_tickets(
id, organization_id, location_id, station_id,
order_id, service_session_id, course_number,
status, priority, fired_at
)
values (
$1,$2,$3,$4,$5,$6,$7,
case when $7 is null then 'fired' else 'held' end,
$8,
case when $7 is null then now() else null end
)`,
[
ticketId,
item.organization_id,
item.location_id,
route.kitchen_station_id,
orderId,
item.service_session_id,
item.course_number,
route.priority,
],
);
created = true;
}
await tx.query(
`insert into kitchen_ticket_items(
id, kitchen_ticket_id, order_item_id, product_name_snapshot,
quantity, seat_number, notes, modifiers_snapshot, allergens_snapshot
)
values ($1,$2,$3,$4,$5,$6,$7,$8::jsonb,$9::jsonb)
on conflict(kitchen_ticket_id, order_item_id) do nothing`,
[
randomUUID(),
ticketId,
item.order_item_id,
item.product_name_snapshot,
item.quantity,
item.seat_number,
item.notes,
JSON.stringify(item.modifiers_snapshot ?? []),
JSON.stringify(item.allergens_snapshot ?? []),
],
);
return created;
});
if (createdTicket) created++;
}
}
return created;
}
async fireCourse(orderId: string, courseNumber: number): Promise<number> {
const tickets = await this.database.query<{ id: string }>(
`update kitchen_tickets
set status='fired', fired_at=coalesce(fired_at, now()), version=version+1
where order_id=$1
and course_number=$2
and status='held'
returning id`,
[orderId, courseNumber],
);
return tickets.length;
}
}