mirror of
https://github.com/Tria-plc/edr-platform.git
synced 2026-08-30 02:58:11 +00:00
automate the schedule
This commit is contained in:
Marshal
@@ -0,0 +1,377 @@
|
||||
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 { TrainSchedule } from '../train-schedules/entities/train-schedule.entity';
|
||||
import { TrainSchedulesRepository } from '../train-schedules/train-schedules.repository';
|
||||
import { TrainScheduleBookingsRepository } from '../train-schedules/train-schedule-bookings.repository';
|
||||
import { BookingNotifierService } from './booking-notifier.service';
|
||||
import {
|
||||
BATCH_CRON,
|
||||
BATCH_TIMEZONE,
|
||||
DEFAULT_WAGONS_PER_BOOKING,
|
||||
PAYMENT_WINDOW_MS,
|
||||
} from './booking-batch.constants';
|
||||
|
||||
/**
|
||||
* Demand-batching engine: every 3h (EAT) it ranks each OPEN schedule's ready pool
|
||||
* by priority, greedily fills the train to capacity (skipping oversized bookings),
|
||||
* 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 here is modelled by wagon count (`schedule.maxWagons`); locomotive weight/length
|
||||
* is still enforced by the existing assignment path when staff pin wagons.
|
||||
*/
|
||||
@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}`);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ---- 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;
|
||||
if (!schedule.trainSetId || !schedule.trainSet?.locomotive) {
|
||||
this.logger.warn(`Schedule ${scheduleId} has no locomotive/train set — skipped.`);
|
||||
return;
|
||||
}
|
||||
|
||||
let remaining = await this.remainingWagons(schedule);
|
||||
if (remaining <= 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.wagonsFor(booking);
|
||||
|
||||
if (need > remaining) {
|
||||
if (booking.isGovernment) {
|
||||
remaining = await this.preemptForGovernment(scheduleId, need, remaining);
|
||||
if (need > remaining) continue; // still doesn't fit even after preempt
|
||||
} else {
|
||||
continue; // skip oversized commercial, try the next
|
||||
}
|
||||
}
|
||||
|
||||
if (booking.isGovernment) {
|
||||
await this.allocate(scheduleId, booking, 'gov');
|
||||
} else {
|
||||
await this.reserve(booking);
|
||||
armed = true;
|
||||
}
|
||||
remaining -= need;
|
||||
if (remaining <= 0) break;
|
||||
}
|
||||
|
||||
if (remaining <= 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: number,
|
||||
remaining: number,
|
||||
): Promise<number> {
|
||||
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),
|
||||
);
|
||||
|
||||
for (const victim of candidates) {
|
||||
if (need <= remaining) 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);
|
||||
remaining += this.wagonsFor(victim);
|
||||
}
|
||||
return remaining;
|
||||
}
|
||||
|
||||
// ---- 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);
|
||||
}
|
||||
|
||||
/** 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
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user