edr-platform/apps/edr-freight-api/src/modules/train-scheduling/booking-batch.service.ts

import {
  BadRequestException,
  ConflictException,
  forwardRef,
  Inject,
  Injectable,
  Logger,
  NotFoundException,
  OnModuleInit,
  Optional,
} from '@nestjs/common';
import { InjectDataSource } from '@nestjs/typeorm';
import { SchedulerRegistry } from '@nestjs/schedule';
import {
  Between,
  DataSource,
  FindOptionsWhere,
  ILike,
  In,
  LessThanOrEqual,
  MoreThanOrEqual,
} from 'typeorm';

import { Booking } from '../bookings/entities/booking.entity';
import { BookingsRepository } from '../bookings/bookings.repository';
import { BookingPricingService } from '../bookings/booking-pricing.service';
import { formatRouteLabel } from '../routes/entities/route.entity';
import { RouteMilestone } from '../routes/entities/route-milestone.entity';
import { Yard } from '../rule-engine/entities/yard.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 { BookingNotifierService } from './booking-notifier.service';
import { TrainSchedulingService } from './train-scheduling.service';
import { eatDay } from './batch-window.util';
import {
  BATCH_BOARD_STATUSES,
  BatchBoardQueryDto,
} from './dto/batch-board-query.dto';
import {
  Freight,
  PaginatedResponse,
  TrainScheduleStatus as TrainScheduleStatusEnum,
} from "@edr/types";
import { BillingService } from "../billing/billing.service";
import {
  buildPaginationMeta,
  normalizePagination,
} from '../../common/utils/pagination.util';


import {
  DEFAULT_BULK_WAGON_CAPACITY_TONS,
  DEFAULT_BULK_WAGON_LENGTH_METERS,
  DEFAULT_BULK_WAGON_TARE_TONS,
  DEFAULT_CONTAINER_WAGON_CAPACITY_TONS,
  DEFAULT_CONTAINER_WAGON_LENGTH_METERS,
  DEFAULT_CONTAINER_WAGON_TARE_TONS,
  DEFAULT_WAGONS_PER_BOOKING,
} from "./booking-batch.constants";
import {
  LocomotiveLimits,
  WagonTypeDimensions,
  bookingCargoTons,
  bookingGrossWeightTons,
  deriveTrainCapacityFromLocomotive,
  sizePartialOfferWagons,
  trainHardCaps,
  trainSetLocomotiveLimits,
  wagonTypeDimensionsFromEntity,
} from './train-capacity.util';
import { WagonType } from '../wagon-types/entities/wagon-type.entity';
import { Wagon } from '../wagons/entities/wagon.entity';
import { ClearanceMilestoneService } from '../contracts/clearance-milestone.service';
import { BookingSplitService } from './booking-split.service';
import { RemainderPlacementService } from './remainder-placement.service';
import { BookingWindowGateway } from './booking-window.gateway';
import {
  MAX_TEU_SLOTS_PER_WAGON,
  containerWagonsForLines,
} from './wagon-plan.util';
import {
  Capacity,
  CorridorBudget,
  CorridorLeg,
  OverageTolerance,
  stopYardsFor,
} from './corridor-capacity.util';

export type { Capacity } from './corridor-capacity.util';

/**
 * A train's fill limits: the base caps the corridor budget spends from, plus
 * the locomotive overage tolerance spendable only on whole-booking admission.
 */
type TrainLimits = { base: Capacity; tolerance: OverageTolerance };

/**
 * Result of the export whole-booking single-train space check. `scheduleId`
 * is the earliest fillable train that carries the whole booking, or null when
 * none can — then `bestAvailable` reports the largest single-train leftover
 * in the booking's own units and `fullMessage` is the customer-facing copy.
 */
export interface ExportSpaceReport {
  scheduleId: string | null;
  trainsForDay: boolean;
  corridorMatched: boolean;
  need: Capacity;
  bestAvailable: { wagons: number; cargoTons: number } | null;
  fullMessage: string | null;
}

/** 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;
}

/** One wagon type's footprint: its length on the train, the tare it adds to the
 * locomotive's gross load, and the payload it carries. */
type PerWagonDims = { lengthMeters: number; tareWeightTons: number; capacityTons: number };

/**
 * Wagon dimensions used to size a booking's capacity draw. `byWagonTypeId` holds
 * every wagon type so a booking is measured on the type its cargo/container type
 * actually rides (the same FK resolution allocation uses); `container`/`bulk` are
 * representative fallbacks for bookings whose type has no wagon type configured.
 */
type WagonDims = {
  container: PerWagonDims;
  bulk: PerWagonDims;
  byWagonTypeId: Map<string, PerWagonDims>;
};

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;
  /** Rule-engine priority score used to rank the batch (higher = boards first). */
  priorityScore: number;
  /** CONTAINER | BULK — for the priority-tracking visuals. */
  freightType: string | null;
}

export type BookingAllocationStatus =
  | "NOT_ATTEMPTED"
  | "ASSIGNED"
  | "DEFERRED"
  | "FAILED";

export interface BatchBoardBookingDetail extends BatchBoardBooking {
  fullyExecutedAt: string | null;
  selectedForBatchAt: string | null;
  allocationStatus: BookingAllocationStatus;
  allocationIssue: string | null;
  /** Set when this booking shares a wagon with a consolidation partner. */
  consolidationPartnerId: string | null;
  consolidationPartnerRef: string | null;
}

export interface BatchBoardCounts {
  allocated: number;
  selectedForBatch: number;
  ready: number;
  waiting: number;
  expired: number;
  pendingContract: number;
}

/** A booking bucket on the detail board (in-window vs pending-contract). */
export interface BatchBoardBucket {
  counts: BatchBoardCounts;
  bookings: BatchBoardBookingDetail[];
}

export interface BatchBoardScheduleDetail {
  scheduleId: string;
  /** Human-facing schedule reference (S-YYYY-NNNNN). */
  scheduleReference: string | null;
  trainNumber: string | null;
  routeName: string | null;
  origin: string | null;
  destination: string | null;
  scheduleDate: string | null;
  status: string;
  bookingWindowStatus: string;
  direction: string | null;
  windowPhase: string | null;
  windowOpensAt: string | null;
  windowClosesAt: string | null;
  docReviewEndsAt: string | null;
  paymentPhaseEndsAt: string | null;
  bookingCycleNo: number;
  /** Built train (Train Builder) behind this departure, when scheduled by train. */
  train: BatchBoardSchedule["train"];
  locomotive: BatchBoardSchedule["locomotive"];
  capacity: BatchBoardSchedule["capacity"];
  counts: BatchBoardSchedule["counts"];
  /** Bookings inside the schedule's booking window (fully-executed contracts). */
  bookings: BatchBoardBookingDetail[];
  pendingContract: BatchBoardBucket;
  allocationViolations: string[];
}

export interface BatchBoardSchedule {
  scheduleId: string;
  /** Human-facing schedule reference (S-YYYY-NNNNN). */
  scheduleReference: string | null;
  trainNumber: string | null;
  routeName: string | null;
  origin: string | null;
  destination: string | null;
  scheduleDate: string | null;
  createdAt: string | null;
  status: string;
  bookingWindowStatus: string;
  direction: string | null;
  windowPhase: string | null;
  windowOpensAt: string | null;
  windowClosesAt: string | null;
  docReviewEndsAt: string | null;
  paymentPhaseEndsAt: string | null;
  bookingCycleNo: number;
  /** Built train (Train Builder) behind this departure, when scheduled by train. */
  train: {
    id: string;
    code: string;
    trainName: string | null;
  } | null;
  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;
    /** Wagon-slot cap for the train (locomotive/wagon-type derived). */
    maxWagons: number | null;
  };
  counts: {
    allocated: number;
    selectedForBatch: number;
    ready: number;
    waiting: number;
    pendingContract: number;
    expired: number;
  };
  bookings: BatchBoardBooking[];
}

/** Paginated batch-board list in the shared `{items, meta}` envelope — the API
 * response wrapper already uses `data`, and the frontend's unwrap() strips one
 * `data` level. */
export type BatchBoardListResponse = PaginatedResponse<BatchBoardSchedule>;

/**
 * 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);

  /**
   * Serialises settle/top-up per schedule. The PAYMENT phase transition and the
   * tick's overdue backstop both call settleDueReservations for the same schedule
   * in the same second; without this they interleave and the top-up runs against a
   * schedule whose phase has already been concluded.
   */
  private readonly scheduleLocks = new Map<string, Promise<void>>();

  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,
    // forwardRef: TrainSchedulingService injects this service back (window
    // refresh after adjust-consist), so the classes load in a cycle.
    @Inject(forwardRef(() => TrainSchedulingService))
    private readonly trainSchedulingService: TrainSchedulingService,
    private readonly billing: BillingService,
    private readonly bookingWindowGateway: BookingWindowGateway,
    @Inject(forwardRef(() => BookingPricingService))
    private readonly pricingService: BookingPricingService,

    @Optional() private readonly milestoneService?: ClearanceMilestoneService,
    @Optional() private readonly splitService?: BookingSplitService,
    @Optional()
    @Inject(forwardRef(() => RemainderPlacementService))
    private readonly remainderPlacement?: RemainderPlacementService,

  ) {}

  /**
   * Auto-place a paid booking's split remainder onto the next fitting train.
   * Gated so it can ship dark: off unless FREIGHT_AUTO_REMAINDER=true.
   */
  private get autoRemainderEnabled(): boolean {
    return process.env.FREIGHT_AUTO_REMAINDER === "true";
  }

  /**
   * Let EXPORT bookings split (offer the largest fitting part, leftover rebooks
   * on the next train). Separate flag from auto-remainder: export touches the
   * FCFS money path, so partial-offer can be enabled independently.
   */
  private get exportSplitEnabled(): boolean {
    return process.env.FREIGHT_EXPORT_SPLIT === "true";
  }

  /** On boot, reconcile OPEN route-days and re-arm settle timers. */
  async onModuleInit(): Promise<void> {
    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<void> {
    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<void> {
    this.logger.log(
      `[BATCH] processRouteDay START ${group.originYardId}->${group.destinationYardId} ${group.day}`,
    );
    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<void> {
    await this.fillSchedule(scheduleId);
    await this.settleDueReservations(scheduleId);
    await this.reconcilePaidUnlinked(scheduleId);
    await this.trainSchedulingService.tryAutoWagonAllocation(scheduleId);
  }

  /**
   * Distinct (origin, destination, EAT day) groups across LEGACY OPEN schedules —
   * schedules with a `windowPhase` are driven exclusively by the window engine
   * (BookingWindowService), never by the periodic legacy fill.
   */
  private async openRouteDayGroups(): Promise<RouteDayGroup[]> {
    const open = (
      await this.trainSchedulesRepository.findAll({
        where: [
          { bookingWindowStatus: "OPEN", status: TrainScheduleStatusEnum.Draft },
          { bookingWindowStatus: "OPEN", status: TrainScheduleStatusEnum.Scheduled },
        ],
      })
    ).filter((s) => s.windowPhase == null);
    const groups = new Map<string, RouteDayGroup>();
    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<void> {
    const booking = await this.dataSource.getRepository(Booking).findOne({
      where: { id: bookingId },
      relations: { company: true },
    });
    if (!booking) return;
    if (!booking.trainScheduleId) {
      // A paid booking with no train is money taken and nothing boarding —
      // scream so staff pin it to a schedule manually (batch board / assign).
      if (booking.paymentStatus === "PAID" || booking.status === "PAID") {
        this.logger.error(
          `PAID booking ${booking.reference ?? bookingId} has no train_schedule_id — ` +
            `its reservation was likely expired before the payment landed. ` +
            `Assign it to a schedule manually from the batch board.`,
        );
      }
      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" });
    }

    // Paying inside the window accepts an open partial offer — reduce the booking
    // to the offered part before it boards (remainder returns to the contract cap).
    if (this.splitService) {
      await this.splitService.applySplit(bookingId);

      // The split only happens on payment (here) — so auto-placing the remainder
      // also only happens once the customer has accepted+paid. Re-read to see if
      // applySplit actually reduced this booking (an open offer existed); if so,
      // auto-create + place the remainder booking on the next fitting train.
      // applySplit committed its own transaction before returning, so this reads
      // the reduced lines. Best-effort: a placement failure never blocks the
      // paid booking from boarding — the remainder falls back to manual rebook.
      if (this.autoRemainderEnabled && this.remainderPlacement) {
        const split = await this.dataSource
          .getRepository(Booking)
          .findOne({ where: { id: bookingId } });
        // Export remainders only auto-place when export split is on — otherwise
        // an export booking never splits in the first place.
        const directionOn =
          split?.tradeDirection !== "EXPORT" || this.exportSplitEnabled;
        if (split?.isSplit && directionOn) {
          await this.remainderPlacement
            .placeRemainder(split)
            .catch((err) =>
              this.logger.error(
                `Auto-place remainder failed for ${split.reference}: ${
                  err instanceof Error ? err.message : String(err)
                }`,
              ),
            );
        }
      }
    }

    const linked =
      await this.trainScheduleBookingsRepository.existsForBooking(bookingId);
    if (!linked) {
      if (await this.holdIfWagonShort(booking.trainScheduleId, booking)) return;
      await this.allocate(booking.trainScheduleId, booking, "paid");
      this.logger.log(
        `Linked PAID booking ${booking.reference ?? bookingId} to schedule ${booking.trainScheduleId}`,
      );
    } else {
      // Already linked at booking time (export FCFS: the customer books a
      // specific train, so allocate() ran up front). allocate() is where the
      // payment-settled tracking milestones are written, so on this branch we
      // record them here — otherwise a paid, already-linked booking leaves
      // FREIGHT_PAYMENT_SETTLED stuck PENDING and the clearance step never ticks.
      void this.completeTrackingMilestones(bookingId, [
        "WAGON_REQUESTED",
        "FREIGHT_PAYMENT_PENDING",
        "FREIGHT_PAYMENT_SETTLED",
      ]);
      void this.markWagonAllocatedMilestone(bookingId);
    }

    const schedule = await this.trainSchedulesRepository.findByIdWithFullGraph(
      booking.trainScheduleId,
    );
    if (schedule && (await this.isTrainFull(schedule))) {
      await this.setWindow(booking.trainScheduleId, "FULL");
      // This payment may have been the last live pay window on a now-full
      // export day — the settle that normally re-runs the sweep finds nothing
      // left to settle, so trigger it here.
      void this.expireLeftoverExportDay(booking.trainScheduleId);
    }

    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}`,
      );
    }
    this.notifyBoardChanged(booking.trainScheduleId, "booking_paid_allocated");
  }

  /** Customer paid — delegate to ensurePaidBookingAllocated. */
  async confirmPaidAndAllocate(bookingId: string): Promise<void> {
    await this.ensurePaidBookingAllocated(bookingId);
  }

  /** Open partial-capacity offer summary for booking detail payloads (null when none). */
  async getOpenOfferSummary(bookingId: string): Promise<{
    offeredWagons: number;
    totalWagons: number;
    offeredAmount: number;
    paymentDeadline: Date;
  } | null> {
    if (!this.splitService) return null;
    const offer = await this.splitService.findOpenOffer(bookingId);
    if (!offer) return null;
    return {
      offeredWagons: offer.offeredWagons,
      totalWagons: offer.totalWagons,
      offeredAmount: Number(offer.offeredAmount),
      paymentDeadline: offer.paymentDeadline,
    };
  }

  // ---- export FCFS -----------------------------------------------------------

  /**
   * Whole-booking single-train space report for an EXPORT booking. Export
   * bookings never split — the entire booking must ride ONE train, so the
   * report scans every fillable export train on the booking's corridor/day
   * (earliest first) for one whose remaining budget fits the whole need. When
   * none fits, `bestAvailable` carries the largest single-train leftover
   * converted into the booking's own units (base caps, no overage tolerance)
   * so the customer can be told exactly how much he COULD book on that day.
   */
  async exportSpaceReport(
    booking: Booking,
    need?: Capacity,
  ): Promise<ExportSpaceReport> {
    if (!booking.scheduledDate) {
      throw new BadRequestException('Booking has no scheduled date');
    }
    const day = eatDay(new Date(booking.scheduledDate));
    // Corridor-aware: any train whose route carries the booking's origin
    // strictly before its destination qualifies — a Dire→Djibouti booking may
    // ride an Addis→…→Djibouti train. The leg check below (legOf) enforces the
    // stop order, so we fetch the day's open trains without endpoint filters.
    const corridor = await this.trainSchedulesRepository.findAll({
      where: [
        { status: TrainScheduleStatusEnum.Draft },
        { status: TrainScheduleStatusEnum.Scheduled },
      ],
    });
    const candidates = corridor
      .filter(
        (s) =>
          s.scheduledDepartureDate != null &&
          eatDay(s.scheduledDepartureDate) === day &&
          this.isFillable(s),
      )
      .sort(
        (a, b) =>
          a.scheduledDepartureDate.getTime() - b.scheduledDepartureDate.getTime(),
      );

    const wagonDims = await this.loadWagonDims();
    const required = need ?? this.needFor(booking, wagonDims);
    const dims = this.dimsFor(booking, wagonDims);
    const report: ExportSpaceReport = {
      scheduleId: null,
      trainsForDay: candidates.length > 0,
      corridorMatched: false,
      need: required,
      bestAvailable: null,
      fullMessage: null,
    };

    for (const candidate of candidates) {
      const schedule = await this.trainSchedulesRepository.findByIdWithFullGraph(
        candidate.id,
      );
      const locomotive = trainSetLocomotiveLimits(schedule?.trainSet);
      if (!schedule || !locomotive) continue;
      const limits = await this.capacityLimits(locomotive);
      const budget = await this.remainingBudget(schedule, limits, wagonDims);
      const leg = budget.legOf(booking.originYardId, booking.destinationYardId);
      if (!leg) continue; // this train's route doesn't carry the booking's leg
      report.corridorMatched = true;
      if (budget.fits(required, leg)) {
        // Earliest fitting train wins — no need to keep sizing leftovers.
        report.scheduleId = schedule.id;
        return report;
      }
      const available = this.bookableWithin(budget.remainingFor(leg), dims);
      if (
        !report.bestAvailable ||
        available.cargoTons > report.bestAvailable.cargoTons ||
        (available.cargoTons === report.bestAvailable.cargoTons &&
          available.wagons > report.bestAvailable.wagons)
      ) {
        report.bestAvailable = available;
      }
    }

    report.fullMessage = this.exportFullMessage(booking, report);
    return report;
  }

  /**
   * Largest booking (in the requester's own wagon-type units) that a single
   * train's leftover base capacity could still admit: bounded by free wagon
   * slots, free train length, and the locomotive's remaining pull weight
   * (gross — each wagon's tare eats into it before any cargo does).
   */
  private bookableWithin(
    remaining: Capacity,
    dims: PerWagonDims,
  ): { wagons: number; cargoTons: number } {
    const byLength =
      dims.lengthMeters > 0
        ? Math.floor(Math.max(0, remaining.lengthMeters) / dims.lengthMeters)
        : Math.floor(Math.max(0, remaining.wagons));
    const maxWagons = Math.max(
      0,
      Math.min(Math.floor(Math.max(0, remaining.wagons)), byLength),
    );
    let bestTons = 0;
    let usableWagons = 0;
    for (let w = 1; w <= maxWagons; w++) {
      if (w * dims.tareWeightTons > remaining.weightTons) break;
      usableWagons = w;
      const tons = Math.min(
        w * dims.capacityTons,
        remaining.weightTons - w * dims.tareWeightTons,
      );
      if (tons > bestTons) bestTons = tons;
    }
    return {
      wagons: usableWagons,
      cargoTons: Math.max(0, Math.floor(bestTons * 1000) / 1000),
    };
  }

  /** Customer-facing "train is full" copy carrying the bookable leftover. */
  private exportFullMessage(booking: Booking, report: ExportSpaceReport): string {
    if (!report.trainsForDay || !report.corridorMatched) {
      return 'No export train is accepting bookings for this day';
    }
    const best = report.bestAvailable;
    const base =
      'Not enough train space — an export booking must ride a single train whole, ' +
      'and no open train on this day can carry it. ';
    if (!best || best.wagons <= 0) {
      return base + 'No capacity is left on this day — pick another shipment day.';
    }
    if (booking.freightType === 'BULK') {
      return (
        base +
        `The largest remaining space is about ${best.cargoTons} tons ` +
        `(${best.wagons} wagon${best.wagons === 1 ? '' : 's'}) — book up to that amount or pick another day.`
      );
    }
    return (
      base +
      `The largest remaining space is ${best.wagons} wagon${best.wagons === 1 ? '' : 's'} ` +
      `(up to ${best.wagons * 2} × 20ft or ${best.wagons} × 40ft, weight permitting) — ` +
      'reduce the booking or pick another day.'
    );
  }

  /**
   * Export is first-come-first-serve: no window cycle, no priority, no batch.
   * Pick the earliest open export train on the booking's corridor/day that still
   * fits the booking. Throws ConflictException when every train is full — the
   * staff accept fails and no more export bookings are taken.
   */
  async pickExportSchedule(booking: Booking, need?: Capacity): Promise<string> {
    const report = await this.exportSpaceReport(booking, need);
    if (report.scheduleId) return report.scheduleId;
    throw new ConflictException(
      report.fullMessage ?? 'Train is full — no export capacity left for this day',
    );
  }

  /**
   * Trains that can carry a booking's leg on a given day, earliest departure
   * first, each with the largest number of wagons it could still admit for the
   * booking's wagon type. Direction-filtered: EXPORT bookings see export trains,
   * IMPORT/DOMESTIC see non-export trains. Measures against the booking's FULL
   * allowed wagon-type set ({@link dimsForAllowed}) so a train stocking a
   * non-primary allowed type still counts. The remainder placer uses this to
   * pick the next fitting train; the `free` wagon count is the best across the
   * allowed types (a train fits under whichever allowed type gives most room).
   */
  async fittingTrainsForDay(
    booking: Booking,
    day: string,
    direction: "IMPORT" | "EXPORT",
  ): Promise<Array<{ scheduleId: string; departure: Date; freeWagons: number }>> {
    const corridor = await this.trainSchedulesRepository.findAll({
      where: [
        { status: TrainScheduleStatusEnum.Draft },
        { status: TrainScheduleStatusEnum.Scheduled },
      ],
    });
    const candidates = corridor
      .filter(
        (s) =>
          s.scheduledDepartureDate != null &&
          eatDay(s.scheduledDepartureDate) === day &&
          s.bookingWindowStatus !== "FULL" &&
          (direction === "EXPORT"
            ? s.direction === "EXPORT"
            : s.direction !== "EXPORT"),
      )
      .sort(
        (a, b) =>
          a.scheduledDepartureDate!.getTime() -
          b.scheduledDepartureDate!.getTime(),
      );

    const wagonDims = await this.loadWagonDims();
    const dimsOptions = this.dimsForAllowed(booking, wagonDims);
    const out: Array<{ scheduleId: string; departure: Date; freeWagons: number }> = [];

    for (const candidate of candidates) {
      const schedule = await this.trainSchedulesRepository.findByIdWithFullGraph(
        candidate.id,
      );
      const locomotive = trainSetLocomotiveLimits(schedule?.trainSet);
      if (!schedule || !locomotive) continue;
      const limits = await this.capacityLimits(locomotive);
      const budget = await this.remainingBudget(schedule, limits, wagonDims);
      const leg = budget.legOf(booking.originYardId, booking.destinationYardId);
      if (!leg) continue; // this train's route doesn't carry the booking's leg
      const room = budget.remainingFor(leg);
      // Best usable wagons across the allowed types — a train fits under
      // whichever configured wagon type gives it the most room.
      let freeWagons = 0;
      for (const dims of dimsOptions) {
        const w = this.bookableWithin(room, dims).wagons;
        if (w > freeWagons) freeWagons = w;
      }
      if (freeWagons > 0) {
        out.push({
          scheduleId: schedule.id,
          departure: schedule.scheduledDepartureDate!,
          freeWagons,
        });
      }
    }
    return out;
  }

  /**
   * Advisory free-wagon count for an IMPORT/DOMESTIC booking on a given day,
   * summed across every train on the booking's corridor that day. Unlike the
   * export gate this does NOT block and does NOT first-fit a single train:
   * import is batched and splittable, so the honest number a customer can plan
   * against is the TOTAL room across the day's trains for the booking's wagon
   * type, in that type's own wagon units.
   *
   * It deliberately skips the `isFillable` window-phase gate. A customer picks a
   * shipment day while its window is still OPEN (or pre-window) — the batch fill
   * only makes those trains fillable after the window closes — so gating on the
   * fill phase here would report 0 for exactly the days customers are choosing.
   * We therefore count any non-FULL train that carries the leg, netting out the
   * capacity already consumed by allocated + live-reserved bookings
   * (`remainingBudget`). The count is an upper bound: the batch engine may still
   * split the booking across trains or defer a remainder to a later window.
   */
  async dayImportAvailability(
    booking: Booking,
    day: string,
  ): Promise<{ freeWagons: number; need: number; trainsForDay: boolean }> {
    const corridor = await this.trainSchedulesRepository.findAll({
      where: [
        { status: TrainScheduleStatusEnum.Draft },
        { status: TrainScheduleStatusEnum.Scheduled },
      ],
    });
    const candidates = corridor.filter(
      (s) =>
        s.scheduledDepartureDate != null &&
        eatDay(s.scheduledDepartureDate) === day &&
        s.bookingWindowStatus !== 'FULL' &&
        s.direction !== 'EXPORT',
    );

    const wagonDims = await this.loadWagonDims();
    const dims = this.dimsFor(booking, wagonDims);
    const need = this.wagonsFor(booking, wagonDims);
    let freeWagons = 0;
    let trainsForDay = false;

    for (const candidate of candidates) {
      const schedule = await this.trainSchedulesRepository.findByIdWithFullGraph(
        candidate.id,
      );
      const locomotive = trainSetLocomotiveLimits(schedule?.trainSet);
      if (!schedule || !locomotive) continue;
      const limits = await this.capacityLimits(locomotive);
      const budget = await this.remainingBudget(schedule, limits, wagonDims);
      const leg = budget.legOf(booking.originYardId, booking.destinationYardId);
      if (!leg) continue; // this train's route doesn't carry the booking's leg
      trainsForDay = true;
      freeWagons += this.bookableWithin(budget.remainingFor(leg), dims).wagons;
    }

    return { freeWagons, need, trainsForDay };
  }

  /**
   * Export split: no single train carries the whole booking, so offer the
   * largest fitting part on the export train with the most room for its leg.
   * Returns true when an offer was opened (the caller must NOT then reserve —
   * the offer already opened its own pay window), false when the booking fits
   * whole somewhere (normal FCFS path) or no meaningful partial exists.
   *
   * Only the offer is written here: the booking is reduced to the offered part
   * on payment (applySplit), and the leftover is auto-placed afterwards. So an
   * unpaid export booking stays whole and the customer may still cancel it.
   */
  private async tryExportPartialOffer(booking: Booking): Promise<boolean> {
    if (!this.splitService) return false;
    const report = await this.exportSpaceReport(booking);
    // A train fits it whole — nothing to split, take the normal path.
    if (report.scheduleId) return false;
    if (!report.bestAvailable || report.bestAvailable.wagons < 1) return false;

    if (!booking.scheduledDate) return false;
    const day = eatDay(new Date(booking.scheduledDate));
    const fitting = await this.fittingTrainsForDay(booking, day, "EXPORT");
    if (!fitting.length) return false;
    // Most room first — the largest single part ships now, the smallest leftover
    // is what has to find another train.
    const target = [...fitting].sort((a, b) => b.freeWagons - a.freeWagons)[0];

    const schedule = await this.trainSchedulesRepository.findByIdWithFullGraph(
      target.scheduleId,
    );
    const locomotive = trainSetLocomotiveLimits(schedule?.trainSet);
    if (!schedule || !locomotive) return false;
    const wagonDims = await this.loadWagonDims();
    const limits = await this.capacityLimits(locomotive);
    const budget = await this.remainingBudget(schedule, limits, wagonDims);
    const leg = budget.legOf(booking.originYardId, booking.destinationYardId);
    if (!leg) return false;

    const offered = await this.tryPartialOffer(
      booking,
      schedule.id,
      budget.remainingFor(leg),
      report.need,
    );
    if (!offered) return false;
    this.logger.log(
      `[EXPORT SPLIT] offered partial to ${booking.reference} on schedule ` +
        `${schedule.id} — leftover rebooks on the next train once paid.`,
    );
    this.notifyBoardChanged(schedule.id, "batch_fill");
    return true;
  }

  /**
   * Accept an export booking into the FCFS flow. Solo bookings reserve immediately.
   * A consolidated booking reserves as a pair only once BOTH partners are ready
   * (FULLY_EXECUTED): the second partner's accept triggers the pair reservation
   * against the combined shared-wagon need; the first partner's accept just waits.
   * Throws ConflictException (before this booking is persisted-ready) when there is
   * no export capacity for the day, so staff accept fails.
   */
  async acceptExportBooking(booking: Booking): Promise<void> {
    const partnerId = booking.consolidationPartnerId ?? null;
    if (!partnerId) {
      // Export split: when no single train carries the whole booking, offer the
      // largest fitting part instead of failing the accept. The customer pays
      // that part; on payment applySplit reduces this booking to it and the
      // leftover is auto-placed as its own booking on the next train. Pairs are
      // excluded (handled below) — a shared wagon is never split.
      if (this.exportSplitEnabled && this.isSplitEligible(booking, false)) {
        const offered = await this.tryExportPartialOffer(booking);
        if (offered) return;
      }
      const scheduleId = await this.pickExportSchedule(booking);
      await this.reserveOnExport([booking], scheduleId);
      return;
    }

    const partner = await this.dataSource
      .getRepository(Booking)
      .findOne({
        where: { id: partnerId },
        relations: {
          company: true,
          bookingContainers: { containerType: true },
          cargoType: true,
        },
      });
    // Partner not yet accepted → this booking is now FULLY_EXECUTED and simply
    // waits; the partner's later accept will reserve the pair.
    if (!partner || partner.status !== 'FULLY_EXECUTED') {
      return;
    }
    const wagonDims = await this.loadWagonDims();
    const need = this.combinedNeed(booking, partner, wagonDims);
    const scheduleId = await this.pickExportSchedule(booking, need);
    await this.reserveOnExport([booking, partner], scheduleId);
  }

  /** Reserve one or two (consolidated) export bookings on a train and open pay windows. */
  private async reserveOnExport(
    bookings: Booking[],
    scheduleId: string,
  ): Promise<void> {
    // H8: the capacity check (pickExportSchedule → budget.fits) and the
    // reservation writes below are not atomic on their own — two concurrent
    // export accepts can each see the same train as fitting and both reserve,
    // overshooting the train's capacity. Serialize reservations against this
    // schedule: take a pessimistic_write lock on the TrainSchedule row
    // (SELECT … FOR UPDATE), then RE-VERIFY budget.fits for these bookings'
    // combined need from freshly-committed state INSIDE the lock before the
    // reserve writes run. A loser (another accept took the space first) gets a
    // ConflictException — the staff accept fails and reverts, exactly as an
    // up-front full train does. Covered: the fits-vs-reserve overshoot on the
    // export FCFS path; the lock is held for the duration of the reserve writes.
    await this.dataSource.transaction(async (manager) => {
      const locked = await manager.findOne(TrainSchedule, {
        where: { id: scheduleId },
        lock: { mode: "pessimistic_write" },
      });
      if (!locked) {
        throw new ConflictException(
          "Export train is no longer available for reservation",
        );
      }
      const schedule =
        await this.trainSchedulesRepository.findByIdWithFullGraph(scheduleId);
      const locomotive = trainSetLocomotiveLimits(schedule?.trainSet);
      if (!schedule || !locomotive) {
        throw new ConflictException(
          "Export train is no longer available for reservation",
        );
      }
      const wagonDims = await this.loadWagonDims();
      const limits = await this.capacityLimits(locomotive);
      const budget = await this.remainingBudget(schedule, limits, wagonDims);
      const leg = budget.legOf(
        bookings[0].originYardId,
        bookings[0].destinationYardId,
      );
      const need =
        bookings.length >= 2
          ? this.combinedNeed(bookings[0], bookings[1], wagonDims)
          : this.needFor(bookings[0], wagonDims);
      if (!leg || !budget.fits(need, leg)) {
        throw new ConflictException(
          "Train is full — no export capacity left for this day",
        );
      }

      for (const b of bookings) await this.reserve(b, scheduleId);
    });
    this.armSettle(scheduleId);
    const schedule =
      await this.trainSchedulesRepository.findByIdWithFullGraph(scheduleId);
    if (schedule && (await this.isTrainFull(schedule))) {
      await this.setWindow(scheduleId, 'FULL');
    }
    this.notifyBoardChanged(scheduleId, 'export_booking_accepted');
  }

  /** Link PAID bookings that have no train_schedule_bookings row (cron backstop). */
  async reconcilePaidUnlinked(scheduleId: string): Promise<void> {
    const unlinked =
      await this.bookingsRepository.findPaidUnlinkedForSchedule(scheduleId);
    for (const booking of unlinked) {
      // Held on purpose (paid, no wagon free) — the cron must not undo it.
      if (booking.schedulingStatus === "WAITING_FOR_WAGON") continue;
      if (await this.holdIfWagonShort(scheduleId, booking)) continue;
      await this.allocate(scheduleId, booking, "paid");
      this.logger.log(
        `Reconciled PAID booking ${booking.reference ?? booking.id} → schedule ${scheduleId}`,
      );
    }
    if (unlinked.length > 0) {
      this.notifyBoardChanged(scheduleId, "paid_reconciled");
    }
  }

  // ---- legacy fill entry point ----------------------------------------------

  /**
   * Legacy periodic fill for schedules without a window phase (DOMESTIC and
   * pre-migration trains). Invoked by BookingWindowService's tick — the old
   * standalone cron was replaced by the window engine.
   */
  async runBatchFill(): Promise<void> {
    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 import schedule — including
   * dispatched, arrived and cancelled history — with its locomotive, capacity
   * usage and its bookings grouped by lifecycle state (allocated / awaiting
   * payment / paid-waiting / pending contract / expired). Paginated and
   * filterable; per-schedule booking summaries are only computed for the
   * requested page.
   */
  async getBatchBoard(
    query: BatchBoardQueryDto = {},
  ): Promise<BatchBoardListResponse> {
    // Board cards are heavy (per-schedule booking summaries), so the default
    // page is smaller than the toolkit-wide 20.
    const { page, pageSize, skip, take } = normalizePagination(query, {
      defaultPageSize: 12,
    });

    // Status filter: any subset of the lifecycle. Omitted = all statuses, so
    // arrived / cancelled / dispatched schedules stay visible as history.
    const allowedStatuses = new Set<string>(BATCH_BOARD_STATUSES);
    const statuses = (query.statuses ?? "")
      .split(",")
      .map((v) => v.trim().toUpperCase())
      .filter((v) => allowedStatuses.has(v));

    const dateRange = (from?: string, to?: string) => {
      const f = from ? new Date(from) : null;
      const t = to ? new Date(to) : null;
      if (f && t) return Between(f, t);
      if (f) return MoreThanOrEqual(f);
      if (t) return LessThanOrEqual(t);
      return undefined;
    };

    // Batch board is IMPORT-only: export is FCFS with no batch/priority calc,
    // and domestic/legacy schedules run the legacy fill, not the window batch.
    const base: FindOptionsWhere<TrainSchedule> = { direction: "IMPORT" };
    if (statuses.length) base.status = In(statuses) as never;
    if (query.bookingWindowStatus) {
      base.bookingWindowStatus = query.bookingWindowStatus;
    }
    const departure = dateRange(query.departureFrom, query.departureTo);
    if (departure) base.scheduledDepartureDate = departure as never;
    const created = dateRange(query.createdFrom, query.createdTo);
    if (created) base.createdAt = created as never;

    // Search fans out across every human-recognizable label. Each OR variant
    // repeats the base filters so the search never widens them.
    const term = query.search?.trim();
    let where: FindOptionsWhere<TrainSchedule> | FindOptionsWhere<TrainSchedule>[] =
      base;
    if (term) {
      const like = ILike(`%${term}%`);
      where = [
        { ...base, trainNumber: like as never },
        { ...base, originStation: { label: like } },
        { ...base, destinationStation: { label: like } },
        { ...base, route: { originYard: { label: like } } },
        { ...base, route: { destinationYard: { label: like } } },
        { ...base, trainSet: { locomotive: { code: like } } },
      ] as FindOptionsWhere<TrainSchedule>[];
    }

    const sortBy = query.sortBy ?? "createdAt";
    const sortOrder = query.sortOrder ?? "DESC";

    const [schedules, total] = await this.trainSchedulesRepository.findAndCount({
      where,
      relations: {
        trainSet: { locomotive: true, train: true },
        originStation: true,
        destinationStation: true,
        // Yards supply the route's display name for `routeName` below;
        // milestones (with yards) give it the full corridor path.
        route: { originYard: true, destinationYard: true, milestones: { yard: true } },
      },
      order: { [sortBy]: sortOrder } as never,
      skip,
      take,
    });

    const wagonDims = await this.loadWagonDims();
    const linkRepo = this.dataSource.getRepository(TrainScheduleBooking);

    // One links query + one bookings query for the whole page (was 2 per card).
    const scheduleIds = schedules.map((s) => s.id);
    const [allLinks, allBookings] = await Promise.all([
      scheduleIds.length
        ? linkRepo.find({ where: { trainScheduleId: In(scheduleIds) } })
        : Promise.resolve([]),
      this.bookingsRepository.findAllBySchedules(scheduleIds),
    ]);
    const linkedIdsBySchedule = new Map<string, Set<string>>();
    for (const l of allLinks) {
      let set = linkedIdsBySchedule.get(l.trainScheduleId);
      if (!set) linkedIdsBySchedule.set(l.trainScheduleId, (set = new Set()));
      set.add(l.bookingId);
    }
    const bookingsBySchedule = new Map<string, Booking[]>();
    for (const b of allBookings) {
      if (!b.trainScheduleId) continue;
      let list = bookingsBySchedule.get(b.trainScheduleId);
      if (!list) bookingsBySchedule.set(b.trainScheduleId, (list = []));
      list.push(b);
    }

    const board: BatchBoardSchedule[] = [];
    for (const s of schedules) {
      const linkedIds = linkedIdsBySchedule.get(s.id) ?? new Set<string>();
      const bookings = bookingsBySchedule.get(s.id) ?? [];

      const items: BatchBoardBooking[] = bookings.map((b) => {
        const need = this.needFor(b, wagonDims);
        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)),
          priorityScore: Number(b.priorityScore ?? 0),
          freightType: b.freightType ?? null,
        };
      });

      board.push(this.buildScheduleSummary(s, items));
    }

    return { items: board, meta: buildPaginationMeta(total, page, pageSize) };
  }

  /** Schedule-level batch board: the schedule's own booking window plus its
   * bookings split into in-window (contract executed) vs pending-contract. */
  async getBatchBoardDetail(
    scheduleId: string,
  ): Promise<BatchBoardScheduleDetail> {
    const s =
      await this.trainSchedulesRepository.findByIdWithFullGraph(scheduleId);
    if (!s)
      throw new NotFoundException(`Train schedule ${scheduleId} not found`);
    // Arrived / cancelled schedules stay viewable — the board is also the
    // historical record of what each train carried.
    // Batch board is IMPORT-only (export is FCFS, no batch/priority calc).
    if (s.direction !== "IMPORT") {
      throw new BadRequestException(
        "The batch board only covers import schedules",
      );
    }

    const wagonDims = await this.loadWagonDims();
    // The full graph already carries the schedule↔booking links — no separate
    // link query needed.
    const linkedIds = new Set(
      (s.scheduleBookings ?? []).map((l) => l.bookingId),
    );
    const bookings = await this.bookingsRepository.findAllBySchedule(s.id);

    let allocationPreview: Awaited<
      ReturnType<TrainSchedulingService["previewAllocationForSchedule"]>
    >;
    try {
      // Reuse the graph loaded above — the preview otherwise re-loads the same
      // heavy schedule graph a second time per request.
      allocationPreview =
        await this.trainSchedulingService.previewAllocationForSchedule(
          s.id,
          s,
        );
    } catch {
      allocationPreview = {
        assignedBookingIds: [],
        deferred: [],
        issues: [],
        violations: [],
      };
    }
    const allocationByBooking = new Map(
      allocationPreview.issues.map((i) => [i.bookingId, i]),
    );

    // Resolve consolidation-partner references for the shared-wagon badge. Most
    // partners are on this same schedule; look up any that aren't in one query.
    const refById = new Map(
      bookings.map((b) => [b.id, b.reference ?? b.id.slice(0, 8)]),
    );
    const missingPartnerIds = [
      ...new Set(
        bookings
          .map((b) => b.consolidationPartnerId)
          .filter((id): id is string => Boolean(id) && !refById.has(id!)),
      ),
    ];
    if (missingPartnerIds.length) {
      const partners = await this.dataSource
        .getRepository(Booking)
        .find({ where: { id: In(missingPartnerIds) } });
      for (const p of partners) {
        refById.set(p.id, p.reference ?? p.id.slice(0, 8));
      }
    }

    const items: BatchBoardBookingDetail[] = bookings.map((b) => {
      const need = this.needFor(b, wagonDims);
      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)),
        priorityScore: Number(b.priorityScore ?? 0),
        freightType: b.freightType ?? null,
        fullyExecutedAt: b.fullyExecutedAt
          ? b.fullyExecutedAt.toISOString()
          : null,
        selectedForBatchAt: b.selectedForBatchAt
          ? b.selectedForBatchAt.toISOString()
          : null,
        allocationStatus: alloc?.status ?? "NOT_ATTEMPTED",
        allocationIssue: alloc?.issue ?? null,
        consolidationPartnerId: b.consolidationPartnerId ?? null,
        consolidationPartnerRef: b.consolidationPartnerId
          ? (refById.get(b.consolidationPartnerId) ?? null)
          : null,
      };
    });

    const loco = trainSetLocomotiveLimits(s.trainSet);

    // The board renders ONE booking window — the schedule's own frozen window
    // (windowOpensAt/windowClosesAt + phase deadlines returned below). Bookings
    // split into two buckets: contract executed (in the window) vs pending
    // contract. The old per-cycle window projection was dropped — the UI never
    // showed it, and reconstructing every cycle cost a config load + grouping
    // pass per request.
    const countFor = (bucket: BatchBoardBookingDetail[]): BatchBoardCounts => {
      const counts: BatchBoardCounts = {
        allocated: 0,
        selectedForBatch: 0,
        ready: 0,
        waiting: 0,
        expired: 0,
        pendingContract: 0,
      };
      for (const b of bucket) {
        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 windowBookings = items.filter((i) => i.fullyExecutedAt);
    const pendingBookings = items.filter((i) => !i.fullyExecutedAt);

    return {
      scheduleId: s.id,
      scheduleReference: s.reference ?? null,
      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,
      direction: s.direction ?? null,
      windowPhase: s.windowPhase ?? null,
      windowOpensAt: s.windowOpensAt ? s.windowOpensAt.toISOString() : null,
      windowClosesAt: s.windowClosesAt ? s.windowClosesAt.toISOString() : null,
      docReviewEndsAt: s.docReviewEndsAt ? s.docReviewEndsAt.toISOString() : null,
      paymentPhaseEndsAt: s.paymentPhaseEndsAt
        ? s.paymentPhaseEndsAt.toISOString()
        : null,
      bookingCycleNo: s.bookingCycleNo ?? 0,
      train: s.trainSet?.train
        ? {
          id: s.trainSet.train.id,
          code: s.trainSet.train.code,
          trainName: s.trainSet.train.trainName ?? null,
        }
        : null,
      // Identity from the primary (legacy) locomotive; limit figures from the
      // whole set's effective minimum — what the fill engine actually spends.
      locomotive: loco
        ? {
          code: s.trainSet?.locomotive?.code ?? '',
          name: s.trainSet?.locomotive?.name ?? null,
          maxPullWeightTons: Number(loco.maxPullWeightTons),
          maxTrainLengthMeters: Number(loco.maxTrainLengthMeters),
        }
        : null,
      capacity: this.computeBoardCapacity(items, loco, s.maxWagons ?? null),
      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: windowBookings,
      pendingContract: {
        counts: countFor(pendingBookings),
        bookings: pendingBookings,
      },
      allocationViolations: allocationPreview.violations,
    };
  }

  /** Run wagon-level allocation for all eligible linked bookings on a schedule. */
  async runWagonAllocation(scheduleId: string) {
    const result =
      await this.trainSchedulingService.tryAutoWagonAllocation(scheduleId);
    if (result.assignedBookingIds.length > 0) {
      this.notifyBoardChanged(scheduleId, "wagon_allocation_run");
    }
    return result;
  }

  /**
   * Board capacity figures. `usedWeightTons` is GROSS (each item's weight already
   * includes the tare of the wagons it occupies), so the ceiling it is measured
   * against must be the same one the fill loop spends from: the locomotive's own
   * limits widened by its overage tolerance (global rule caps do not apply, same
   * as {@link capacityLimits}). Reading the raw `loco.maxPullWeightTons` here
   * showed staff a ceiling the batch engine did not use.
   */
  private computeBoardCapacity(
    items: Array<{
      state: BatchBoardBookingState;
      wagons: number;
      weightTons: number;
      lengthMeters: number;
    }>,
    loco: LocomotiveLimits | null,
    maxWagons: number | null,
  ): BatchBoardSchedule["capacity"] {
    const allocated = items.filter((i) => i.state === "ALLOCATED");
    // Every booking still targeting this train holds gross weight — including
    // PAID ones waiting for wagon allocation (WAITING) and post-dispatch
    // catch-all states. Counting only ALLOCATED + SELECTED_FOR_BATCH zeroed the
    // board's weight the moment customers paid. Only EXPIRED released its hold.
    const committed = items.filter((i) => i.state !== "EXPIRED");
    const caps = loco
      ? trainHardCaps({
          maxPullWeightTons: Number(loco.maxPullWeightTons),
          maxTrainLengthMeters: Number(loco.maxTrainLengthMeters),
          overageToleranceTons: Number(loco.overageToleranceTons) || 0,
          overageToleranceMeters: Number(loco.overageToleranceMeters) || 0,
        })
      : null;
    const round2 = (value: number) => Math.round(value * 100) / 100;

    return {
      allocatedWagons: allocated.reduce((sum, i) => sum + i.wagons, 0),
      allocatedLengthMeters: round2(
        allocated.reduce((sum, i) => sum + i.lengthMeters, 0),
      ),
      maxLengthMeters: caps ? caps.maxLengthMeters : null,
      usedWeightTons: round2(committed.reduce((sum, i) => sum + i.weightTons, 0)),
      maxWeightTons: caps ? caps.maxWeightTons : null,
      maxWagons: maxWagons ?? null,
    };
  }

  private buildScheduleSummary(
    s: TrainSchedule,
    items: BatchBoardBooking[],
  ): BatchBoardSchedule {
    const loco = trainSetLocomotiveLimits(s.trainSet);

    return {
      scheduleId: s.id,
      scheduleReference: s.reference ?? null,
      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,
      createdAt: s.createdAt ? s.createdAt.toISOString() : null,
      status: s.status,
      bookingWindowStatus: s.bookingWindowStatus,
      direction: s.direction ?? null,
      windowPhase: s.windowPhase ?? null,
      windowOpensAt: s.windowOpensAt ? s.windowOpensAt.toISOString() : null,
      windowClosesAt: s.windowClosesAt ? s.windowClosesAt.toISOString() : null,
      docReviewEndsAt: s.docReviewEndsAt ? s.docReviewEndsAt.toISOString() : null,
      paymentPhaseEndsAt: s.paymentPhaseEndsAt
        ? s.paymentPhaseEndsAt.toISOString()
        : null,
      bookingCycleNo: s.bookingCycleNo ?? 0,
      train: s.trainSet?.train
        ? {
          id: s.trainSet.train.id,
          code: s.trainSet.train.code,
          trainName: s.trainSet.train.trainName ?? null,
        }
        : null,
      // Identity from the primary (legacy) locomotive; limit figures from the
      // whole set's effective minimum — what the fill engine actually spends.
      locomotive: loco
        ? {
          code: s.trainSet?.locomotive?.code ?? '',
          name: s.trainSet?.locomotive?.name ?? null,
          maxPullWeightTons: Number(loco.maxPullWeightTons),
          maxTrainLengthMeters: Number(loco.maxTrainLengthMeters),
        }
        : null,
      capacity: this.computeBoardCapacity(items, loco, s.maxWagons ?? null),
      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 ------------------------------------------------------------

  /**
   * Whether the batch engine may reserve/allocate onto this schedule right now.
   * Legacy (no window phase): the customer-facing OPEN gate doubles as the fill gate.
   * Import window cycle: the engine fills while the customer window is CLOSED —
   * during DOC_REVIEW (early staff trigger) and PAYMENT (batch run + top-ups).
   * Export: FCFS while the booking window is open.
   */
  isFillable(schedule: TrainSchedule): boolean {
    if (schedule.bookingWindowStatus === "FULL") return false;
    if (!schedule.windowPhase) return schedule.bookingWindowStatus === "OPEN";
    if (schedule.direction === "EXPORT") {
      return schedule.windowPhase === "OPEN" && schedule.bookingWindowStatus === "OPEN";
    }
    return schedule.windowPhase === "DOC_REVIEW" || schedule.windowPhase === "PAYMENT";
  }

  /**
   * Fill one schedule from its priority-ordered pool until full. Returns the
   * number of commercial units it RESERVED this pass (0 for government-only or
   * no-fit passes) so a top-up caller can extend the payment phase only when a
   * fresh pay window actually opened.
   */
  async fillSchedule(scheduleId: string): Promise<number> {
    const schedule =
      await this.trainSchedulesRepository.findByIdWithFullGraph(scheduleId);
    if (!schedule || !this.isFillable(schedule)) return 0;
    const locomotive = trainSetLocomotiveLimits(schedule.trainSet);
    if (!schedule.trainSetId || !locomotive) {
      this.logger.warn(
        `Schedule ${scheduleId} has no locomotive/train set — skipped.`,
      );
      return 0;
    }

    const wagonDims = await this.loadWagonDims();
    const limits = await this.capacityLimits(locomotive);
    await this.syncScheduleMaxWagons(schedule, locomotive);
    const budget = await this.remainingBudget(schedule, limits, wagonDims);
    const minPerWagon = this.minPerWagonNeed(wagonDims);
    if (budget.isExhausted(minPerWagon)) {
      await this.setWindow(scheduleId, "FULL");
      return 0;
    }

    const pool = await this.bookingsRepository.findBatchPool(scheduleId);
    // Same bulk re-score as fillRouteDayInternal — the legacy per-schedule fill
    // must rank bulk bookings by their wagon-derived priority too.
    await this.recomputeBulkPriorities(pool, wagonDims);
    this.resortPoolByPriority(pool);
    const units = this.groupConsolidatedPool(pool);
    let armed = false;
    let preempted = false;
    let reservedThisPass = 0;
    let commercialReserved = 0;

    // Batch fill trace: caps + pool at entry. Kept on debug level — invaluable when
    // reservations trickle instead of landing in one pass (a reserve() throwing
    // mid-loop, e.g. schema drift, or a mis-synced capacity cap).
    this.logger.debug(
      `[fillSchedule ${scheduleId}] limits=${JSON.stringify(limits)} ` +
        `maxWagons=${schedule.maxWagons} remaining=${JSON.stringify(budget.maxRemaining())} ` +
        `poolSize=${pool.length} units=${units.length}`,
    );

    for (const unit of units) {
      const { primary: booking, partner } = unit;
      const isPair = partner != null;
      const need = isPair
        ? this.combinedNeed(booking, partner, wagonDims)
        : this.needFor(booking, wagonDims);
      const isGov = booking.isGovernment || (partner?.isGovernment ?? false);
      // Consolidated partners always share one corridor, so the primary's leg
      // stands for the pair.
      const leg = budget.legForYards(booking.originYardId, booking.destinationYardId);

      // Per-unit fit trace: which axis (wagons/weight/length) admits or rejects.
      this.logger.debug(
        `[fillSchedule ${scheduleId}] unit ${booking.reference}: need=${JSON.stringify(need)} ` +
          `roomOnLeg=${JSON.stringify(budget.remainingFor(leg))} fits=${budget.fits(need, leg)}`,
      );

      if (!budget.fits(need, leg)) {
        if (isGov) {
          const freed = await this.preemptForGovernment(
            scheduleId,
            need,
            leg,
            budget,
            wagonDims,
          );
          preempted = true;
          if (!freed) continue; // still doesn't fit even after preempt
        } else {
          // Doesn't fit whole. A split-eligible import booking is offered the part
          // that fits in the remaining room (top-up path splits the boundary
          // booking, mirroring fillRouteDay); otherwise skip and try the next.
          const cand: { id: string; budget: CorridorBudget; armed: boolean } = {
            id: scheduleId,
            budget,
            armed,
          };
          if (await this.maybeOfferPartial(booking, isPair, [cand], need)) {
            armed = cand.armed;
            continue;
          }
          continue; // skip a unit that exceeds weight/length/wagons, try the next
        }
      }

      // Isolate each unit so a throw in reserve/allocate (e.g. billing hiccup)
      // can't abort the whole top-up pass and leave the rest to trickle in one
      // per tick. Log + skip the failing unit, keep going.
      try {
        if (isGov) {
          await this.allocate(scheduleId, booking, "gov");
          if (partner) await this.allocate(scheduleId, partner, "gov");
        } else {
          await this.reserve(booking, scheduleId);
          if (partner) await this.reserve(partner, scheduleId);
          armed = true;
          commercialReserved += 1;
        }
        budget.subtract(need, leg);
        reservedThisPass += 1;
      } catch (err) {
        this.logger.error(
          `[fillSchedule ${scheduleId}] reserve/allocate FAILED for ${booking.reference} ` +
            `— skipping this unit, continuing: ${(err as Error).message}`,
        );
        continue;
      }
      if (budget.maxRemaining().wagons <= 0) break; // every leg exhausted — nothing more can board
    }

    this.logger.log(
      `[fillSchedule ${scheduleId}] reserved ${reservedThisPass}/${units.length} unit(s) this pass`,
    );
    if (budget.isExhausted(minPerWagon)) await this.setWindow(scheduleId, "FULL");
    if (armed) this.armSettle(scheduleId);
    // One push per fill pass (never per booking) — only when rows changed.
    if (reservedThisPass > 0 || armed || preempted) {
      this.notifyBoardChanged(scheduleId, "batch_fill");
    }
    void this.triggerWagonAllocation(scheduleId);
    return commercialReserved;
  }

  /**
   * 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<string[]> {
    const { scheduleIds } = await this.fillRouteDayInternal(
      originYardId,
      destinationYardId,
      day,
    );
    return scheduleIds;
  }

  /**
   * Route-day top-up for a single schedule: re-run the DAY pool over the whole
   * corridor the schedule belongs to, and report how many commercial units got a
   * fresh pay window.
   *
   * `fillSchedule` cannot do this job. Its pool (`findBatchPool`) is keyed on
   * `booking.train_schedule_id = :scheduleId`, but under day-level pooling a
   * booking that has not been reserved yet has a NULL `train_schedule_id` — it is
   * only pinned by `reserve()`. So the schedule-scoped top-up returned zero rows
   * and the waiting list never boarded after an expiry freed capacity; bookings
   * trickled in one per window cycle instead.
   */
  private async topUpFill(scheduleId: string): Promise<number> {
    const schedule = await this.trainSchedulesRepository.findById(scheduleId);
    if (!schedule?.scheduledDepartureDate) return 0;
    const { commercialReserved } = await this.fillRouteDayInternal(
      schedule.originStationId,
      schedule.destinationStationId,
      eatDay(schedule.scheduledDepartureDate),
    );
    return commercialReserved;
  }

  private async fillRouteDayInternal(
    originYardId: string,
    destinationYardId: string,
    day: string,
  ): Promise<{ scheduleIds: string[]; commercialReserved: number }> {
    // The day's fillable schedules on this exact corridor, earliest first. Fillable
    // covers legacy OPEN trains and window-cycle trains in DOC_REVIEW/PAYMENT —
    // the batch must run while the customer window is closed.
    const corridor = await this.trainSchedulesRepository.findAll({
      where: [
        {
          originStationId: originYardId,
          destinationStationId: destinationYardId,
          status: TrainScheduleStatusEnum.Draft,
        },
        {
          originStationId: originYardId,
          destinationStationId: destinationYardId,
          status: TrainScheduleStatusEnum.Scheduled,
        },
      ],
    });
    const onDay = corridor
      .filter(
        (s) =>
          s.scheduledDepartureDate != null &&
          eatDay(s.scheduledDepartureDate) === day,
      )
      .sort(
        (a, b) =>
          a.scheduledDepartureDate.getTime() - b.scheduledDepartureDate.getTime(),
      );

    // A schedule flagged FULL is rejected by isFillable() before its budget is
    // ever consulted. Re-derive that flag from live capacity first, so a train
    // whose bookings all expired is not skipped forever with an empty consist.
    for (const s of onDay) {
      if (s.bookingWindowStatus === "FULL") {
        await this.refreshWindowStatus(s.id);
        const fresh = await this.trainSchedulesRepository.findById(s.id);
        if (fresh) s.bookingWindowStatus = fresh.bookingWindowStatus;
      }
    }

    const scheduleIds = onDay.filter((s) => this.isFillable(s)).map((s) => s.id);

    if (scheduleIds.length === 0) {
      return { scheduleIds: [], commercialReserved: 0 };
    }

    const wagonDims = await this.loadWagonDims();

    // Live per-schedule corridor budget + arm/changed flags, in departure order.
    const trains: Array<{
      id: string;
      budget: CorridorBudget;
      armed: boolean;
      changed: boolean;
    }> = [];
    for (const id of scheduleIds) {
      const schedule =
        await this.trainSchedulesRepository.findByIdWithFullGraph(id);
      const locomotive = trainSetLocomotiveLimits(schedule?.trainSet);
      if (!schedule || !schedule.trainSetId || !locomotive) {
        this.logger.warn(
          `Schedule ${id} has no locomotive/train set — skipped.`,
        );
        continue;
      }
      const limits = await this.capacityLimits(locomotive);
      await this.syncScheduleMaxWagons(schedule, locomotive);
      const budget = await this.remainingBudget(schedule, limits, wagonDims);
      trains.push({ id, budget, armed: false, changed: false });
    }
    if (trains.length === 0) return { scheduleIds, commercialReserved: 0 };

    // The day pool covers every booking whose leg lies somewhere on one of the
    // day's corridors — full-route AND sub-corridor (e.g. Dire→Djibouti on an
    // Addis→Djibouti train). Which train actually takes a booking is decided
    // by the per-train legOf check below.
    const corridorYards = [...new Set(trains.flatMap((t) => t.budget.stops))];
    const pool = await this.bookingsRepository.findBatchPoolByCorridorDay(
      corridorYards,
      day,
    );
    // BULK bookings only get their real (wagon-derived) priority score now, at
    // batch time — stamp it and re-rank before the fill consumes the pool.
    await this.recomputeBulkPriorities(pool, wagonDims);
    this.resortPoolByPriority(pool);
    // Consolidated partners collapse into one atomic unit (both-or-neither); a
    // consolidated booking whose partner isn't ready this cycle is skipped.
    const units = this.groupConsolidatedPool(pool);

    // Batch fill trace: each train's caps + the day pool size at entry.
    this.logger.debug(
      `[fillRouteDay ${originYardId}->${destinationYardId} ${day}] ` +
        `trains=${trains.map((t) => `${t.id}:${JSON.stringify(t.budget.maxRemaining())}`).join(",")} ` +
        `poolSize=${pool.length} units=${units.length}`,
    );
    let reservedThisPass = 0;
    let commercialReserved = 0;

    for (const unit of units) {
      const { primary: booking, partner } = unit;
      const isPair = partner != null;
      const need = isPair
        ? this.combinedNeed(booking, partner, wagonDims)
        : this.needFor(booking, wagonDims);
      const isGov = booking.isGovernment || (partner?.isGovernment ?? false);

      const legOn = (t: { budget: CorridorBudget }): CorridorLeg | null =>
        t.budget.legOf(booking.originYardId, booking.destinationYardId);

      // First train (earliest departure) whose corridor carries this booking's
      // leg and still fits it as-is.
      let target = trains.find((t) => {
        const leg = legOn(t);
        return leg != null && t.budget.fits(need, leg);
      });

      // Per-unit trace: chosen train + each train's remaining room on this leg.
      this.logger.debug(
        `[fillRouteDay] unit ${booking.reference}: need=${JSON.stringify(need)} ` +
          `targetTrain=${target?.id ?? "none"} ` +
          `rooms=${trains
            .map((t) => {
              const leg = legOn(t);
              return leg ? `${t.id}:${JSON.stringify(t.budget.remainingFor(leg))}` : `${t.id}:offleg`;
            })
            .join(",")}`,
      );

      if (!target && isGov) {
        // Government fits nowhere on its own — try to preempt commercial
        // on each corridor-matching train (earliest first) until one frees room.
        for (const t of trains) {
          const leg = legOn(t);
          if (!leg) continue;
          const freed = await this.preemptForGovernment(
            t.id,
            need,
            leg,
            t.budget,
            wagonDims,
          );
          // Preempt may have displaced (expired) victims even when the need
          // still doesn't fit — the board must refresh either way.
          t.changed = true;
          if (freed) {
            target = t;
            break;
          }
        }
      }

      if (!target) {
        // Fits no train whole. A split-eligible booking is offered the largest
        // part that fits on the train with the most free wagons on its leg (this
        // covers both "fits nowhere" and the boundary case where earlier bookings
        // already consumed most of the room). Consolidated pairs / government /
        // non-import never split — isSplitEligible guards that. Passing the live
        // `trains` entries lets maybeOfferPartial mutate the chosen budget/armed.
        const offered = await this.maybeOfferPartial(booking, isPair, trains, need);
        if (offered) {
          // A partial offer opens a real commercial pay window, same as reserve().
          commercialReserved += 1;
          reservedThisPass += 1;
          continue;
        }
        // Stays in the pool, retried next batch/window cycle.
        this.notifier.unplaced(booking, day);
        if (partner) this.notifier.unplaced(partner, day);
        continue;
      }

      // A throw here (e.g. a billing/invoice hiccup inside reserve) must NOT abort
      // the whole pass — otherwise only the bookings before the failure get a pay
      // window and the rest trickle in one-per-tick on later retries (the
      // "selected one at a time / staggered" symptom). Isolate each unit: log +
      // skip a failing one, keep reserving the others. The skipped unit stays in
      // the pool and is retried next cycle.
      try {
        if (isGov) {
          await this.allocate(target.id, booking, "gov");
          if (partner) await this.allocate(target.id, partner, "gov");
        } else {
          await this.reserve(booking, target.id);
          if (partner) await this.reserve(partner, target.id);
          target.armed = true;
          commercialReserved += 1;
        }
        target.budget.subtract(need, legOn(target)!);
        target.changed = true;
        reservedThisPass += 1;
      } catch (err) {
        this.logger.error(
          `[fillRouteDay] reserve/allocate FAILED for ${booking.reference} on ${target.id} ` +
            `— skipping this unit, continuing the batch: ${(err as Error).message}`,
        );
      }
    }

    this.logger.log(
      `[fillRouteDay ${originYardId}->${destinationYardId} ${day}] reserved ${reservedThisPass}/${units.length} unit(s) this pass`,
    );

    const minPerWagon = this.minPerWagonNeed(wagonDims);
    for (const t of trains) {
      if (t.budget.isExhausted(minPerWagon)) await this.setWindow(t.id, "FULL");
      if (t.armed) this.armSettle(t.id);
      // One push per touched train per pass (never per booking). `armed` covers
      // commercial reserves + partial offers; `changed` covers gov allocations
      // and preemption.
      if (t.armed || t.changed) this.notifyBoardChanged(t.id, "batch_fill");
      void this.triggerWagonAllocation(t.id);
    }

    return { scheduleIds: trains.map((t) => t.id), commercialReserved };
  }

  /**
   * A lone commercial booking on a GENERAL or ONE_TIME contract may be offered a
   * partial (split-on-payment). Consolidated pairs never split (both-or-neither
   * shared wagon) and government bookings never split (they preempt).
   *
   * IMPORT is always eligible. EXPORT is eligible only when export split is
   * enabled: export historically rides one train whole, so splitting it changes
   * the FCFS money path — each split part still rides ONE train whole, and the
   * leftover becomes its own booking on the next train.
   */
  private isSplitEligible(booking: Booking, isPair: boolean): boolean {
    const directionOk =
      booking.tradeDirection === "IMPORT" ||
      (booking.tradeDirection === "EXPORT" && this.exportSplitEnabled);
    return (
      !isPair &&
      !booking.isGovernment &&
      directionOk &&
      (booking.contractKind === "GENERAL" || booking.contractKind === "ONE_TIME") &&
      this.splitService != null
    );
  }

  /**
   * Offer the largest fitting part of a booking that does not fit any candidate
   * train whole, on the train with the most free wagons on the booking's leg.
   * Mutates the chosen candidate's budget + armed flag in place. Returns true when
   * an offer was opened (caller should `continue` past this unit), false otherwise.
   * Shared by fillRouteDay (multi-train) and fillSchedule (single train). The leg
   * is computed per candidate from the booking's yards, so callers pass their live
   * train entries and only leg-carrying trains are considered.
   */
  private async maybeOfferPartial(
    booking: Booking,
    isPair: boolean,
    candidates: Array<{ id: string; budget: CorridorBudget; armed: boolean }>,
    need: Capacity,
  ): Promise<boolean> {
    if (!this.isSplitEligible(booking, isPair)) return false;
    const target = candidates
      .map((c) => {
        const leg = c.budget.legOf(booking.originYardId, booking.destinationYardId);
        return leg ? { c, leg, room: c.budget.remainingFor(leg) } : null;
      })
      .filter((x): x is NonNullable<typeof x> => x != null && x.room.wagons >= 1)
      .sort((a, b) => b.room.wagons - a.room.wagons)[0];
    if (!target) return false;
    const offered = await this.tryPartialOffer(
      booking,
      target.c.id,
      target.room,
      need,
    );
    if (!offered) return false;
    target.c.budget.subtract(offered, target.leg);
    target.c.armed = true;
    return true;
  }

  /**
   * Offer the largest fitting part of an over-capacity booking as a partial
   * (split-on-payment). Returns the capacity the offer consumes, or null when no
   * meaningful partial fits / an offer is already open.
   */
  private async tryPartialOffer(
    booking: Booking,
    scheduleId: string,
    budget: Capacity,
    need: Capacity,
  ): Promise<Capacity | null> {
    if (!this.splitService) return null;
    // A consolidated booking is already half of a shared wagon — never split it.
    if (booking.consolidationPartnerId) return null;
    if (await this.splitService.findOpenOffer(booking.id)) return null;

    const wagonDims = await this.loadWagonDims();

    // The wagon-slot axis alone under-constrains the offer. On a weight- or
    // length-limited train (slots to spare, but e.g. only 798T of pull weight
    // left) sizing by slots either produced an offer the fits() check below
    // rejected, or — when the free slots exceeded the booking's own wagon
    // count — sizeOffer refused outright, so a bulk booking on a weight-bound
    // train was never offered a split at all. Size across all three axes,
    // measured on the booking's REAL wagon type — the same one allocation
    // validates against. Bulk splits ride FULL wagons only: the offer never
    // part-loads its last wagon.
    const perWagon = this.dimsFor(booking, wagonDims);
    const partial = sizePartialOfferWagons(budget, need.wagons, perWagon, {
      fullWagonsOnly: booking.freightType === "BULK",
    });
    if (!partial) return null;

    const sized = await this.splitService.sizeOffer(
      booking,
      partial.wagons,
      need.wagons,
      perWagon.capacityTons,
      partial.maxCargoTons,
    );
    if (!sized) return null;

    const offeredNeed: Capacity = {
      wagons: sized.offeredWagons,
      weightTons: bookingGrossWeightTons(
        sized.offeredWeightTons,
        sized.offeredWagons,
        perWagon.tareWeightTons,
      ),
      lengthMeters: sized.offeredWagons * perWagon.lengthMeters,
    };
    if (!this.fits(offeredNeed, budget)) return null;

    const deadline = new Date(Date.now() + (await this.paymentWindowMs()));
    await this.splitService.createOffer(booking, scheduleId, sized, deadline);
    // Reserve like a normal batch selection, but the partial invoice + partial
    // pay-now notification were already produced by createOffer.
    await this.bookingsRepository.update(booking.id, {
      trainScheduleId: scheduleId,
      status: "SELECTED_FOR_BATCH",
      selectedForBatchAt: new Date(),
      paymentDeadline: deadline,
    } as never);
    booking.trainScheduleId = scheduleId;
    return offeredNeed;
  }

  /**
   * Settle a schedule's reserved bookings. `expireUnpaidUnknownDeadline` decides
   * how to treat a reservation with no deadline (durable path: leave it; timeout
   * path: expire it). Consolidated pairs settle atomically: both allocate only
   * when both paid; if either partner expires, both expire (a half-paid shared
   * wagon must not ship). Returns whether anything changed.
   */
  private async settleReserved(
    scheduleId: string,
    expireUnpaidUnknownDeadline: boolean,
  ): Promise<boolean> {
    const reserved =
      await this.bookingsRepository.findReservedForSchedule(scheduleId);
    const now = Date.now();
    const byId = new Map(reserved.map((b) => [b.id, b]));
    const done = new Set<string>();
    let anySettled = false;
    this.logger.debug(
      `[settleReserved ${scheduleId}] ${reserved.length} reserved booking(s) to settle`,
    );

    const isPaid = (b: Booking) =>
      b.paymentStatus === "PAID" || b.status === "PAID";
    const isExpired = (b: Booking) =>
      b.paymentDeadline
        ? b.paymentDeadline.getTime() <= now
        : expireUnpaidUnknownDeadline;

    for (const booking of reserved) {
      if (done.has(booking.id)) continue;
      const partner = booking.consolidationPartnerId
        ? (byId.get(booking.consolidationPartnerId) ?? null)
        : null;

      if (partner) {
        done.add(booking.id);
        done.add(partner.id);
        // Both-or-neither: allocate the shared wagon only when both partners paid;
        // if either lapsed, expire both so no half-paid wagon rides.
        if (isPaid(booking) && isPaid(partner)) {
          await this.allocate(scheduleId, booking, "paid");
          await this.allocate(scheduleId, partner, "paid");
          anySettled = true;
        } else if (isExpired(booking) || isExpired(partner)) {
          await this.expire(booking);
          await this.expire(partner);
          anySettled = true;
        }
        continue;
      }

      done.add(booking.id);
      if (isPaid(booking)) {
        if (!(await this.holdIfWagonShort(scheduleId, booking))) {
          await this.allocate(scheduleId, booking, "paid");
        }
        anySettled = true;
      } else if (isExpired(booking)) {
        await this.expire(booking);
        anySettled = true;
      }
    }
    return anySettled;
  }

  /**
   * Durable settle: allocate paid / expire overdue reservations, then top up the
   * freed capacity from the waiting list.
   *
   * Serialised per schedule. Two callers race here every time a payment phase
   * ends: `advanceImport`'s PAYMENT branch and the tick's `settleOverdueReservations`
   * backstop. Both read the same reserved rows in the same second, so without the
   * lock the second caller re-settles rows the first is mid-way through expiring,
   * and `concludeCycle` observes capacity that is neither pre- nor post-expiry.
   */
  async settleDueReservations(scheduleId: string): Promise<void> {
    await this.withScheduleLock(scheduleId, () =>
      this.settleAndTopUp(scheduleId, false),
    );
  }

  /**
   * Conclude-time retry: promote whatever still fits from the route-day waiting
   * list, opening fresh pay windows. Returns how many commercial units got
   * reserved — corridor-wide, since the fill is day-level and may reserve onto a
   * sibling train; the caller must check `hasLiveReservations` for its OWN
   * schedule before deciding to stay in PAYMENT.
   */
  async fillFromWaitingList(scheduleId: string): Promise<number> {
    return this.withScheduleLock(scheduleId, async () => {
      let promoted = 0;
      for (let round = 0; round < 10; round += 1) {
        const reservedThisRound = await this.topUpFill(scheduleId);
        if (reservedThisRound <= 0) break;
        promoted += reservedThisRound;
        await this.extendPaymentPhaseForTopUp(scheduleId);
      }
      if (promoted > 0) {
        this.notifyBoardChanged(scheduleId, "conclude_waiting_list_fill");
      }
      return promoted;
    });
  }

  /**
   * Settle, then keep promoting the waiting list until the train can take no more.
   * Returns whether anything settled.
   *
   * One top-up pass is not enough: expiring an N-wagon booking can free room for
   * several smaller ones, and reserving those can in turn leave room for the next
   * size down. Loop until a pass reserves nothing, so the batch ends with the train
   * as full as the pool allows — rather than leaving a booking stranded until the
   * next window cycle.
   *
   * Each round that opens a fresh pay window pushes `paymentPhaseEndsAt` out, so
   * `concludeCycle` cannot fire before the promoted customers' deadlines.
   */
  private async settleAndTopUp(
    scheduleId: string,
    expireUnpaidUnknownDeadline: boolean,
  ): Promise<boolean> {
    const anySettled = await this.settleReserved(
      scheduleId,
      expireUnpaidUnknownDeadline,
    );
    if (!anySettled) return false;

    this.logger.log(
      `[BATCH] settle changed state on ${scheduleId} — running top-up fill for the waiting list`,
    );

    // Bounded: every round either reserves at least one unit (shrinking the pool)
    // or breaks. The cap is a backstop against a pathological reserve/expire cycle.
    let promoted = 0;
    for (let round = 0; round < 10; round += 1) {
      const reservedThisRound = await this.topUpFill(scheduleId);
      if (reservedThisRound <= 0) break;
      promoted += reservedThisRound;
      await this.extendPaymentPhaseForTopUp(scheduleId);
    }

    if (promoted > 0) {
      this.logger.log(
        `[BATCH] top-up promoted ${promoted} waiting booking(s) onto ${scheduleId} ` +
          `— payment phase extended for them`,
      );
    }
    // The settle may have resolved the last pay window on a full export day
    // (paid → allocated, and the top-up found nothing else that fits) — sweep
    // the date's leftover bookings. Self-guarded: no-op for import/domestic
    // and while any train on the day can still take bookings.
    await this.expireLeftoverExportDay(scheduleId);
    // Emitted here (not in settleDueReservations/settleBatch, which both wrap
    // this) so one settle produces one push, after every allocation/expiry/
    // top-up extension for this schedule has been persisted.
    this.notifyBoardChanged(scheduleId, "reservations_settled");
    return true;
  }

  /**
   * Run `fn` with exclusive access to `scheduleId`. Concurrent callers await the
   * in-flight run rather than interleaving with it. Single-process only — a second
   * API replica would need a row lock on the schedule instead.
   */
  private async withScheduleLock<T>(
    scheduleId: string,
    fn: () => Promise<T>,
  ): Promise<T> {
    const inFlight = this.scheduleLocks.get(scheduleId) ?? Promise.resolve();
    // Chain onto the previous holder; swallow its rejection so one failure does
    // not poison every later caller's lock.
    const run = inFlight.catch(() => undefined).then(fn);
    const gate = run.then(
      () => undefined,
      () => undefined,
    );
    this.scheduleLocks.set(scheduleId, gate);
    try {
      return await run;
    } finally {
      // Last one out clears the slot so the map does not grow without bound.
      if (this.scheduleLocks.get(scheduleId) === gate) {
        this.scheduleLocks.delete(scheduleId);
      }
    }
  }

  // ---- settle (1h after a batch) -------------------------------------------

  /** Allocate paid reservations, expire the rest, then top up. */
  async settleBatch(scheduleId: string): Promise<void> {
    this.removeTimeout(scheduleId);
    await this.withScheduleLock(scheduleId, () =>
      this.settleAndTopUp(scheduleId, true),
    );
    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}`,
        ),
      );
  }

  /**
   * Announce that a schedule's batch-board data changed so open boards refetch.
   * Called AFTER the state change is persisted; a push failure only logs — it
   * must never break the business transaction that triggered it.
   */
  private notifyBoardChanged(scheduleId: string, reason: string): void {
    try {
      this.bookingWindowGateway.emitBatchChanged(scheduleId, reason);
    } catch (err) {
      this.logger.warn(
        `Batch-board push (${reason}) 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<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" });
    if (!(await this.holdIfWagonShort(booking.trainScheduleId, booking))) {
      await this.allocate(booking.trainScheduleId, booking, "paid");
    }

    const schedule = await this.trainSchedulesRepository.findByIdWithFullGraph(
      booking.trainScheduleId,
    );
    if (schedule && (await this.isTrainFull(schedule))) {
      await this.setWindow(booking.trainScheduleId, "FULL");
      // Same as the webhook path: a staff mark-paid can settle the last live
      // pay window on a now-full export day — sweep the date's leftovers.
      void this.expireLeftoverExportDay(booking.trainScheduleId);
    }
    void this.triggerWagonAllocation(booking.trainScheduleId!);
    this.notifyBoardChanged(booking.trainScheduleId, "booking_marked_paid");
  }

  /**
   * 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",
      );
    }
    const stops = await this.stopsForSchedule(schedule);
    const fromIdx = stops.indexOf(booking.originYardId);
    const toIdx = stops.indexOf(booking.destinationYardId);
    if (fromIdx < 0 || toIdx < 0 || fromIdx >= toIdx) {
      throw new BadRequestException(
        "Target schedule is not on the booking route",
      );
    }

    const sourceScheduleId = booking.trainScheduleId ?? null;
    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,
        // A paid booking still hunting for a wagon keeps its flag through the
        // move — it only clears when wagons are actually assigned.
        schedulingStatus:
          booking.schedulingStatus === "WAITING_FOR_WAGON"
            ? "WAITING_FOR_WAGON"
            : "ELIGIBLE",
        paymentDeadline: null,
        selectedForBatchAt: null,
      } as never);
    });
    // Both boards changed: the booking left the source train and joined the target.
    if (sourceScheduleId && sourceScheduleId !== newScheduleId) {
      this.notifyBoardChanged(sourceScheduleId, "booking_moved");
    }
    this.notifyBoardChanged(newScheduleId, "booking_moved");
  }

  /** 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`);
    // Capture the train before expire() detaches the booking from it — the
    // top-up has to run against the schedule whose wagons were just freed.
    const freedScheduleId = booking.trainScheduleId;
    await this.expire(booking);
    if (freedScheduleId) {
      const topUpReserved = await this.topUpFill(freedScheduleId);
      if (topUpReserved > 0) {
        await this.extendPaymentPhaseForTopUp(freedScheduleId);
      }
      // After the top-up + phase extension so one push carries the final state.
      this.notifyBoardChanged(freedScheduleId, "reservation_expired");
    }
  }

  // ---- intercity ride-along API ---------------------------------------------

  /**
   * Remaining corridor capacity budget (per-edge wagons / weight / length) for
   * a schedule, and the per-booking need calculator — exposed for the intercity
   * accept flow, which reserves ride-along bookings onto import/export trains
   * outside the batch engine. Segment-based: an intercity booking fits whenever
   * ITS leg has room, even if the train is full on other legs.
   */
  async intercityCapacity(scheduleId: string): Promise<{
    budget: CorridorBudget;
    needFor: (booking: Booking) => Capacity;
  } | null> {
    const schedule =
      await this.trainSchedulesRepository.findByIdWithFullGraph(scheduleId);
    const locomotive = trainSetLocomotiveLimits(schedule?.trainSet);
    if (!schedule || !locomotive) return null;
    const wagonDims = await this.loadWagonDims();
    const limits = await this.capacityLimits(locomotive);
    // Built trains use the leg-aware corridor budget too: the wagon planner
    // consumes stock PER EDGE (planWagonsWithStock legs), so a consist wagon
    // that runs empty Gelan→Adama genuinely can carry an intercity booking
    // there before its export cargo boards at Adama. A train full on one leg
    // still accepts ride-alongs on its empty legs — that is the whole point
    // of the ride-along flow.
    const budget = await this.remainingBudget(schedule, limits, wagonDims);
    return { budget, needFor: (booking) => this.needFor(booking, wagonDims) };
  }

  /**
   * Accept an intercity booking onto the given train. Commercial bookings get
   * the same pay-window lifecycle as a batch reservation (deadline, invoice
   * due-date sync, pay-now notify, settle on the window tick), so payment →
   * allocation needs no special path. Government bookings allocate directly.
   */
  async acceptIntercity(booking: Booking, scheduleId: string): Promise<void> {
    if (booking.isGovernment) {
      await this.dataSource
        .getRepository(Booking)
        .update(booking.id, { trainScheduleId: scheduleId });
      booking.trainScheduleId = scheduleId;
      await this.allocate(scheduleId, booking, 'gov');
      this.notifyBoardChanged(scheduleId, 'intercity_accepted');
      return;
    }
    await this.reserve(booking, scheduleId);
    this.armSettle(scheduleId);
    this.notifyBoardChanged(scheduleId, 'intercity_accepted');
  }

  // ---- 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<void> {
    // Idempotency guard: a booking already reserved (pay window open) or already
    // paid on THIS schedule must never be re-reserved — that would fire a second
    // `payNow` and reset its deadline, the "asked to pay again after paying"
    // symptom. Read fresh state (the in-memory `booking` may be stale from the
    // pooled query). Only bookings not yet committed to this train pass through.
    const fresh = await this.dataSource
      .getRepository(Booking)
      .findOne({ where: { id: booking.id } });
    if (
      fresh &&
      fresh.trainScheduleId === scheduleId &&
      (fresh.status === "SELECTED_FOR_BATCH" ||
        fresh.status === "AWAITING_PAYMENT" ||
        fresh.status === "PAID" ||
        fresh.paymentStatus === "PAID")
    ) {
      this.logger.debug(
        `[BATCH] reserve skipped for ${booking.reference} — already ` +
          `${fresh.status}/${fresh.paymentStatus} on schedule ${scheduleId}`,
      );
      return;
    }
    const now = new Date();
    let deadline = new Date(now.getTime() + (await this.paymentWindowMs()));
    // EXPORT parity: pay windows on an export train never outlive its booking
    // window — export bookings expire at close, so anything reserved onto the
    // same train (FCFS export or an intercity ride-along) must too. Import
    // keeps the plain payment window; its cycles re-fill after settle.
    const targetSchedule = await this.dataSource
      .getRepository(TrainSchedule)
      .findOne({ where: { id: scheduleId } });
    if (targetSchedule?.direction === "EXPORT") {
      const cutoff =
        targetSchedule.windowClosesAt ?? targetSchedule.scheduledDepartureDate;
      if (cutoff && cutoff.getTime() <= now.getTime()) {
        throw new BadRequestException(
          "Export booking window has closed — cannot open a pay window on this train",
        );
      }
      if (cutoff && cutoff.getTime() < deadline.getTime()) {
        deadline = new Date(cutoff);
      }
    }
    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 DRAFT at booking creation / operation-accept,
    // before this pay window existed. Reserving is the moment the booking becomes
    // payable (SELECTED_FOR_BATCH + a real deadline), so issue the draft here and
    // print the deadline as its due date — never earlier, or the customer could
    // settle an invoice for a slot they have not been offered yet. Idempotent: a
    // re-reserve only refreshes `dueAt`.
    await this.billing.issuePayable(
      Freight.InvoiceSource.Booking,
      booking.id,
      deadline,
      "PREPAID",
    );
    await this.notifier.payNow(booking, deadline);
    const reservedWagons = this.wagonsFor(booking, await this.loadWagonDims());
    this.logger.log(
      `[BATCH] RESERVED ${booking.reference} (${reservedWagons}w, ` +
        `priority ${booking.priorityScore ?? 0}) on schedule ${scheduleId} — ` +
        `pay by ${deadline.toISOString()}`,
    );
    // Customer tracking: a wagon slot is reserved and the freight pay window is
    // open. Doc-trigger path — silent no-op for bookings without milestone rows.
    void this.completeTrackingMilestones(booking.id, [
      "WAGON_REQUESTED",
      "FREIGHT_PAYMENT_PENDING",
    ]);
  }

  /** Allocate a booking to the schedule's train (creates the TrainScheduleBooking link). */
  /**
   * Fleet preflight shared by every single-booking paid-allocation path: when
   * no wagon of the booking's required type is free, hold it OUT of the train
   * instead of linking — it stays PAID + unlinked in the (route, day) pool,
   * flagged WAITING_FOR_WAGON, and staff place it on any same-day schedule from
   * the workspace "Paid · unassigned" panel once a wagon frees up. Returns true
   * when the booking was held. Consolidated pairs are exempt (the shared wagon
   * is both-or-neither and settles atomically in settleReserved).
   */
  private async holdIfWagonShort(
    scheduleId: string,
    booking: Booking,
  ): Promise<boolean> {
    if (booking.consolidationPartnerId) return false;
    const shortage =
      await this.trainSchedulingService.previewPaidBookingWagonShortage(
        scheduleId,
        booking.id,
      );
    if (!shortage) return false;

    await this.dataSource.getRepository(Booking).update(booking.id, {
      status: "PAID",
      paymentStatus: "PAID",
      schedulingStatus: "WAITING_FOR_WAGON",
      paymentDeadline: null,
      selectedForBatchAt: null,
    } as never);
    // Payment landed — record it even though nothing boards yet. The wagon
    // milestone stays pending until staff assign one.
    void this.completeTrackingMilestones(booking.id, [
      "FREIGHT_PAYMENT_PENDING",
      "FREIGHT_PAYMENT_SETTLED",
    ]);
    this.logger.warn(
      `PAID booking ${booking.reference ?? booking.id} is WAITING FOR WAGON: ` +
        `needs ${shortage.wagonsNeeded} × ${shortage.wagonTypeCodes}, ` +
        `${shortage.wagonsAvailable} available (short ${shortage.wagonsShort}). ` +
        `Held in the day pool for manual placement.`,
    );
    this.notifyBoardChanged(scheduleId, "booking_waiting_wagon");
    return true;
  }

  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,
        selectedForBatchAt: null,
      } as never);
    });
    this.logger.log(
      `[BATCH] ALLOCATED ${booking.reference} (${reason}) to train on schedule ${scheduleId}`,
    );
    this.notifier.secured(booking, reason, scheduleId);
    void this.triggerWagonAllocation(scheduleId);
    void this.markWagonAllocatedMilestone(booking.id);
    // Customer tracking: freight payment settled (commercial pay-window path).
    // Government allocations don't pay upfront — theirs stay pending.
    if (reason === 'paid') {
      void this.completeTrackingMilestones(booking.id, [
        'WAGON_REQUESTED',
        'FREIGHT_PAYMENT_PENDING',
        'FREIGHT_PAYMENT_SETTLED',
      ]);
    }
  }

  private async markWagonAllocatedMilestone(bookingId: string): Promise<void> {
    if (!this.milestoneService) return;
    try {
      await this.milestoneService.completeForBooking(bookingId, 'WAGON_ALLOCATED');
    } catch {
      // Booking may have no milestone rows (non-contract path).
    }
  }

  /**
   * Complete customer-tracking milestones on lifecycle events via the
   * doc-trigger path — a silent no-op for bookings without milestone rows
   * (non-customs bookings). Never blocks the batch action.
   */
  private async completeTrackingMilestones(
    bookingId: string,
    codes: string[],
  ): Promise<void> {
    if (!this.milestoneService) return;
    for (const code of codes) {
      try {
        await this.milestoneService.completeByDocTrigger({ bookingId }, code);
      } catch (err) {
        this.logger.warn(
          `Milestone ${code} completion failed for booking ${bookingId}: ${(err as Error).message}`,
        );
      }
    }
  }

  /**
   * 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.
   * `reason` picks the customer message: 'payment' (pay window lapsed) or
   * 'no-capacity' (no train on the chosen day could take the booking).
   *
   * PAID GUARD: a booking whose payment has landed is never expired — money was
   * taken, so it boards, even when the webhook arrived after the deadline or the
   * settle read a stale row. It allocates onto the train it was selected for; if
   * the wagon planner then finds no physical wagon, the booking stays linked and
   * staff assign wagons manually. Consolidated bookings are exempt from the
   * rescue: the shared wagon is both-or-neither, and settleReserved owns that
   * pair decision.
   */
  private async expire(
    booking: Booking,
    reason: "payment" | "no-capacity" = "payment",
  ): Promise<void> {
    if (!booking.consolidationPartnerId) {
      const fresh = await this.dataSource
        .getRepository(Booking)
        .findOne({ where: { id: booking.id }, relations: { company: true } });
      const paid =
        fresh != null &&
        (fresh.paymentStatus === "PAID" || fresh.status === "PAID");
      const paidScheduleId = fresh?.trainScheduleId ?? booking.trainScheduleId;
      if (paid && paidScheduleId) {
        this.logger.log(
          `[BATCH] expire skipped for ${booking.reference} — payment already ` +
            `landed; allocating on schedule ${paidScheduleId} instead`,
        );
        if (!(await this.holdIfWagonShort(paidScheduleId, fresh))) {
          await this.allocate(paidScheduleId, fresh, "paid");
        }
        return;
      }
    }
    const freedScheduleId = booking.trainScheduleId;
    await this.bookingsRepository.update(booking.id, {
      trainScheduleId: null,
      status: "EXPIRED",
      schedulingStatus: "ELIGIBLE",
      paymentDeadline: null,
      selectedForBatchAt: null,
    } as never);
    booking.trainScheduleId = null;
    // The wagons this reservation held are back — a schedule parked at FULL
    // because of it must reopen, or it can never be filled again.
    if (freedScheduleId) await this.refreshWindowStatus(freedScheduleId);
    // An unpaid partial offer dies with the reservation — the booking stays whole.
    if (this.splitService) {
      await this.splitService.expireOpenOffer(booking.id);
    }
    // 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");
    if (reason === "no-capacity") {
      this.notifier.expiredNoCapacity(booking);
    } else {
      this.notifier.expired(booking);
    }
    this.logger.log(
      `[BATCH] EXPIRED ${booking.reference} — ` +
        (reason === "no-capacity"
          ? "no train on its day had capacity left"
          : "payment window passed; freed its wagons back to the pool for top-up"),
    );
  }

  /**
   * End-of-day sweep: once a schedule's window cycle concludes and NO other
   * train on the same route-day can still run a cycle, the waiting pool for
   * that day is dead — a FULLY_EXECUTED booking left in it would wait forever.
   * Expire every leftover commercial booking and tell the customers to rebook
   * another day. Government bookings are never auto-expired (they preempt).
   * Returns how many bookings were expired.
   */
  async expireLeftoverDayPool(scheduleId: string): Promise<number> {
    const schedule = await this.trainSchedulesRepository.findById(scheduleId);
    if (!schedule?.scheduledDepartureDate) return 0;
    const day = eatDay(schedule.scheduledDepartureDate);
    const group: RouteDayGroup = {
      originYardId: schedule.originStationId,
      destinationYardId: schedule.destinationStationId,
      day,
    };

    // Another train on this route-day that can still take bookings keeps the
    // pool alive — when IT concludes, its own sweep runs this check again.
    const siblings = await this.trainSchedulesRepository.findAll({
      where: [
        {
          originStationId: group.originYardId,
          destinationStationId: group.destinationYardId,
          status: TrainScheduleStatusEnum.Draft,
        },
        {
          originStationId: group.originYardId,
          destinationStationId: group.destinationYardId,
          status: TrainScheduleStatusEnum.Scheduled,
        },
      ],
    });
    const anotherTrainStillOpen = siblings.some(
      (s) =>
        s.id !== schedule.id &&
        s.scheduledDepartureDate != null &&
        eatDay(s.scheduledDepartureDate) === day &&
        s.windowPhase !== "DONE" &&
        s.bookingWindowStatus !== "FULL",
    );
    if (anotherTrainStillOpen) return 0;

    const corridorYards = await this.corridorYardsForRouteDay(group);
    const pool = corridorYards.length
      ? await this.bookingsRepository.findBatchPoolByCorridorDay(corridorYards, day)
      : await this.bookingsRepository.findBatchPoolByRouteDay(
          group.originYardId,
          group.destinationYardId,
          day,
        );
    const leftovers = pool.filter((b) => !b.isGovernment);
    // Capture pinned schedules BEFORE expire() clears trainScheduleId, so each
    // touched board gets exactly one push at the end of the sweep.
    const touchedScheduleIds = new Set<string>();
    if (leftovers.length) touchedScheduleIds.add(scheduleId);
    for (const booking of leftovers) {
      if (booking.trainScheduleId) touchedScheduleIds.add(booking.trainScheduleId);
      await this.expire(booking, "no-capacity");
    }
    if (leftovers.length) {
      this.logger.log(
        `[BATCH] ${this.groupLabel(group)}: no train left with capacity — ` +
          `expired ${leftovers.length} waiting booking(s)`,
      );
    }
    for (const id of touchedScheduleIds) {
      this.notifyBoardChanged(id, "day_pool_expired");
    }
    return leftovers.length;
  }

  /**
   * EXPORT counterpart of the conclude-time sweep. Export has no batch cycle,
   * so nothing ever concluded its day: bookings still waiting when the trains
   * filled up or the window closed stayed pending forever. Once every export
   * train on this route-day is shut — window DONE, or FULL with no pay window
   * still live that could lapse and free space — the date is dead: expire the
   * un-accepted bookings staff can no longer accept AND the ready
   * (FULLY_EXECUTED) bookings that never got a reservation (consolidation
   * waiters). Runs at export window close and whenever an export train's
   * fullness settles.
   */
  async expireLeftoverExportDay(scheduleId: string): Promise<void> {
    const schedule = await this.trainSchedulesRepository.findById(scheduleId);
    if (schedule?.direction !== "EXPORT" || !schedule.scheduledDepartureDate) {
      return;
    }
    const day = eatDay(schedule.scheduledDepartureDate);
    const trains = (
      await this.trainSchedulesRepository.findAll({
        where: [
          {
            originStationId: schedule.originStationId,
            destinationStationId: schedule.destinationStationId,
            status: TrainScheduleStatusEnum.Draft,
          },
          {
            originStationId: schedule.originStationId,
            destinationStationId: schedule.destinationStationId,
            status: TrainScheduleStatusEnum.Scheduled,
          },
        ],
      })
    ).filter(
      (s) =>
        s.scheduledDepartureDate != null &&
        eatDay(s.scheduledDepartureDate) === day,
    );
    for (const s of trains) {
      // Any train still taking bookings keeps the date alive.
      if (s.windowPhase !== "DONE" && s.bookingWindowStatus !== "FULL") return;
      // A FULL train whose reservations are still inside their pay windows can
      // reopen when one lapses unpaid — defer; the settle re-runs this sweep.
      if (s.windowPhase !== "DONE" && (await this.hasLiveReservations(s.id))) {
        return;
      }
    }
    await this.expireUnacceptedForRouteDay({
      originYardId: schedule.originStationId,
      destinationYardId: schedule.destinationStationId,
      day,
    });
    await this.expireLeftoverDayPool(scheduleId);
  }

  /**
   * Union of stop yards across the day's fillable schedules on this corridor —
   * the same pool scope fillRouteDay uses, so full-route AND sub-corridor bookings
   * are covered. Empty when no fillable schedule exists for the group.
   */
  private async corridorYardsForRouteDay(
    group: RouteDayGroup,
  ): Promise<string[]> {
    const corridor = await this.trainSchedulesRepository.findAll({
      where: [
        {
          originStationId: group.originYardId,
          destinationStationId: group.destinationYardId,
          status: TrainScheduleStatusEnum.Draft,
        },
        {
          originStationId: group.originYardId,
          destinationStationId: group.destinationYardId,
          status: TrainScheduleStatusEnum.Scheduled,
        },
      ],
    });
    const yards = new Set<string>();
    for (const schedule of corridor) {
      if (
        schedule.scheduledDepartureDate == null ||
        eatDay(schedule.scheduledDepartureDate) !== group.day
      ) {
        continue;
      }
      for (const yardId of await this.stopsForSchedule(schedule)) {
        yards.add(yardId);
      }
    }
    return [...yards];
  }

  /**
   * Sweep bookings on a route-day whose operation request staff did NOT accept by
   * the time the window's document-review phase ends. They never reached
   * FULLY_EXECUTED, so they never enter the batch — expire them (customer must
   * rebook a new window). No reservation and no invoice exists yet at this stage,
   * so this is a lighter expiry than `expire()`: just flip status + notify, and
   * best-effort close any payable if one was issued early. Government/export are
   * excluded by the query.
   */
  async expireUnacceptedForRouteDay(group: RouteDayGroup): Promise<void> {
    const corridorYards = await this.corridorYardsForRouteDay(group);
    if (corridorYards.length === 0) return;
    const unaccepted = await this.bookingsRepository.findUnacceptedForRouteDay(
      corridorYards,
      group.day,
    );
    if (unaccepted.length > 0) {
      this.logger.log(
        `[BATCH] doc-review end: expiring ${unaccepted.length} un-accepted booking(s) ` +
          `on ${group.originYardId}->${group.destinationYardId} ${group.day}`,
      );
    }
    // Only bookings pinned to a train show on a board — collect their schedules
    // and push once per schedule after the sweep (most unaccepted rows are
    // unpinned under day-level pooling, so this usually emits nothing).
    const touchedScheduleIds = new Set<string>();
    for (const booking of unaccepted) {
      if (booking.trainScheduleId) touchedScheduleIds.add(booking.trainScheduleId);
      await this.bookingsRepository.update(booking.id, {
        status: "EXPIRED",
        schedulingStatus: "ELIGIBLE",
        // Free the shipment day so the customer can rebook a fresh window.
        scheduledDate: null,
      } as never);
      // Close any payable issued before doc-review end (normally none — the invoice
      // is created at ops-accept, which by definition has not happened here).
      await this.billing
        .expirePayable(Freight.InvoiceSource.Booking, booking.id, "PREPAID")
        .catch(() => undefined);
      this.notifier.expired(booking);
      this.logger.log(
        `[BATCH] EXPIRED (unaccepted) ${booking.reference}:${booking.id} at doc-review end`,
      );
    }
    for (const id of touchedScheduleIds) {
      this.notifyBoardChanged(id, "unaccepted_expired");
    }
  }

  /**
   * Free capacity for a government booking by displacing the lowest-priority commercial
   * bookings (reserved first, then allocated — including PAID). Displaced → EXPIRED + notified.
   * Only victims whose legs overlap the government booking's leg actually free useful
   * room, so others are skipped. Mutates `budget`; returns whether the need now fits.
   */
  private async preemptForGovernment(
    scheduleId: string,
    need: Capacity,
    leg: CorridorLeg,
    budget: CorridorBudget,
    wagonDims: WagonDims,
  ): Promise<boolean> {
    if (budget.fits(need, leg)) return true;
    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 (budget.fits(need, leg)) break;
      const victimLeg = budget.legForYards(
        victim.originYardId,
        victim.destinationYardId,
      );
      // Displacing a booking on a disjoint leg frees nothing the government
      // booking can use — don't kill it for nothing.
      const overlaps = victimLeg.fromEdge < leg.toEdge && leg.fromEdge < victimLeg.toEdge;
      if (!overlaps) continue;
      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);
      budget.add(this.needFor(victim, wagonDims), victimLeg);
      // Displacing frees wagons the same way an expiry does — don't leave the
      // schedule stuck at FULL.
      await this.refreshWindowStatus(scheduleId);
    }
    return budget.fits(need, leg);
  }

  // ---- capacity helpers -----------------------------------------------------

  /**
   * Collapse consolidated partners into single pool entries so the fill treats a
   * shared-wagon pair as one atomic unit (both-or-neither). For each pool entry:
   *  - no `consolidationPartnerId` → passes through as a lone booking.
   *  - consolidated + partner also in this pool → emitted ONCE (at the position of
   *    whichever partner ranks first) as a pair; the partner is not emitted again.
   *  - consolidated + partner NOT in this pool → dropped (can't ship half a wagon;
   *    it waits for the partner to become ready in a later cycle).
   * The pool is already priority-ordered, so emitting the pair at the first-seen
   * partner's slot ranks it by the stronger (max-priority) partner automatically.
   */
  private groupConsolidatedPool(
    pool: Booking[],
  ): Array<{ primary: Booking; partner: Booking | null }> {
    const byId = new Map(pool.map((b) => [b.id, b]));
    const emitted = new Set<string>();
    const units: Array<{ primary: Booking; partner: Booking | null }> = [];
    for (const booking of pool) {
      if (emitted.has(booking.id)) continue;
      const partnerId = booking.consolidationPartnerId ?? null;
      if (!partnerId) {
        emitted.add(booking.id);
        units.push({ primary: booking, partner: null });
        continue;
      }
      const partner = byId.get(partnerId) ?? null;
      if (!partner) {
        // Both-or-neither: partner not ready in this pool → skip the pair entirely.
        emitted.add(booking.id);
        continue;
      }
      emitted.add(booking.id);
      emitted.add(partner.id);
      units.push({ primary: booking, partner });
    }
    return units;
  }

  /**
   * Combined capacity need of a consolidated pair sharing wagons. The whole point of
   * consolidation is that the two partial 20ft counts pack onto the SAME wagons, so
   * the shared wagon count is ceil((c1+c2)/2) — strictly fewer than summing the two
   * independently-rounded-up needs (that is the capacity consolidation saves).
   */
  private combinedNeed(
    primary: Booking,
    partner: Booking,
    wagonDims: WagonDims,
  ): Capacity {
    const containers = (b: Booking): number =>
      (b.bookingContainers ?? []).reduce((sum, c) => sum + Number(c.quantity ?? 0), 0);
    const totalContainers = containers(primary) + containers(partner);
    const cargoTons = bookingCargoTons(primary) + bookingCargoTons(partner);

    // Consolidation shares TEU slots, never rated payload: the pair still needs
    // enough wagons to carry its combined cargo, so the weight axis bounds the
    // shared count exactly as it bounds an individual booking's. A pair shares
    // wagons, so the primary's wagon type stands for both partners.
    const dims = this.dimsFor(primary, wagonDims);
    const capacityTons = dims.capacityTons;
    const byWeight =
      cargoTons > 0 && capacityTons > 0 ? Math.ceil(cargoTons / capacityTons) : 0;
    const byLength =
      totalContainers > 0
        ? Math.ceil(totalContainers / MAX_TEU_SLOTS_PER_WAGON)
        : this.wagonsFor(primary, wagonDims) + this.wagonsFor(partner, wagonDims);
    const sharedWagons = Math.max(byLength, byWeight);

    return {
      wagons: sharedWagons,
      // Consolidation saves tare as well as slots: the pair rides `sharedWagons`
      // wagons, so it is charged `sharedWagons` tares, not one per booking.
      weightTons: bookingGrossWeightTons(
        cargoTons,
        sharedWagons,
        dims.tareWeightTons,
      ),
      lengthMeters: sharedWagons * dims.lengthMeters,
    };
  }

  /**
   * Stamp real priority scores on the pool's BULK bookings before the batch
   * ranks it. Submit-time scoring runs with totalWagons = 0 for bulk (a bulk
   * booking has no container lines to carry a wagon count), so every
   * wagon-range priority config missed and bulk import bookings entered the
   * batch at score 0 — they were never prioritized. Their wagon footprint is
   * derivable from tonnage vs. live wagon capacity (wagonsFor), so the score
   * is computed here — when doc review closes and the batch runs — and
   * persisted so the priority board shows the same ranking. The pool arrives
   * SQL-ordered by the old scores; the caller must re-sort after this.
   */
  private async recomputeBulkPriorities(
    pool: Booking[],
    wagonDims: WagonDims,
  ): Promise<void> {
    for (const booking of pool) {
      if (booking.freightType !== 'BULK') continue;
      try {
        const wagons = this.wagonsFor(booking, wagonDims);
        const score = await this.pricingService.computeSubmitPriorityScore(
          booking,
          wagons,
        );
        if (Number(booking.priorityScore ?? 0) === score) continue;
        await this.dataSource
          .getRepository(Booking)
          .update(booking.id, { priorityScore: score });
        booking.priorityScore = score;
      } catch (err) {
        // A failed recompute keeps the stored score — never blocks the batch.
        this.logger.warn(
          `Bulk priority recompute failed for ${booking.reference ?? booking.id}: ` +
            `${(err as Error).message}`,
        );
      }
    }
  }

  /** Restore the batch pool ordering (mirrors findBatchPool's ORDER BY) after scores changed. */
  private resortPoolByPriority(pool: Booking[]): void {
    pool.sort(
      (a, b) =>
        Number(b.isGovernment) - Number(a.isGovernment) ||
        Number(b.priorityScore ?? 0) - Number(a.priorityScore ?? 0) ||
        (a.fullyExecutedAt?.getTime() ?? Infinity) -
          (b.fullyExecutedAt?.getTime() ?? Infinity) ||
        a.createdAt.getTime() - b.createdAt.getTime(),
    );
  }

  /**
   * Wagons a booking occupies. Two axes bind independently and the booking needs
   * enough wagons to satisfy BOTH, so the count is the larger of:
   *
   *   weight — ceil(cargoTons / wagonType.capacityTons), the rated payload
   *   length — TEU geometry, two 20ft to a wagon (container bookings only)
   *
   * The weight axis was missing entirely. A BULK booking carries no container
   * lines, so `containerWagonsForLines` returned 0 and every bulk booking
   * collapsed to a single wagon no matter its tonnage — a 2590T fertilizer
   * booking counted as 1 wagon, and `needFor` then charged 1 tare instead of 37.
   * That under-reported the board and let the fill loop overbook the train.
   */
  private wagonsFor(booking: Booking, wagonDims: WagonDims): number {
    // Stored wagonsRequired is a candidate, never an early return: rows written
    // while sumWagonsRequired hardcoded BULK to 1 wagon are still in the DB, and
    // trusting them charged one tare for a whole bulk consist (a 700T booking on
    // 70T wagons read 700 + 1 tare instead of 700 + 10 tares).
    const stored =
      booking.wagonsRequired && booking.wagonsRequired > 0
        ? Math.ceil(booking.wagonsRequired)
        : 0;

    // TEU-aware: two 20ft share one wagon (half a wagon each). The old fallback
    // summed raw container QUANTITY, so 20×20ft counted as 20 wagons, not 10.
    const byLength = containerWagonsForLines(booking.bookingContainers ?? []);

    const capacityTons = this.dimsFor(booking, wagonDims).capacityTons;
    const cargoTons = bookingCargoTons(booking);
    const byWeight =
      cargoTons > 0 && capacityTons > 0 ? Math.ceil(cargoTons / capacityTons) : 0;

    return Math.max(DEFAULT_WAGONS_PER_BOOKING, stored, byLength, byWeight);
  }

  /**
   * What one booking consumes along all three capacity axes.
   *
   * The weight axis is GROSS — cargo plus the tare of every wagon the booking
   * occupies — because it is spent against the locomotive's pull limit, which
   * governs the whole train and not just its payload. Charging cargo alone let a
   * 37-wagon box-wagon train read 2590T when it really weighed 3522T.
   */
  private needFor(booking: Booking, wagonDims: WagonDims): Capacity {
    const wagons = this.wagonsFor(booking, wagonDims);
    const dims = this.dimsFor(booking, wagonDims);
    return {
      wagons,
      weightTons: bookingGrossWeightTons(
        bookingCargoTons(booking),
        wagons,
        dims.tareWeightTons,
      ),
      lengthMeters: wagons * dims.lengthMeters,
    };
  }

  private fits(need: Capacity, budget: Capacity): boolean {
    return (
      need.wagons <= budget.wagons &&
      need.weightTons <= budget.weightTons &&
      need.lengthMeters <= budget.lengthMeters
    );
  }

  /**
   * Caps for a schedule's train: gross pull weight, train length, and the
   * length-derived wagon slot count (never a fixed 53). Bookings spend against
   * `base` via {@link needFor}, whose weight axis is gross. The locomotive's
   * overage tolerance is returned separately — the corridor budget spends it
   * only to admit a booking whole, never to size a split.
   *
   * Limits come from the LOCOMOTIVE ALONE — the global-rules weight/length
   * caps deliberately do not apply here (a mis-set global row once capped
   * every train at 14m and no export booking could board).
   */
  private async capacityLimits(locomotive: LocomotiveLimits): Promise<TrainLimits> {
    const wagonTypes = await this.loadWagonTypeDimensions();
    const derived = deriveTrainCapacityFromLocomotive(
      {
        maxPullWeightTons: Number(locomotive.maxPullWeightTons),
        maxTrainLengthMeters: Number(locomotive.maxTrainLengthMeters),
        overageToleranceTons: Number(locomotive.overageToleranceTons) || 0,
        overageToleranceMeters: Number(locomotive.overageToleranceMeters) || 0,
      },
      wagonTypes,
    );
    return {
      base: {
        wagons: derived.maxWagonSlots,
        weightTons: derived.baseWeightTons,
        lengthMeters: derived.baseLengthMeters,
      },
      tolerance: {
        weightTons: derived.toleranceTons,
        lengthMeters: derived.toleranceMeters,
      },
    };
  }

  /**
   * Keep schedule.max_wagons aligned with the train's boarding limit. A built
   * train's limit is its physical consist — the wagon count staff marshalled
   * (and may change via adjust-consist). Only schedules WITHOUT a built train
   * fall back to the locomotive's length-derived slot count, where bookings
   * are admitted on length/weight alone and yard staff attach the wagons
   * manually before departure.
   */
  private async syncScheduleMaxWagons(
    schedule: TrainSchedule,
    locomotive: LocomotiveLimits,
  ): Promise<void> {
    const physicalWagons = await this.builtTrainWagonCount(schedule);
    const maxWagons =
      physicalWagons ?? (await this.capacityLimits(locomotive)).base.wagons;
    if ((schedule.maxWagons ?? 0) !== maxWagons) {
      await this.dataSource
        .getRepository(TrainSchedule)
        .update(schedule.id, { maxWagons });
      schedule.maxWagons = maxWagons;
    }
  }

  /**
   * Every active wagon type, so the slot count is derived from the shortest wagon
   * the fleet can actually marshal rather than from an arbitrary two-code sample.
   */
  private async loadWagonTypeDimensions(): Promise<WagonTypeDimensions[]> {
    const types = await this.dataSource
      .getRepository(WagonType)
      .find({ where: { isActive: true } });
    if (types.length) return types.map(wagonTypeDimensionsFromEntity);
    return [
      {
        lengthMeters: DEFAULT_CONTAINER_WAGON_LENGTH_METERS,
        capacityTons: 70,
        tareWeightTons: DEFAULT_CONTAINER_WAGON_TARE_TONS,
      },
      {
        lengthMeters: DEFAULT_BULK_WAGON_LENGTH_METERS,
        capacityTons: 60,
        tareWeightTons: DEFAULT_BULK_WAGON_TARE_TONS,
      },
    ];
  }

  /**
   * Every wagon type keyed by id (drives per-booking dims via the cargo/container
   * type's wagon_type_id FK), plus representative fallbacks per freight type
   * (NW5 flat for containers, CW3 gondola for bulk) for bookings whose type has
   * no wagon type configured yet.
   */
  /** Wagon types are near-static reference data — a short TTL cache spares one
   * table scan per board/detail request without letting edits go stale long. */
  private wagonDimsCache: { value: WagonDims; expiresAt: number } | null = null;

  private async loadWagonDims(): Promise<WagonDims> {
    if (this.wagonDimsCache && this.wagonDimsCache.expiresAt > Date.now()) {
      return this.wagonDimsCache.value;
    }
    const types = await this.dataSource.getRepository(WagonType).find();
    const byCode = new Map(
      types.map((t) => [t.code, wagonTypeDimensionsFromEntity(t)]),
    );
    const byWagonTypeId = new Map(
      types.map((t) => [t.id, wagonTypeDimensionsFromEntity(t)]),
    );
    const nw5 = byCode.get("NW5");
    const cw3 = byCode.get("CW3");
    // capacityTons divides a bulk booking's cargo, so a 0 or missing rated payload
    // must fall back rather than yield an infinite wagon count.
    const payload = (value: number | undefined, fallback: number): number =>
      value && value > 0 ? value : fallback;
    const value: WagonDims = {
      container: {
        lengthMeters: nw5?.lengthMeters ?? DEFAULT_CONTAINER_WAGON_LENGTH_METERS,
        tareWeightTons: nw5?.tareWeightTons ?? DEFAULT_CONTAINER_WAGON_TARE_TONS,
        capacityTons: payload(nw5?.capacityTons, DEFAULT_CONTAINER_WAGON_CAPACITY_TONS),
      },
      bulk: {
        lengthMeters: cw3?.lengthMeters ?? DEFAULT_BULK_WAGON_LENGTH_METERS,
        tareWeightTons: cw3?.tareWeightTons ?? DEFAULT_BULK_WAGON_TARE_TONS,
        capacityTons: payload(cw3?.capacityTons, DEFAULT_BULK_WAGON_CAPACITY_TONS),
      },
      byWagonTypeId,
    };
    this.wagonDimsCache = { value, expiresAt: Date.now() + 60_000 };
    return value;
  }

  /**
   * Dimensions of the wagon type THIS booking rides: bulk resolves through its
   * cargo type's allowed wagon-type list, container through the first container
   * line's — the same list resolution the scheduling planner applies when the
   * paid booking is allocated. Board/fill math measured on a representative
   * wagon while allocation validated the real one let a selected batch flunk
   * the post-payment gross-weight check; sharing the resolution closes that
   * gap. Uses the first configured type (the fill engine has no train context);
   * falls back to the representative dims when the list or relation is absent.
   */
  private dimsFor(booking: Booking, wagonDims: WagonDims): PerWagonDims {
    const fallback =
      booking.freightType === "BULK" ? wagonDims.bulk : wagonDims.container;
    const wagonTypeId =
      booking.freightType === "BULK"
        ? booking.cargoType?.wagonTypes?.[0]?.id
        : (booking.bookingContainers ?? [])
            .flatMap((line) => line.containerType?.wagonTypes ?? [])
            .map((wagonType) => wagonType.id)
            .find((id): id is string => Boolean(id));
    const dims = wagonTypeId ? wagonDims.byWagonTypeId.get(wagonTypeId) : undefined;
    if (!dims) return fallback;
    return {
      ...dims,
      capacityTons: dims.capacityTons > 0 ? dims.capacityTons : fallback.capacityTons,
    };
  }

  /**
   * EVERY wagon-type dimension a booking may ride — its cargo/container type's
   * full allowed (many-to-many) wagon-type list, not just the first like
   * {@link dimsFor}. The remainder placer needs the whole set so a train that
   * stocks a non-primary allowed type still counts as fitting: a container type
   * mapped to both NW5 and (say) NW7 must be measured against whichever a given
   * train actually has free. Deduped by wagon-type id; falls back to the single
   * representative dims when no allowed type is configured.
   */
  private dimsForAllowed(booking: Booking, wagonDims: WagonDims): PerWagonDims[] {
    const fallback =
      booking.freightType === "BULK" ? wagonDims.bulk : wagonDims.container;
    const ids =
      booking.freightType === "BULK"
        ? (booking.cargoType?.wagonTypes ?? []).map((wt) => wt.id)
        : (booking.bookingContainers ?? [])
            .flatMap((line) => line.containerType?.wagonTypes ?? [])
            .map((wt) => wt.id);
    const seen = new Set<string>();
    const dims: PerWagonDims[] = [];
    for (const id of ids) {
      if (!id || seen.has(id)) continue;
      seen.add(id);
      const d = wagonDims.byWagonTypeId.get(id);
      if (d) {
        dims.push({
          ...d,
          capacityTons: d.capacityTons > 0 ? d.capacityTons : fallback.capacityTons,
        });
      }
    }
    return dims.length ? dims : [fallback];
  }

  /**
   * Ordered stop yards of the schedule's route (origin → milestones →
   * destination); the legacy two-stop pseudo-route when milestones are absent.
   */
  private async stopsForSchedule(schedule: TrainSchedule): Promise<string[]> {
    let milestoneYards: string[] | null = null;
    if (schedule.routeId) {
      const milestones = await this.dataSource
        .getRepository(RouteMilestone)
        .find({ where: { routeId: schedule.routeId }, order: { sequenceNo: 'ASC' } });
      if (milestones.length >= 2) milestoneYards = milestones.map((m) => m.yardId);
    }
    return stopYardsFor(
      milestoneYards,
      schedule.originStationId,
      schedule.destinationStationId,
    );
  }

  /**
   * Remaining capacity per corridor edge = hard caps minus what allocated +
   * reserved bookings already use ON THEIR OWN LEGS. A booking riding only
   * Dire→Djibouti leaves the Addis→Dire edges untouched.
   *
   * Two capacity regimes, decided by the schedule's train:
   * - Built train (Train Builder consist with physical wagons): the consist IS
   *   the capacity. Wagon slots = physical wagon count; weight and length are
   *   NOT re-checked here — the builder and adjust-consist already enforced the
   *   locomotive's pull/length limits when the consist was assembled.
   * - No built train (legacy schedules): the locomotive's length-derived slot
   *   count plus its weight/length budgets, as before — yard staff attach the
   *   missing wagons manually before wagon assignment.
   */
  private async remainingBudget(
    schedule: TrainSchedule,
    limits: TrainLimits,
    wagonDims: WagonDims,
  ): Promise<CorridorBudget> {
    const physicalWagons = await this.builtTrainWagonCount(schedule);
    if (physicalWagons != null) {
      limits = {
        base: {
          wagons: physicalWagons,
          weightTons: Number.POSITIVE_INFINITY,
          lengthMeters: Number.POSITIVE_INFINITY,
        },
        tolerance: { weightTons: 0, lengthMeters: 0 },
      };
    }
    // Built trains keep the leg-aware multi-edge corridor too: the wagon
    // planner consumes stock per edge (planWagonsWithStock legs), so a consist
    // wagon serves disjoint legs — capacity freed past an alight yard is real.
    const stops = await this.stopsForSchedule(schedule);
    const budget = new CorridorBudget(stops, limits.base, limits.tolerance);
    const allocated = (schedule.scheduleBookings ?? [])
      .map((sb) => sb.booking)
      .filter((b): b is Booking => Boolean(b));
    const reserved = await this.bookingsRepository.findReservedForSchedule(
      schedule.id,
    );
    for (const b of [...allocated, ...reserved]) {
      budget.subtract(
        this.needFor(b, wagonDims),
        budget.legForYards(b.originYardId, b.destinationYardId),
      );
    }
    return budget;
  }

  /**
   * Physical wagons marshalled in the schedule's built train, or null when the
   * schedule has NO built train and the legacy locomotive-derived capacity must
   * apply. This count is what caps a built train's bookings: 50 wagons coupled
   * → 50 wagon slots, no more.
   *
   * A built train with an EMPTY consist returns 0, NOT null: zero coupled
   * wagons means zero capacity. Folding that case into null used to hand an
   * un-consisted train the abstract locomotive budget, so an empty train
   * advertised its full maxWagons as free space and accepted bookings the
   * allocator could never place.
   */
  private async builtTrainWagonCount(
    schedule: TrainSchedule,
  ): Promise<number | null> {
    const trainId = schedule.trainSet?.train?.id;
    if (!trainId) return null;
    return this.dataSource.getRepository(Wagon).count({ where: { trainId } });
  }

  /**
   * Wagon slots still boardable somewhere on the corridor (most-open edge).
   * ≤ 0 means no leg can take another booking. Slot axis ONLY — the train-wide
   * FULL signal is {@link isTrainFull}, which also closes weight/length-bound
   * trains that still show free slots.
   */
  private async remainingWagons(schedule: TrainSchedule): Promise<number> {
    const wagonDims = await this.loadWagonDims();
    const budget = await this.remainingBudget(
      schedule,
      {
        base: {
          wagons: schedule.maxWagons ?? 0,
          weightTons: Number.POSITIVE_INFINITY,
          lengthMeters: Number.POSITIVE_INFINITY,
        },
        tolerance: { weightTons: 0, lengthMeters: 0 },
      },
      wagonDims,
    );
    return budget.maxRemaining().wagons;
  }

  async setWindow(
    scheduleId: string,
    status: "OPEN" | "FULL" | "CLOSED",
  ): Promise<void> {
    await this.dataSource
      .getRepository(TrainSchedule)
      .update(scheduleId, { bookingWindowStatus: status });
    // Push the change (open / train full / closed) so portal home and GL cards
    // flip in real time — FULL in particular happens outside the window tick
    // (batch fill, staff mark-paid) and had no live signal before.
    try {
      const fresh = await this.trainSchedulesRepository.findById(scheduleId);
      if (fresh) this.bookingWindowGateway.emitPhase(fresh);
    } catch (err) {
      this.logger.warn(
        `Booking-window push failed for ${scheduleId}: ${(err as Error).message}`,
      );
    }
  }

  /**
   * A reservation on this schedule still has time left to pay.
   *
   * The PAYMENT phase ends a hair BEFORE its own reservations do: `paymentPhaseEndsAt`
   * is stamped when the phase starts, then `reserve()` gives each booking
   * `now + paymentWindow` a few hundred milliseconds later, one booking at a time. So
   * the first settle after the phase deadline finds every reservation still in date,
   * expires nothing, reports `anySettled = false`, runs no top-up — and the caller
   * concludes the cycle out from under customers who still had time to pay. The next
   * tick then expires them with no cycle left to promote the waiting list into.
   *
   * Callers must not conclude the cycle while this returns true.
   */
  async hasLiveReservations(scheduleId: string): Promise<boolean> {
    const reserved =
      await this.bookingsRepository.findReservedForSchedule(scheduleId);
    const now = Date.now();
    return reserved.some(
      (b) =>
        b.paymentStatus !== "PAID" &&
        b.status !== "PAID" &&
        b.paymentDeadline != null &&
        b.paymentDeadline.getTime() > now,
    );
  }

  /**
   * FULL is DIRECTIONAL: the schedule's trade direction is full when the
   * border-crossing edge (which every export/import must ride) can't take one
   * more minimal wagon on any axis — slots for built trains (the consist is
   * the capacity, weight/length settled at build), all three axes otherwise
   * (PW2: weight binds at 37 wagons = 3522.4T of 3500+90T, slots bind at 44).
   * Home-side legs may still run empty; intercity ride-alongs keep filling
   * them via the per-leg budget and never consult this flag. Domestic routes
   * (no border) are full only when every edge is closed.
   */
  async isScheduleFull(scheduleId: string): Promise<boolean> {
    const schedule =
      await this.trainSchedulesRepository.findByIdWithFullGraph(scheduleId);
    if (!schedule) return false;
    return this.isTrainFull(schedule);
  }

  /**
   * Wagon-slot usage snapshot for staff UIs (adjust-consist dialog): the
   * schedule's slot capacity, how many slots allocated + reserved bookings
   * already hold on the busiest edge, how many are still free on the most-open
   * edge, and by how many slots the consist has been trimmed BELOW what is
   * already committed (0 when nothing is over-allocated).
   */
  async scheduleWagonUsage(scheduleId: string): Promise<{
    maxWagons: number;
    allocatedWagons: number;
    remainingSlots: number;
    overAllocatedBy: number;
  } | null> {
    const schedule =
      await this.trainSchedulesRepository.findByIdWithFullGraph(scheduleId);
    if (!schedule) return null;
    const capacity =
      (await this.builtTrainWagonCount(schedule)) ?? schedule.maxWagons ?? 0;
    const wagonDims = await this.loadWagonDims();
    const budget = await this.remainingBudget(
      schedule,
      {
        base: {
          wagons: capacity,
          weightTons: Number.POSITIVE_INFINITY,
          lengthMeters: Number.POSITIVE_INFINITY,
        },
        tolerance: { weightTons: 0, lengthMeters: 0 },
      },
      wagonDims,
    );
    const tightest = budget.remainingFor(budget.fullLeg()).wagons;
    return {
      maxWagons: capacity,
      allocatedWagons: capacity - tightest,
      remainingSlots: Math.max(0, budget.maxRemaining().wagons),
      overAllocatedBy: Math.max(0, -tightest),
    };
  }

  /** See {@link isScheduleFull} — same check for callers that already hold the full graph. */
  private async isTrainFull(schedule: TrainSchedule): Promise<boolean> {
    // "Full" means full FOR THE TRAIN'S TRADE DIRECTION. Every export and
    // every import must cross the ET↔DJ border edge, so once that edge can't
    // take one more minimal wagon the booking window may close — even while
    // home-side legs still run empty. Intercity ride-alongs never consult this
    // flag; they keep booking the free legs through the per-leg budget.
    // A single-country (domestic) corridor has no mandatory edge, so it is
    // full only when EVERY edge is closed on some axis.
    const wagonDims = await this.loadWagonDims();
    const physicalWagons = await this.builtTrainWagonCount(schedule);
    let limits: TrainLimits;
    if (physicalWagons != null) {
      // The consist is the capacity; weight/length were settled at build time.
      // remainingBudget swaps in the physical wagon count per edge itself.
      limits = {
        base: {
          wagons: physicalWagons,
          weightTons: Number.POSITIVE_INFINITY,
          lengthMeters: Number.POSITIVE_INFINITY,
        },
        tolerance: { weightTons: 0, lengthMeters: 0 },
      };
    } else {
      const locomotive = trainSetLocomotiveLimits(schedule.trainSet);
      // No loco, no built train: only the slot axis exists to bind against.
      if (!locomotive) return (await this.remainingWagons(schedule)) <= 0;
      limits = await this.capacityLimits(locomotive);
    }
    const budget = await this.remainingBudget(schedule, limits, wagonDims);
    const minNeed = this.minPerWagonNeed(wagonDims);
    const border = await this.borderLeg(budget.stops);
    if (border) {
      return !budget.fits(
        {
          wagons: 1,
          weightTons: minNeed.grossWeightTons,
          lengthMeters: minNeed.lengthMeters,
        },
        border,
      );
    }
    return budget.isExhausted(minNeed);
  }

  /**
   * The corridor's single border-crossing edge (last home-country stop → first
   * far-country stop), or null when every stop is in one country. This is the
   * edge every EXPORT and IMPORT booking must ride, whichever sub-corridor it
   * books — which makes it the train's directional fullness gauge.
   */
  private async borderLeg(stops: string[]): Promise<CorridorLeg | null> {
    if (stops.length < 2) return null;
    const yards = await this.dataSource
      .getRepository(Yard)
      .find({ where: { id: In(stops) } });
    const countryOf = new Map(yards.map((y) => [y.id, y.country]));
    const first = countryOf.get(stops[0]);
    if (!first) return null;
    const crossIdx = stops.findIndex((id) => {
      const country = countryOf.get(id);
      return country != null && country !== first;
    });
    if (crossIdx <= 0) return null;
    return { fromEdge: crossIdx - 1, toEdge: crossIdx };
  }

  /**
   * Smallest gross weight / shortest length one more wagon could add: the
   * lightest wagon type at its rated payload. Feeds CorridorBudget.isExhausted,
   * so FULL is only declared when not even this wagon fits anywhere.
   */
  private minPerWagonNeed(wagonDims: WagonDims): {
    grossWeightTons: number;
    lengthMeters: number;
  } {
    const all = [
      wagonDims.container,
      wagonDims.bulk,
      ...wagonDims.byWagonTypeId.values(),
    ];
    return {
      grossWeightTons: Math.min(
        ...all.map((d) => d.tareWeightTons + d.capacityTons),
      ),
      lengthMeters: Math.min(...all.map((d) => d.lengthMeters)),
    };
  }

  /**
   * Re-derive `bookingWindowStatus` from live capacity after wagons were freed
   * (a reservation expired, a booking was displaced, a link was removed).
   *
   * FULL used to be a one-way door: `isFillable()` rejects a FULL schedule before
   * it ever looks at the budget, and the only writers of OPEN skip a FULL row. So
   * a train that filled once and then lost every booking to expiry stayed FULL
   * with all its wagons free — permanently unfillable, cycling PRE_WINDOW→PAYMENT
   * forever while `concludeCycle` (which reads real capacity, not the flag) kept
   * reopening it. Clearing FULL here is what lets the next batch actually run.
   *
   * Only the customer-facing OPEN phases may go back to OPEN; a schedule mid
   * DOC_REVIEW/PAYMENT drops to CLOSED, which `isFillable()` still admits.
   */
  async refreshWindowStatus(scheduleId: string): Promise<void> {
    const schedule =
      await this.trainSchedulesRepository.findByIdWithFullGraph(scheduleId);
    if (!schedule || schedule.bookingWindowStatus !== "FULL") return;
    // Symmetric with isScheduleFull: a weight/length-bound FULL is not stale
    // just because slots remain — clearing it here would reopen a train
    // nothing can board.
    if (await this.isTrainFull(schedule)) return;

    const customerWindowOpen =
      schedule.windowPhase == null || schedule.windowPhase === "OPEN";
    await this.setWindow(scheduleId, customerWindowOpen ? "OPEN" : "CLOSED");
    this.logger.log(
      `[BATCH] ${scheduleId} cleared stale FULL — wagons freed, window is now ` +
        `${customerWindowOpen ? "OPEN" : "CLOSED"} and the batch can fill it again`,
    );
  }

  // ---- timer plumbing -------------------------------------------------------

  /** Configured customer pay window in ms (global rules, with defaults). */
  private async paymentWindowMs(): Promise<number> {
    const cfg = await this.trainSchedulingService.getWindowConfig();
    return cfg.paymentWindowMinutes * 60_000;
  }

  private timeoutName(scheduleId: string): string {
    return `settle:${scheduleId}`;
  }

  /**
   * In-process accelerator only — the durable settle enforcement is the window
   * engine's minute tick calling settleDueReservations off `paymentDeadline`.
   */
  private armSettle(scheduleId: string): void {
    void this.paymentWindowMs()
      .then((delayMs) => {
        this.removeTimeout(scheduleId);
        const handle = setTimeout(() => {
          void this.settleBatch(scheduleId).catch((err) =>
            this.logger.error(
              `settleBatch ${scheduleId} failed: ${(err as Error).message}`,
            ),
          );
        }, delayMs);
        this.scheduler.addTimeout(this.timeoutName(scheduleId), handle);
      })
      .catch((err) =>
        this.logger.warn(
          `armSettle ${scheduleId} skipped: ${(err as Error).message}`,
        ),
      );
  }

  /**
   * A top-up reservation (settle freed capacity mid-cycle, so the next waiting
   * booking got a fresh pay window) sets a NEW paymentDeadline. But the schedule's
   * `paymentPhaseEndsAt` — which the window tick watches to end PAYMENT and run
   * concludeCycle — was frozen when the phase started. Without this, concludeCycle
   * fires before the top-up customer's deadline and expires a booking that still
   * had time to pay. Push `paymentPhaseEndsAt` to at least cover a full payment
   * window from now, but never past departure. Only while the schedule is still
   * in the PAYMENT phase (a reopened cycle manages its own phase).
   */
  async extendPaymentPhaseForTopUp(scheduleId: string): Promise<void> {
    const schedule = await this.dataSource
      .getRepository(TrainSchedule)
      .findOne({ where: { id: scheduleId } });
    if (!schedule || schedule.windowPhase !== "PAYMENT") return;
    const windowMs = await this.paymentWindowMs();
    let target = new Date(Date.now() + windowMs);
    if (
      schedule.scheduledDepartureDate &&
      target > schedule.scheduledDepartureDate
    ) {
      target = schedule.scheduledDepartureDate;
    }
    // Only ever push the deadline OUT, never pull it in.
    if (
      schedule.paymentPhaseEndsAt &&
      schedule.paymentPhaseEndsAt.getTime() >= target.getTime()
    ) {
      return;
    }
    await this.dataSource
      .getRepository(TrainSchedule)
      .update(scheduleId, { paymentPhaseEndsAt: target });
    this.logger.log(
      `[BATCH] extended PAYMENT phase for ${scheduleId} to ${target.toISOString()} ` +
        `(top-up reservation opened a fresh pay window)`,
    );
  }

  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
    }
  }
}