import { BadRequestException, Injectable, Logger, NotFoundException, OnModuleInit, Optional, } from '@nestjs/common'; import { InjectDataSource } from '@nestjs/typeorm'; import { Cron, SchedulerRegistry } from '@nestjs/schedule'; import { DataSource } from 'typeorm'; import { Booking } from '../bookings/entities/booking.entity'; import { BookingsRepository } from '../bookings/bookings.repository'; import { Locomotive } from '../locomotives/entities/locomotive.entity'; import { formatRouteLabel } from '../routes/entities/route.entity'; import { TrainSchedule } from '../train-schedules/entities/train-schedule.entity'; import { TrainScheduleBooking } from '../train-schedules/entities/train-schedule-booking.entity'; import { TrainSchedulesRepository } from '../train-schedules/train-schedules.repository'; import { TrainScheduleBookingsRepository } from '../train-schedules/train-schedule-bookings.repository'; import { TrainSchedulingGlobalRules } from './entities/train-scheduling-global-rules.entity'; import { BookingNotifierService } from './booking-notifier.service'; import { TrainSchedulingService } from './train-scheduling.service'; import { eatDay, groupBookingsIntoBoardWindows } from './batch-window.util'; import { Freight } from "@edr/types"; import { BillingService } from "../billing/billing.service"; import { BATCH_CRON, BATCH_TIMEZONE, DEFAULT_BULK_WAGON_LENGTH_METERS, DEFAULT_CONTAINER_WAGON_LENGTH_METERS, DEFAULT_WAGONS_PER_BOOKING, PAYMENT_WINDOW_MS, } from "./booking-batch.constants"; import { bookingTrainLengthMeters, deriveTrainCapacityFromLocomotive, wagonTypeDimensionsFromEntity, } from './train-capacity.util'; import { WagonType } from '../wagon-types/entities/wagon-type.entity'; import { ClearanceMilestoneService } from '../contracts/clearance-milestone.service'; /** A train's remaining capacity along the three physical limits the batch enforces. */ interface Capacity { wagons: number; weightTons: number; lengthMeters: number; } /** A day-level pool key: all trains on this route departing on this EAT day. */ interface RouteDayGroup { originYardId: string; destinationYardId: string; /** EAT calendar day, `yyyy-MM-dd`. */ day: string; } type WagonLengths = { container: number; bulk: number }; export type BatchBoardBookingState = | "ALLOCATED" | "SELECTED_FOR_BATCH" | "READY" | "WAITING" | "PENDING_CONTRACT" | "EXPIRED"; export interface BatchBoardBooking { id: string; reference: string; company: string; isGovernment: boolean; wagons: number; weightTons: number; lengthMeters: number; paymentDeadline: string | null; state: BatchBoardBookingState; } export type BookingAllocationStatus = | "NOT_ATTEMPTED" | "ASSIGNED" | "DEFERRED" | "FAILED"; export interface BatchBoardBookingDetail extends BatchBoardBooking { fullyExecutedAt: string | null; selectedForBatchAt: string | null; allocationStatus: BookingAllocationStatus; allocationIssue: string | null; } export interface BatchWindowGroup { key: string; label: string; /** EAT calendar day as ISO `YYYY-MM-DD` (empty for the pending-contract bucket). */ date: string; /** Human label for the day, e.g. `Thu, 05 Jun` (empty for pending-contract). */ dateLabel: string; start: string; end: string; counts: { allocated: number; selectedForBatch: number; ready: number; waiting: number; expired: number; pendingContract: number; }; bookings: BatchBoardBookingDetail[]; } export interface BatchBoardScheduleDetail { scheduleId: string; trainNumber: string | null; routeName: string | null; origin: string | null; destination: string | null; scheduleDate: string | null; status: string; bookingWindowStatus: string; locomotive: BatchBoardSchedule["locomotive"]; capacity: BatchBoardSchedule["capacity"]; counts: BatchBoardSchedule["counts"]; windows: BatchWindowGroup[]; pendingContract: BatchWindowGroup; allocationViolations: string[]; } export interface BatchBoardSchedule { scheduleId: string; trainNumber: string | null; routeName: string | null; origin: string | null; destination: string | null; scheduleDate: string | null; status: string; bookingWindowStatus: string; locomotive: { code: string; name: string | null; maxPullWeightTons: number; maxTrainLengthMeters: number; } | null; capacity: { /** Wagons on bookings already linked to the train (ALLOCATED only). */ allocatedWagons: number; /** Train length used by allocated bookings (from wagon-type dimensions). */ allocatedLengthMeters: number; maxLengthMeters: number | null; /** Weight committed on the train (allocated + selected-for-batch). */ usedWeightTons: number; maxWeightTons: number | null; }; counts: { allocated: number; selectedForBatch: number; ready: number; waiting: number; pendingContract: number; expired: number; }; bookings: BatchBoardBooking[]; } /** * Demand-batching engine: every 3h (EAT) it ranks each OPEN schedule's ready pool * by priority, greedily fills the train to capacity (skipping bookings that don't fit), * reserves a 1h pay window for commercial customers (government allocated unpaid, * preempting lower-priority commercial if needed), then settles each batch 1h later — * allocating those who paid and expiring those who didn't, topping up from the waiting list. * Capacity is bounded on three axes at once: wagon count (`schedule.maxWagons`), the * locomotive's max pull weight, and its max train length (also capped by global rules). */ @Injectable() export class BookingBatchService implements OnModuleInit { private readonly logger = new Logger(BookingBatchService.name); constructor( @InjectDataSource() private readonly dataSource: DataSource, private readonly bookingsRepository: BookingsRepository, private readonly trainSchedulesRepository: TrainSchedulesRepository, private readonly trainScheduleBookingsRepository: TrainScheduleBookingsRepository, private readonly notifier: BookingNotifierService, private readonly scheduler: SchedulerRegistry, private readonly trainSchedulingService: TrainSchedulingService, private readonly billing: BillingService, @Optional() private readonly milestoneService?: ClearanceMilestoneService, ) {} /** On boot, reconcile OPEN route-days and re-arm settle timers. */ async onModuleInit(): Promise { const groups = await this.openRouteDayGroups(); for (const group of groups) { try { await this.processRouteDay(group); } catch (err) { this.logger.warn( `Boot reconcile failed for ${this.groupLabel(group)}: ${(err as Error).message}`, ); } } const reserved = await this.dataSource .getRepository(Booking) .createQueryBuilder("b") .select("DISTINCT b.train_schedule_id", "scheduleId") .where(`b.status IN ('SELECTED_FOR_BATCH', 'AWAITING_PAYMENT')`) .andWhere("b.train_schedule_id IS NOT NULL") .getRawMany<{ scheduleId: string }>(); for (const { scheduleId } of reserved) this.armSettle(scheduleId); } /** * Fire-and-forget batch pipeline for the (route, day) a schedule belongs to * (contract sign, payment). Day-level pooling distributes across all of that * day's trains, so a single schedule id maps to its whole route-day group. */ enqueueScheduleProcessing(scheduleId: string): void { void this.processRouteDayForSchedule(scheduleId).catch((err) => this.logger.error( `processRouteDay for schedule ${scheduleId} failed: ${(err as Error).message}`, ), ); } /** * Fire-and-forget batch pipeline for a (route, day) directly — used when a * booking enters the pool without a target train yet (e.g. after the * operations team accepts an operation request). The booking is already * FULLY_EXECUTED with its scheduled_date set, so the day-level fill will pick * it up; this just runs that fill immediately instead of waiting for the cron. */ enqueueRouteDayProcessing( originYardId: string, destinationYardId: string, day: string, ): void { void this.processRouteDay({ originYardId, destinationYardId, day }).catch( (err) => this.logger.error( `processRouteDay for ${originYardId}→${destinationYardId} on ${day} failed: ${(err as Error).message}`, ), ); } /** Resolve a schedule's (route, day) group and run the day-level pipeline. */ private async processRouteDayForSchedule(scheduleId: string): Promise { const schedule = await this.trainSchedulesRepository.findById(scheduleId); if (!schedule?.scheduledDepartureDate) return; await this.processRouteDay({ originYardId: schedule.originStationId, destinationYardId: schedule.destinationStationId, day: eatDay(schedule.scheduledDepartureDate), }); } /** * Day-level pipeline: distribute the (route, day) pool across all its trains, * then settle / reconcile / assign wagons per schedule (those steps stay * schedule-scoped — only the fill is day-level). */ async processRouteDay(group: RouteDayGroup): Promise { const scheduleIds = await this.fillRouteDay( group.originYardId, group.destinationYardId, group.day, ); for (const scheduleId of scheduleIds) { await this.settleDueReservations(scheduleId); await this.reconcilePaidUnlinked(scheduleId); await this.trainSchedulingService.tryAutoWagonAllocation(scheduleId); } } /** Fill pool, settle due reservations, link orphaned PAID, then assign wagons. */ async processSchedule(scheduleId: string): Promise { await this.fillSchedule(scheduleId); await this.settleDueReservations(scheduleId); await this.reconcilePaidUnlinked(scheduleId); await this.trainSchedulingService.tryAutoWagonAllocation(scheduleId); } /** Distinct (origin, destination, EAT day) groups across all OPEN schedules. */ private async openRouteDayGroups(): Promise { const open = await this.trainSchedulesRepository.findAll({ where: { bookingWindowStatus: "OPEN" }, }); const groups = new Map(); for (const s of open) { if (!s.scheduledDepartureDate) continue; const day = eatDay(s.scheduledDepartureDate); const key = `${s.originStationId}|${s.destinationStationId}|${day}`; if (!groups.has(key)) { groups.set(key, { originYardId: s.originStationId, destinationYardId: s.destinationStationId, day, }); } } return [...groups.values()]; } private groupLabel(group: RouteDayGroup): string { return `${group.originYardId}→${group.destinationYardId} on ${group.day}`; } /** * Idempotent: link a paid batch booking to its schedule and assign wagons. * Handles SELECTED_FOR_BATCH, PAID-without-link, and PAID-already-linked cases. */ async ensurePaidBookingAllocated(bookingId: string): Promise { const booking = await this.dataSource.getRepository(Booking).findOne({ where: { id: bookingId }, relations: { company: true }, }); if (!booking?.trainScheduleId) return; const isBatchPaid = booking.status === "SELECTED_FOR_BATCH" || booking.status === "AWAITING_PAYMENT" || booking.status === "PAID" || booking.paymentStatus === "PAID"; if (!isBatchPaid) return; if ( booking.status === "SELECTED_FOR_BATCH" || booking.status === "AWAITING_PAYMENT" ) { await this.dataSource .getRepository(Booking) .update(bookingId, { paymentStatus: "PAID", status: "PAID" }); } else if (booking.paymentStatus !== "PAID") { await this.dataSource .getRepository(Booking) .update(bookingId, { paymentStatus: "PAID" }); } const linked = await this.trainScheduleBookingsRepository.existsForBooking(bookingId); if (!linked) { await this.allocate(booking.trainScheduleId, booking, "paid"); this.logger.log( `Linked PAID booking ${booking.reference ?? bookingId} to schedule ${booking.trainScheduleId}`, ); } const schedule = await this.trainSchedulesRepository.findByIdWithFullGraph( booking.trainScheduleId, ); if (schedule && (await this.remainingWagons(schedule)) <= 0) { await this.setWindow(booking.trainScheduleId, "FULL"); } const result = await this.trainSchedulingService.tryAutoWagonAllocation( booking.trainScheduleId, ); if (result.assignedBookingIds.length) { this.logger.log( `Wagon allocation for ${booking.reference ?? bookingId}: ${result.assignedBookingIds.length} assigned`, ); } if ( result.issues.some( (i) => i.bookingId === bookingId && i.status !== "ASSIGNED", ) ) { const issue = result.issues.find((i) => i.bookingId === bookingId); this.logger.warn( `Wagon allocation issue for ${booking.reference ?? bookingId}: ${issue?.issue ?? issue?.status}`, ); } } /** Customer paid — delegate to ensurePaidBookingAllocated. */ async confirmPaidAndAllocate(bookingId: string): Promise { await this.ensurePaidBookingAllocated(bookingId); } /** Link PAID bookings that have no train_schedule_bookings row (cron backstop). */ async reconcilePaidUnlinked(scheduleId: string): Promise { const unlinked = await this.bookingsRepository.findPaidUnlinkedForSchedule(scheduleId); for (const booking of unlinked) { await this.allocate(scheduleId, booking, "paid"); this.logger.log( `Reconciled PAID booking ${booking.reference ?? booking.id} → schedule ${scheduleId}`, ); } } // ---- cron entry point ----------------------------------------------------- @Cron(BATCH_CRON, { name: "booking-batch-fill", timeZone: BATCH_TIMEZONE }) async runBatchFill(): Promise { const groups = await this.openRouteDayGroups(); this.logger.log(`Batch fill: ${groups.length} OPEN route-day group(s).`); for (const group of groups) { try { await this.processRouteDay(group); } catch (err) { this.logger.error( `Batch fill failed for ${this.groupLabel(group)}: ${(err as Error).message}`, ); } } } // ---- monitoring board ----------------------------------------------------- /** * Read model for the batch monitoring page: every still-relevant schedule (not arrived/ * cancelled) with its locomotive, capacity usage and its bookings grouped by lifecycle * state (allocated / awaiting payment / paid-waiting / pending contract / expired). */ async getBatchBoard(): Promise { const schedules = await this.trainSchedulesRepository.findAll({ relations: { trainSet: { locomotive: true }, originStation: true, destinationStation: true, route: true, }, order: { scheduledDepartureDate: "ASC" }, }); const wagonLengths = await this.loadWagonLengths(); const linkRepo = this.dataSource.getRepository(TrainScheduleBooking); const board: BatchBoardSchedule[] = []; for (const s of schedules) { if (s.status === "ARRIVED" || s.status === "CANCELLED") continue; const links = await linkRepo.find({ where: { trainScheduleId: s.id } }); const linkedIds = new Set(links.map((l) => l.bookingId)); const bookings = await this.bookingsRepository.findAllBySchedule(s.id); const items: BatchBoardBooking[] = bookings.map((b) => { const need = this.needFor(b, wagonLengths); return { id: b.id, reference: b.reference ?? b.id.slice(0, 8), company: b.isGovernment ? (b.governmentInstitution ?? "Government") : (b.company?.name ?? "—"), isGovernment: Boolean(b.isGovernment), wagons: need.wagons, weightTons: need.weightTons, lengthMeters: need.lengthMeters, paymentDeadline: b.paymentDeadline ? b.paymentDeadline.toISOString() : null, state: this.boardState(b, linkedIds.has(b.id)), }; }); board.push(this.buildScheduleSummary(s, items)); } return board; } /** Schedule-level batch board with EAT 3h windows grouped by fullyExecutedAt. */ async getBatchBoardDetail( scheduleId: string, ): Promise { const s = await this.trainSchedulesRepository.findByIdWithFullGraph(scheduleId); if (!s) throw new NotFoundException(`Train schedule ${scheduleId} not found`); if (s.status === "ARRIVED" || s.status === "CANCELLED") { throw new BadRequestException("Schedule is no longer active"); } const wagonLengths = await this.loadWagonLengths(); const linkRepo = this.dataSource.getRepository(TrainScheduleBooking); const links = await linkRepo.find({ where: { trainScheduleId: s.id } }); const linkedIds = new Set(links.map((l) => l.bookingId)); const bookings = await this.bookingsRepository.findAllBySchedule(s.id); let allocationPreview: Awaited< ReturnType >; try { allocationPreview = await this.trainSchedulingService.previewAllocationForSchedule(s.id); } catch { allocationPreview = { assignedBookingIds: [], deferred: [], issues: [], violations: [], }; } const allocationByBooking = new Map( allocationPreview.issues.map((i) => [i.bookingId, i]), ); const items: BatchBoardBookingDetail[] = bookings.map((b) => { const need = this.needFor(b, wagonLengths); const alloc = allocationByBooking.get(b.id); return { id: b.id, reference: b.reference ?? b.id.slice(0, 8), company: b.isGovernment ? (b.governmentInstitution ?? "Government") : (b.company?.name ?? "—"), isGovernment: Boolean(b.isGovernment), wagons: need.wagons, weightTons: need.weightTons, lengthMeters: need.lengthMeters, paymentDeadline: b.paymentDeadline ? b.paymentDeadline.toISOString() : null, state: this.boardState(b, linkedIds.has(b.id)), fullyExecutedAt: b.fullyExecutedAt ? b.fullyExecutedAt.toISOString() : null, selectedForBatchAt: b.selectedForBatchAt ? b.selectedForBatchAt.toISOString() : null, allocationStatus: alloc?.status ?? "NOT_ATTEMPTED", allocationIssue: alloc?.issue ?? null, }; }); const loco = s.trainSet?.locomotive ?? null; // Display windows span the whole booking window: from when it opened // (schedule creation) through the scheduled departure, in 3-hour EAT slots. const openDate = s.createdAt ?? s.scheduledDepartureDate ?? new Date(); const departureDate = s.scheduledDepartureDate ?? new Date(); const windowBuckets = groupBookingsIntoBoardWindows( items, (item) => (item.fullyExecutedAt ? new Date(item.fullyExecutedAt) : null), openDate, departureDate, ); const emptyCounts = () => ({ allocated: 0, selectedForBatch: 0, ready: 0, waiting: 0, expired: 0, pendingContract: 0, }); const countFor = (bookingsInWindow: BatchBoardBookingDetail[]) => { const counts = emptyCounts(); for (const b of bookingsInWindow) { if (b.state === "ALLOCATED") counts.allocated += 1; else if (b.state === "SELECTED_FOR_BATCH") counts.selectedForBatch += 1; else if (b.state === "READY") counts.ready += 1; else if (b.state === "WAITING") counts.waiting += 1; else if (b.state === "EXPIRED") counts.expired += 1; else counts.pendingContract += 1; } return counts; }; const windows: BatchWindowGroup[] = []; for (const [key, bucket] of windowBuckets) { if (key === "pending-contract" || !bucket.window) continue; const w = bucket.window; windows.push({ key: w.key, label: w.label, date: w.date, dateLabel: w.dateLabel, start: w.start.toISOString(), end: w.end.toISOString(), counts: countFor(bucket.items), bookings: bucket.items, }); } windows.sort( (a, b) => new Date(a.start).getTime() - new Date(b.start).getTime(), ); const pendingBookings = windowBuckets.get("pending-contract")?.items ?? []; return { scheduleId: s.id, trainNumber: s.trainNumber ?? null, routeName: s.route ? formatRouteLabel(s.route) : null, origin: s.originStation?.label ?? s.originStation?.code ?? null, destination: s.destinationStation?.label ?? s.destinationStation?.code ?? null, scheduleDate: s.scheduledDepartureDate ? s.scheduledDepartureDate.toISOString() : null, status: s.status, bookingWindowStatus: s.bookingWindowStatus, locomotive: loco ? { code: loco.code, name: loco.name ?? null, maxPullWeightTons: Number(loco.maxPullWeightTons), maxTrainLengthMeters: Number(loco.maxTrainLengthMeters), } : null, capacity: this.computeBoardCapacity(items, loco), counts: { allocated: items.filter((i) => i.state === "ALLOCATED").length, selectedForBatch: items.filter((i) => i.state === "SELECTED_FOR_BATCH") .length, ready: items.filter((i) => i.state === "READY").length, waiting: items.filter((i) => i.state === "WAITING").length, pendingContract: items.filter((i) => i.state === "PENDING_CONTRACT") .length, expired: items.filter((i) => i.state === "EXPIRED").length, }, windows, pendingContract: { key: "pending-contract", label: "Pending contract", date: "", dateLabel: "", start: "", end: "", counts: countFor(pendingBookings), bookings: pendingBookings, }, allocationViolations: allocationPreview.violations, }; } /** Run wagon-level allocation for all eligible linked bookings on a schedule. */ async runWagonAllocation(scheduleId: string) { return this.trainSchedulingService.tryAutoWagonAllocation(scheduleId); } private computeBoardCapacity( items: Array<{ state: BatchBoardBookingState; wagons: number; weightTons: number; lengthMeters: number; }>, loco: Locomotive | null, ): BatchBoardSchedule["capacity"] { const allocated = items.filter((i) => i.state === "ALLOCATED"); const committed = items.filter( (i) => i.state === "ALLOCATED" || i.state === "SELECTED_FOR_BATCH", ); return { allocatedWagons: allocated.reduce((sum, i) => sum + i.wagons, 0), allocatedLengthMeters: Math.round( allocated.reduce((sum, i) => sum + i.lengthMeters, 0) * 100, ) / 100, maxLengthMeters: loco ? Number(loco.maxTrainLengthMeters) : null, usedWeightTons: Math.round(committed.reduce((sum, i) => sum + i.weightTons, 0) * 100) / 100, maxWeightTons: loco ? Number(loco.maxPullWeightTons) : null, }; } private buildScheduleSummary( s: TrainSchedule, items: BatchBoardBooking[], ): BatchBoardSchedule { const loco = s.trainSet?.locomotive ?? null; return { scheduleId: s.id, trainNumber: s.trainNumber ?? null, routeName: s.route ? formatRouteLabel(s.route) : null, origin: s.originStation?.label ?? s.originStation?.code ?? null, destination: s.destinationStation?.label ?? s.destinationStation?.code ?? null, scheduleDate: s.scheduledDepartureDate ? s.scheduledDepartureDate.toISOString() : null, status: s.status, bookingWindowStatus: s.bookingWindowStatus, locomotive: loco ? { code: loco.code, name: loco.name ?? null, maxPullWeightTons: Number(loco.maxPullWeightTons), maxTrainLengthMeters: Number(loco.maxTrainLengthMeters), } : null, capacity: this.computeBoardCapacity(items, loco), counts: { allocated: items.filter((i) => i.state === "ALLOCATED").length, selectedForBatch: items.filter((i) => i.state === "SELECTED_FOR_BATCH") .length, ready: items.filter((i) => i.state === "READY").length, waiting: items.filter((i) => i.state === "WAITING").length, pendingContract: items.filter((i) => i.state === "PENDING_CONTRACT") .length, expired: items.filter((i) => i.state === "EXPIRED").length, }, bookings: items.slice(0, 3), }; } private boardState( booking: Booking, linked: boolean, ): BatchBoardBookingState { if (linked) return "ALLOCATED"; if ( booking.status === "SELECTED_FOR_BATCH" || booking.status === "AWAITING_PAYMENT" ) { return "SELECTED_FOR_BATCH"; } if (booking.status === "EXPIRED") return "EXPIRED"; if (booking.status === "FULLY_EXECUTED" && booking.fullyExecutedAt) return "READY"; if (booking.status === "PAID") return "WAITING"; return "PENDING_CONTRACT"; } // ---- core fill ------------------------------------------------------------ /** Fill one schedule from its priority-ordered pool until full. */ async fillSchedule(scheduleId: string): Promise { const schedule = await this.trainSchedulesRepository.findByIdWithFullGraph(scheduleId); if (!schedule || schedule.bookingWindowStatus !== "OPEN") return; const locomotive = schedule.trainSet?.locomotive; if (!schedule.trainSetId || !locomotive) { this.logger.warn( `Schedule ${scheduleId} has no locomotive/train set — skipped.`, ); return; } const rules = await this.loadGlobalRules(); const wagonLengths = await this.loadWagonLengths(); const limits = await this.capacityLimits(locomotive, rules); await this.syncScheduleMaxWagons(schedule, locomotive, rules); let budget = await this.remainingCapacity(schedule, limits, wagonLengths); if (budget.wagons <= 0) { await this.setWindow(scheduleId, "FULL"); return; } const pool = await this.bookingsRepository.findBatchPool(scheduleId); let armed = false; for (const booking of pool) { const need = this.needFor(booking, wagonLengths); if (!this.fits(need, budget)) { if (booking.isGovernment) { budget = await this.preemptForGovernment( scheduleId, need, budget, wagonLengths, ); if (!this.fits(need, budget)) continue; // still doesn't fit even after preempt } else { continue; // skip a booking that exceeds weight/length/wagons, try the next } } if (booking.isGovernment) { await this.allocate(scheduleId, booking, "gov"); } else { await this.reserve(booking, scheduleId); armed = true; } budget = this.subtract(budget, need); if (budget.wagons <= 0) break; // no wagon slots left — nothing more can board } if (budget.wagons <= 0) await this.setWindow(scheduleId, "FULL"); if (armed) this.armSettle(scheduleId); void this.triggerWagonAllocation(scheduleId); } /** * Distribute one (route, day) pool across ALL of that day's OPEN trains, by * priority, filling each train (earliest departure first) until it's full and * spilling overflow to the next. Government bookings that fit no train preempt * lower-priority commercial; bookings that fit no train at all stay pending and * trigger a staff `unplaced` warning. Returns the schedule ids that were touched * (or that had remaining pool work) so the caller can settle them per-schedule. */ async fillRouteDay( originYardId: string, destinationYardId: string, day: string, ): Promise { // The day's OPEN bookable schedules on this exact corridor, earliest first. const bookable = await this.trainSchedulingService.getBookableSchedules( originYardId, destinationYardId, ); const scheduleIds = bookable .filter( (s) => s.bookingWindowStatus === "OPEN" && s.scheduleDate != null && eatDay(new Date(s.scheduleDate)) === day, ) .sort( (a, b) => new Date(a.scheduleDate).getTime() - new Date(b.scheduleDate).getTime(), ) .map((s) => s.id); if (scheduleIds.length === 0) return []; const rules = await this.loadGlobalRules(); const wagonLengths = await this.loadWagonLengths(); // Live per-schedule budget + arm flag, in departure order. const trains: Array<{ id: string; budget: Capacity; armed: boolean }> = []; for (const id of scheduleIds) { const schedule = await this.trainSchedulesRepository.findByIdWithFullGraph(id); const locomotive = schedule?.trainSet?.locomotive; if (!schedule || !schedule.trainSetId || !locomotive) { this.logger.warn( `Schedule ${id} has no locomotive/train set — skipped.`, ); continue; } const limits = await this.capacityLimits(locomotive, rules); await this.syncScheduleMaxWagons(schedule, locomotive, rules); const budget = await this.remainingCapacity( schedule, limits, wagonLengths, ); trains.push({ id, budget, armed: false }); } if (trains.length === 0) return []; const pool = await this.bookingsRepository.findBatchPoolByRouteDay( originYardId, destinationYardId, day, ); for (const booking of pool) { const need = this.needFor(booking, wagonLengths); // First train (earliest departure) that fits this booking as-is. let target = trains.find((t) => this.fits(need, t.budget)); if (!target && booking.isGovernment) { // Government booking fits nowhere on its own — try to preempt commercial // on each train (earliest first) until one frees enough room. for (const t of trains) { t.budget = await this.preemptForGovernment( t.id, need, t.budget, wagonLengths, ); if (this.fits(need, t.budget)) { target = t; break; } } } if (!target) { // Fits no train this day — stays in the pool, retried next batch. this.notifier.unplaced(booking, day); continue; } if (booking.isGovernment) { await this.allocate(target.id, booking, "gov"); } else { await this.reserve(booking, target.id); target.armed = true; } target.budget = this.subtract(target.budget, need); } for (const t of trains) { if (t.budget.wagons <= 0) await this.setWindow(t.id, "FULL"); if (t.armed) this.armSettle(t.id); void this.triggerWagonAllocation(t.id); } return trains.map((t) => t.id); } /** Durable settle: allocate paid / expire overdue reservations, then top up. */ async settleDueReservations(scheduleId: string): Promise { const reserved = await this.bookingsRepository.findReservedForSchedule(scheduleId); const now = Date.now(); let anySettled = false; for (const booking of reserved) { const paid = booking.paymentStatus === "PAID" || booking.status === "PAID"; const expired = booking.paymentDeadline ? booking.paymentDeadline.getTime() <= now : false; if (paid) { await this.allocate(scheduleId, booking, "paid"); anySettled = true; } else if (expired) { await this.expire(booking); anySettled = true; } } if (anySettled) await this.fillSchedule(scheduleId); } // ---- settle (1h after a batch) ------------------------------------------- /** Allocate paid reservations, expire the rest, then top up. */ async settleBatch(scheduleId: string): Promise { this.removeTimeout(scheduleId); const reserved = await this.bookingsRepository.findReservedForSchedule(scheduleId); const now = Date.now(); for (const booking of reserved) { const paid = booking.paymentStatus === "PAID" || booking.status === "PAID"; const expired = booking.paymentDeadline ? booking.paymentDeadline.getTime() <= now : true; if (paid) { await this.allocate(scheduleId, booking, "paid"); } else if (expired) { await this.expire(booking); } // else: still within window (rare at settle) → leave for the re-armed timeout } await this.fillSchedule(scheduleId); void this.triggerWagonAllocation(scheduleId); } private triggerWagonAllocation(scheduleId: string): void { void this.trainSchedulingService .tryAutoWagonAllocation(scheduleId) .catch((err) => this.logger.warn( `Auto wagon allocation failed for ${scheduleId}: ${(err as Error).message}`, ), ); } // ---- staff override actions ---------------------------------------------- /** Staff "mark paid" override → set PAID and allocate immediately (don't wait for settle). */ async markPaid(bookingId: string): Promise { const booking = await this.dataSource .getRepository(Booking) .findOne({ where: { id: bookingId } }); if (!booking) throw new NotFoundException(`Booking ${bookingId} not found`); if (!booking.trainScheduleId) { throw new BadRequestException( "Booking has no target schedule to allocate to", ); } await this.dataSource .getRepository(Booking) .update(bookingId, { paymentStatus: "PAID" }); await this.allocate(booking.trainScheduleId, booking, "paid"); const schedule = await this.trainSchedulesRepository.findByIdWithFullGraph( booking.trainScheduleId, ); if (schedule && (await this.remainingWagons(schedule)) <= 0) { await this.setWindow(booking.trainScheduleId, "FULL"); } void this.triggerWagonAllocation(booking.trainScheduleId!); } /** * Re-point a booking to another OPEN same-route schedule (keeps approval/contract + priority). * Used for EXPIRED or full-schedule bookings — no re-approval. */ async moveToSchedule( bookingId: string, newScheduleId: string, ): Promise { const booking = await this.dataSource .getRepository(Booking) .findOne({ where: { id: bookingId } }); if (!booking) throw new NotFoundException(`Booking ${bookingId} not found`); const schedule = await this.dataSource .getRepository(TrainSchedule) .findOne({ where: { id: newScheduleId } }); if (!schedule) throw new NotFoundException(`Train schedule ${newScheduleId} not found`); if (schedule.bookingWindowStatus !== "OPEN") { throw new BadRequestException( "Target schedule is not accepting bookings", ); } if ( schedule.originStationId !== booking.originYardId || schedule.destinationStationId !== booking.destinationYardId ) { throw new BadRequestException( "Target schedule is not on the booking route", ); } await this.dataSource.transaction(async (manager) => { if (booking.trainScheduleId) { await this.trainScheduleBookingsRepository.deleteByScheduleAndBooking( booking.trainScheduleId, bookingId, manager, ); } const restoredStatus = booking.status === "EXPIRED" ? booking.isGovernment ? "APPROVED" : "FULLY_EXECUTED" : booking.status; await manager.getRepository(Booking).update(bookingId, { trainScheduleId: newScheduleId, status: restoredStatus, schedulingStatus: "ELIGIBLE", paymentDeadline: null, selectedForBatchAt: null, } as never); }); } /** Staff "expire" override → free a reservation now (booking becomes EXPIRED). */ async expireReservation(bookingId: string): Promise { const booking = await this.dataSource .getRepository(Booking) .findOne({ where: { id: bookingId } }); if (!booking) throw new NotFoundException(`Booking ${bookingId} not found`); await this.expire(booking); if (booking.trainScheduleId) await this.fillSchedule(booking.trainScheduleId); } // ---- mutations ------------------------------------------------------------ /** * Reserve capacity for a commercial booking on a specific train and open its * pay window. `scheduleId` is persisted so the settle/allocate lifecycle * (settleDueReservations, settleBatch, ensurePaidBookingAllocated, markPaid), * which is all keyed off `booking.trainScheduleId`, can find the train — with * day-level pooling the booking arrives here with `trainScheduleId` still null, * so the engine sets it as it picks the train. */ private async reserve(booking: Booking, scheduleId: string): Promise { const now = new Date(); const deadline = new Date(now.getTime() + PAYMENT_WINDOW_MS); await this.bookingsRepository.update(booking.id, { trainScheduleId: scheduleId, status: "SELECTED_FOR_BATCH", selectedForBatchAt: now, paymentDeadline: deadline, } as never); booking.trainScheduleId = scheduleId; // The invoice was generated at booking creation/approval, before this pay // window opened — refresh its printed due date to the real deadline. await this.billing.syncPayableDueDate( Freight.InvoiceSource.Booking, booking.id, deadline, "PREPAID", ); await this.notifier.payNow(booking, deadline); } /** Allocate a booking to the schedule's train (creates the TrainScheduleBooking link). */ private async allocate( scheduleId: string, booking: Booking, reason: "paid" | "gov", ): Promise { await this.dataSource.transaction(async (manager) => { const exists = await this.trainScheduleBookingsRepository.existsForBooking( booking.id, manager, ); if (!exists) { await this.trainScheduleBookingsRepository.createMany( [{ trainScheduleId: scheduleId, bookingId: booking.id }], manager, ); } await manager.getRepository(Booking).update(booking.id, { status: reason === "paid" ? "PAID" : booking.status, schedulingStatus: "SCHEDULED", scheduledAt: new Date(), paymentDeadline: null, selectedForBatchAt: null, } as never); }); this.notifier.secured(booking, reason); void this.triggerWagonAllocation(scheduleId); void this.markWagonAllocatedMilestone(booking.id); } private async markWagonAllocatedMilestone(bookingId: string): Promise { if (!this.milestoneService) return; try { await this.milestoneService.completeForBooking(bookingId, 'WAGON_ALLOCATED'); } catch { // Booking may have no milestone rows (non-contract path). } } /** * Expire an unpaid reservation and free its capacity. With day-level pooling we * also clear `trainScheduleId` so the booking is no longer pinned to the train * it failed to pay for — it's back in the day pool for staff to act on. */ private async expire(booking: Booking): Promise { await this.bookingsRepository.update(booking.id, { trainScheduleId: null, status: "EXPIRED", schedulingStatus: "ELIGIBLE", paymentDeadline: null, selectedForBatchAt: null, } as never); booking.trainScheduleId = null; // Pay window closed before settlement → expire the booking's open invoice too // (emits `booking.invoice.expired`). Domain owns the reaction; billing stays // source-agnostic. await this.billing.expirePayable(Freight.InvoiceSource.Booking, booking.id, "PREPAID"); this.notifier.expired(booking); } /** * Free capacity for a government booking by displacing the lowest-priority commercial * bookings (reserved first, then allocated — including PAID). Displaced → EXPIRED + notified. */ private async preemptForGovernment( scheduleId: string, need: Capacity, budget: Capacity, wagonLengths: WagonLengths, ): Promise { const reservedCommercial = ( await this.bookingsRepository.findReservedForSchedule(scheduleId) ).filter((b) => !b.isGovernment); const allocatedCommercial = await this.bookingsRepository.findAllocatedCommercialForSchedule( scheduleId, ); // lowest priority first; reserved are cheaper to free than allocated const candidates = [...reservedCommercial, ...allocatedCommercial].sort( (a, b) => (a.priorityScore ?? 0) - (b.priorityScore ?? 0), ); let freed = budget; for (const victim of candidates) { if (this.fits(need, freed)) break; await this.dataSource.transaction(async (manager) => { await this.trainScheduleBookingsRepository.deleteByScheduleAndBooking( scheduleId, victim.id, manager, ); await manager.getRepository(Booking).update(victim.id, { status: "EXPIRED", schedulingStatus: "ELIGIBLE", paymentDeadline: null, selectedForBatchAt: null, } as never); // Displaced → EXPIRED: close its open invoice too, so a dead booking // can't still be paid (mirrors `expire()`; enlisted in this txn). await this.billing.expirePayable( Freight.InvoiceSource.Booking, victim.id, "PREPAID", manager, ); }); this.notifier.displaced(victim); freed = this.add(freed, this.needFor(victim, wagonLengths)); } return freed; } // ---- capacity helpers ----------------------------------------------------- private wagonsFor(booking: Booking): number { if (booking.wagonsRequired && booking.wagonsRequired > 0) { return Math.ceil(booking.wagonsRequired); } const fromContainers = (booking.bookingContainers ?? []).reduce( (sum, c) => sum + Number(c.quantity ?? 0), 0, ); return Math.max( DEFAULT_WAGONS_PER_BOOKING, fromContainers || DEFAULT_WAGONS_PER_BOOKING, ); } /** What one booking consumes along all three capacity axes. */ private needFor(booking: Booking, wagonLengths: WagonLengths): Capacity { const wagons = this.wagonsFor(booking); return { wagons, weightTons: Number(booking.cargoTotalWeightVgm ?? 0), lengthMeters: bookingTrainLengthMeters(booking.freightType, wagons, { container: wagonLengths.container, bulk: wagonLengths.bulk, }), }; } private fits(need: Capacity, budget: Capacity): boolean { return ( need.wagons <= budget.wagons && need.weightTons <= budget.weightTons && need.lengthMeters <= budget.lengthMeters ); } private subtract(budget: Capacity, need: Capacity): Capacity { return { wagons: budget.wagons - need.wagons, weightTons: budget.weightTons - need.weightTons, lengthMeters: budget.lengthMeters - need.lengthMeters, }; } private add(budget: Capacity, freed: Capacity): Capacity { return { wagons: budget.wagons + freed.wagons, weightTons: budget.weightTons + freed.weightTons, lengthMeters: budget.lengthMeters + freed.lengthMeters, }; } /** Locomotive + wagon-type-derived caps (weight, length, wagon slots — not a fixed 53). */ private async capacityLimits( locomotive: Locomotive, rules: TrainSchedulingGlobalRules | null, ): Promise { const wagonTypes = await this.loadWagonTypeDimensions(); const derived = deriveTrainCapacityFromLocomotive( { maxPullWeightTons: Number(locomotive.maxPullWeightTons), maxTrainLengthMeters: Number(locomotive.maxTrainLengthMeters), }, wagonTypes, { maxTrainWeightTons: rules?.maxTrainWeightTons ? Number(rules.maxTrainWeightTons) : undefined, maxTrainLengthMeters: rules?.maxTrainLengthMeters ? Number(rules.maxTrainLengthMeters) : undefined, }, ); return { wagons: derived.maxWagonSlots, weightTons: derived.maxWeightTons, lengthMeters: derived.maxLengthMeters, }; } /** Keep schedule.max_wagons aligned with locomotive physical limits. */ private async syncScheduleMaxWagons( schedule: TrainSchedule, locomotive: Locomotive, rules: TrainSchedulingGlobalRules | null, ): Promise { const limits = await this.capacityLimits(locomotive, rules); if ((schedule.maxWagons ?? 0) !== limits.wagons) { await this.dataSource .getRepository(TrainSchedule) .update(schedule.id, { maxWagons: limits.wagons }); schedule.maxWagons = limits.wagons; } } private async loadWagonTypeDimensions(): Promise< Array<{ lengthMeters: number; capacityTons: number }> > { const types = await this.dataSource.getRepository(WagonType).find({ where: [{ code: "NW5" }, { code: "CW3" }], }); if (types.length) return types.map(wagonTypeDimensionsFromEntity); return [ { lengthMeters: DEFAULT_CONTAINER_WAGON_LENGTH_METERS, capacityTons: 70 }, { lengthMeters: DEFAULT_BULK_WAGON_LENGTH_METERS, capacityTons: 60 }, ]; } private async loadWagonLengths(): Promise { const types = await this.dataSource.getRepository(WagonType).find({ where: [{ code: "NW5" }, { code: "CW3" }], }); const byCode = new Map( types.map((t) => [t.code, wagonTypeDimensionsFromEntity(t)]), ); return { container: byCode.get("NW5")?.lengthMeters ?? DEFAULT_CONTAINER_WAGON_LENGTH_METERS, bulk: byCode.get("CW3")?.lengthMeters ?? DEFAULT_BULK_WAGON_LENGTH_METERS, }; } private async loadGlobalRules(): Promise { return this.dataSource .getRepository(TrainSchedulingGlobalRules) .findOne({ where: {} }); } /** Remaining capacity = hard caps minus what allocated + reserved bookings already use. */ private async remainingCapacity( schedule: TrainSchedule, limits: Capacity, wagonLengths: WagonLengths, ): Promise { const allocated = (schedule.scheduleBookings ?? []) .map((sb) => sb.booking) .filter((b): b is Booking => Boolean(b)); const reserved = await this.bookingsRepository.findReservedForSchedule( schedule.id, ); const used = [...allocated, ...reserved].reduce( (acc, b) => this.add(acc, this.needFor(b, wagonLengths)), { wagons: 0, weightTons: 0, lengthMeters: 0 }, ); return this.subtract(limits, used); } /** maxWagons minus wagons already taken by allocated + reserved bookings. */ private async remainingWagons(schedule: TrainSchedule): Promise { const allocated = (schedule.scheduleBookings ?? []) .map((sb) => sb.booking) .filter((b): b is Booking => Boolean(b)); const reserved = await this.bookingsRepository.findReservedForSchedule( schedule.id, ); const used = allocated.reduce((s, b) => s + this.wagonsFor(b), 0) + reserved.reduce((s, b) => s + this.wagonsFor(b), 0); return (schedule.maxWagons ?? 0) - used; } private async setWindow( scheduleId: string, status: "OPEN" | "FULL" | "CLOSED", ): Promise { await this.dataSource .getRepository(TrainSchedule) .update(scheduleId, { bookingWindowStatus: status }); } // ---- timer plumbing ------------------------------------------------------- private timeoutName(scheduleId: string): string { return `settle:${scheduleId}`; } private armSettle(scheduleId: string): void { this.removeTimeout(scheduleId); const handle = setTimeout(() => { void this.settleBatch(scheduleId).catch((err) => this.logger.error( `settleBatch ${scheduleId} failed: ${(err as Error).message}`, ), ); }, PAYMENT_WINDOW_MS); this.scheduler.addTimeout(this.timeoutName(scheduleId), handle); } private removeTimeout(scheduleId: string): void { const name = this.timeoutName(scheduleId); try { if (this.scheduler.doesExist("timeout", name)) { this.scheduler.deleteTimeout(name); } } catch { // ignore — not armed } } }