Files
edr-platform/apps/edr-freight-api/src/modules/train-scheduling/booking-batch.service.ts
2026-06-11 20:59:19 +00:00

624 lines
23 KiB
TypeScript

import {
BadRequestException,
Injectable,
Logger,
NotFoundException,
OnModuleInit,
} 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 { 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 {
BATCH_CRON,
BATCH_TIMEZONE,
DEFAULT_WAGON_LENGTH_METERS,
DEFAULT_WAGONS_PER_BOOKING,
PAYMENT_WINDOW_MS,
} from './booking-batch.constants';
/** A train's remaining capacity along the three physical limits the batch enforces. */
interface Capacity {
wagons: number;
weightTons: number;
lengthMeters: number;
}
export type BatchBoardBookingState =
| 'ALLOCATED'
| 'AWAITING_PAYMENT'
| 'WAITING'
| 'PENDING_CONTRACT'
| 'EXPIRED';
export interface BatchBoardBooking {
id: string;
reference: string;
company: string;
isGovernment: boolean;
wagons: number;
weightTons: number;
paymentDeadline: string | null;
state: BatchBoardBookingState;
}
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: {
maxWagons: number;
usedWagons: number;
remainingWagons: number;
usedWeightTons: number;
maxWeightTons: number | null;
};
counts: {
allocated: number;
awaitingPayment: 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,
) {}
/** On boot, re-arm a settle timeout for any schedule that still has live reservations. */
async onModuleInit(): Promise<void> {
const reserved = await this.dataSource
.getRepository(Booking)
.createQueryBuilder('b')
.select('DISTINCT b.train_schedule_id', 'scheduleId')
.where(`b.status = 'AWAITING_PAYMENT'`)
.andWhere('b.train_schedule_id IS NOT NULL')
.getRawMany<{ scheduleId: string }>();
for (const { scheduleId } of reserved) this.armSettle(scheduleId);
}
// ---- cron entry point -----------------------------------------------------
@Cron(BATCH_CRON, { name: 'booking-batch-fill', timeZone: BATCH_TIMEZONE })
async runBatchFill(): Promise<void> {
const open = await this.trainSchedulesRepository.findAll({
where: { bookingWindowStatus: 'OPEN' },
});
this.logger.log(`Batch fill: ${open.length} OPEN schedule(s).`);
for (const s of open) {
try {
await this.fillSchedule(s.id);
} catch (err) {
this.logger.error(`Batch fill failed for ${s.id}: ${(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<BatchBoardSchedule[]> {
const schedules = await this.trainSchedulesRepository.findAll({
relations: {
trainSet: { locomotive: true },
originStation: true,
destinationStation: true,
route: true,
},
order: { scheduledDepartureDate: 'ASC' },
});
const rules = await this.loadGlobalRules();
const perWagonLength = this.perWagonLength(rules);
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, perWagonLength);
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,
paymentDeadline: b.paymentDeadline ? b.paymentDeadline.toISOString() : null,
state: this.boardState(b, linkedIds.has(b.id)),
};
});
const usedWagons = items
.filter((i) => i.state === 'ALLOCATED' || i.state === 'AWAITING_PAYMENT')
.reduce((sum, i) => sum + i.wagons, 0);
const usedWeight = items
.filter((i) => i.state === 'ALLOCATED' || i.state === 'AWAITING_PAYMENT')
.reduce((sum, i) => sum + i.weightTons, 0);
const loco = s.trainSet?.locomotive ?? null;
board.push({
scheduleId: s.id,
trainNumber: s.trainNumber ?? null,
routeName: s.route?.name ?? 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: {
maxWagons: s.maxWagons ?? 0,
usedWagons,
remainingWagons: Math.max(0, (s.maxWagons ?? 0) - usedWagons),
usedWeightTons: Math.round(usedWeight * 100) / 100,
maxWeightTons: loco ? Number(loco.maxPullWeightTons) : null,
},
counts: {
allocated: items.filter((i) => i.state === 'ALLOCATED').length,
awaitingPayment: items.filter((i) => i.state === 'AWAITING_PAYMENT').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,
});
}
return board;
}
private boardState(booking: Booking, linked: boolean): BatchBoardBookingState {
if (linked) return 'ALLOCATED';
if (booking.status === 'AWAITING_PAYMENT') return 'AWAITING_PAYMENT';
if (booking.status === 'EXPIRED') return 'EXPIRED';
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<void> {
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 perWagonLength = this.perWagonLength(rules);
const limits = this.capacityLimits(schedule, locomotive, rules);
let budget = await this.remainingCapacity(schedule, limits, perWagonLength);
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, perWagonLength);
if (!this.fits(need, budget)) {
if (booking.isGovernment) {
budget = await this.preemptForGovernment(scheduleId, need, budget, perWagonLength);
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);
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);
}
// ---- settle (1h after a batch) -------------------------------------------
/** Allocate paid reservations, expire the rest, then top up. */
async settleBatch(scheduleId: string): Promise<void> {
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);
}
// ---- staff override actions ----------------------------------------------
/** Staff "mark paid" override → set PAID and allocate immediately (don't wait for settle). */
async markPaid(bookingId: string): Promise<void> {
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');
}
}
/**
* 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<void> {
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,
} as never);
});
}
/** Staff "expire" override → free a reservation now (booking becomes EXPIRED). */
async expireReservation(bookingId: string): Promise<void> {
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 and open its 1h pay window. */
private async reserve(booking: Booking): Promise<void> {
const deadline = new Date(Date.now() + PAYMENT_WINDOW_MS);
await this.bookingsRepository.update(booking.id, {
status: 'AWAITING_PAYMENT',
paymentDeadline: deadline,
} as never);
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<void> {
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,
} as never);
});
this.notifier.secured(booking, reason);
}
/** Expire an unpaid reservation and free its capacity. */
private async expire(booking: Booking): Promise<void> {
await this.bookingsRepository.update(booking.id, {
status: 'EXPIRED',
schedulingStatus: 'ELIGIBLE',
paymentDeadline: null,
} as never);
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,
perWagonLength: number,
): Promise<Capacity> {
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,
} as never);
});
this.notifier.displaced(victim);
freed = this.add(freed, this.needFor(victim, perWagonLength));
}
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, perWagonLength: number): Capacity {
const wagons = this.wagonsFor(booking);
return {
wagons,
weightTons: Number(booking.cargoTotalWeightVgm ?? 0),
lengthMeters: wagons * perWagonLength,
};
}
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,
};
}
/** The train's hard caps: wagon count, locomotive pull weight, locomotive/global length. */
private capacityLimits(
schedule: TrainSchedule,
locomotive: Locomotive,
rules: TrainSchedulingGlobalRules | null,
): Capacity {
const locoWeight = Number(locomotive.maxPullWeightTons) || Infinity;
const locoLength = Number(locomotive.maxTrainLengthMeters) || Infinity;
const ruleWeight = rules?.maxTrainWeightTons ? Number(rules.maxTrainWeightTons) : Infinity;
const ruleLength = rules?.maxTrainLengthMeters ? Number(rules.maxTrainLengthMeters) : Infinity;
return {
wagons: schedule.maxWagons ?? 0,
weightTons: Math.min(locoWeight, ruleWeight),
lengthMeters: Math.min(locoLength, ruleLength),
};
}
/** Per-wagon length, derived from global rules (maxLength / maxWagons) or a fallback. */
private perWagonLength(rules: TrainSchedulingGlobalRules | null): number {
const len = rules ? Number(rules.maxTrainLengthMeters) : 0;
const wagons = rules ? Number(rules.maxWagonsPerTrain) : 0;
if (len > 0 && wagons > 0) return len / wagons;
return DEFAULT_WAGON_LENGTH_METERS;
}
private async loadGlobalRules(): Promise<TrainSchedulingGlobalRules | null> {
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,
perWagonLength: number,
): Promise<Capacity> {
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<Capacity>(
(acc, b) => this.add(acc, this.needFor(b, perWagonLength)),
{ 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<number> {
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<void> {
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
}
}
}