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

import { Booking } from '../bookings/entities/booking.entity';
import { BookingsRepository } from '../bookings/bookings.repository';
import { BookingPricingService } from '../bookings/booking-pricing.service';
import { Locomotive } from '../locomotives/entities/locomotive.entity';
import { formatRouteLabel } from '../routes/entities/route.entity';
import { RouteMilestone } from '../routes/entities/route-milestone.entity';
import { TrainSchedule } from '../train-schedules/entities/train-schedule.entity';
import { TrainScheduleBooking } from '../train-schedules/entities/train-schedule-booking.entity';
import { TrainSchedulesRepository } from '../train-schedules/train-schedules.repository';
import { TrainScheduleBookingsRepository } from '../train-schedules/train-schedule-bookings.repository';
import { TrainSchedulingGlobalRules } from './entities/train-scheduling-global-rules.entity';
import { BookingNotifierService } from './booking-notifier.service';
import { TrainSchedulingService } from './train-scheduling.service';
import { eatDay, groupBookingsIntoBoardWindows } from './batch-window.util';
import { Freight, TrainScheduleStatus as TrainScheduleStatusEnum } from "@edr/types";
import { BillingService } from "../billing/billing.service";


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 {
  WagonTypeDimensions,
  bookingGrossWeightTons,
  bookingTrainLengthMeters,
  deriveTrainCapacityFromLocomotive,
  sizePartialOfferWagons,
  trainHardCaps,
  wagonTypeDimensionsFromEntity,
} from './train-capacity.util';
import { WagonType } from '../wagon-types/entities/wagon-type.entity';
import { ClearanceMilestoneService } from '../contracts/clearance-milestone.service';
import { BookingSplitService } from './booking-split.service';
import { BookingWindowGateway } from './booking-window.gateway';
import {
  MAX_TEU_SLOTS_PER_WAGON,
  containerWagonsForLines,
} from './wagon-plan.util';
import {
  Capacity,
  CorridorBudget,
  CorridorLeg,
  stopYardsFor,
} from './corridor-capacity.util';

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

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

/**
 * Per-freight-type wagon dimensions used to size a booking's capacity draw:
 * its length on the train and the tare it adds to the locomotive's gross load.
 */
type WagonDims = {
  container: { lengthMeters: number; tareWeightTons: number; capacityTons: number };
  bulk: { lengthMeters: number; tareWeightTons: number; capacityTons: number };
};

export type BatchBoardBookingState =
  | "ALLOCATED"
  | "SELECTED_FOR_BATCH"
  | "READY"
  | "WAITING"
  | "PENDING_CONTRACT"
  | "EXPIRED";

export interface BatchBoardBooking {
  id: string;
  reference: string;
  company: string;
  isGovernment: boolean;
  wagons: number;
  weightTons: number;
  lengthMeters: number;
  paymentDeadline: string | null;
  state: BatchBoardBookingState;
  /** 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 BatchWindowGroup {
  key: string;
  label: string;
  /** EAT calendar day as ISO `YYYY-MM-DD` (empty for the pending-contract bucket). */
  date: string;
  /** Human label for the day, e.g. `Thu, 05 Jun` (empty for pending-contract). */
  dateLabel: string;
  start: string;
  end: string;
  counts: {
    allocated: number;
    selectedForBatch: number;
    ready: number;
    waiting: number;
    expired: number;
    pendingContract: number;
  };
  bookings: BatchBoardBookingDetail[];
}

export interface BatchBoardScheduleDetail {
  scheduleId: string;
  trainNumber: string | null;
  routeName: string | null;
  origin: string | null;
  destination: string | null;
  scheduleDate: string | null;
  status: string;
  bookingWindowStatus: string;
  direction: string | null;
  windowPhase: string | null;
  windowOpensAt: string | null;
  windowClosesAt: string | null;
  docReviewEndsAt: string | null;
  paymentPhaseEndsAt: string | null;
  bookingCycleNo: number;
  locomotive: BatchBoardSchedule["locomotive"];
  capacity: BatchBoardSchedule["capacity"];
  counts: BatchBoardSchedule["counts"];
  windows: BatchWindowGroup[];
  pendingContract: BatchWindowGroup;
  allocationViolations: string[];
}

export interface BatchBoardSchedule {
  scheduleId: string;
  trainNumber: string | null;
  routeName: string | null;
  origin: string | null;
  destination: string | null;
  scheduleDate: string | null;
  status: string;
  bookingWindowStatus: string;
  direction: string | null;
  windowPhase: string | null;
  windowOpensAt: string | null;
  windowClosesAt: string | null;
  docReviewEndsAt: string | null;
  paymentPhaseEndsAt: string | null;
  bookingCycleNo: number;
  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[];
}

/**
 * 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,
    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,

  ) {}

  /** 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?.trainScheduleId) return;

    const isBatchPaid =
      booking.status === "SELECTED_FOR_BATCH" ||
      booking.status === "AWAITING_PAYMENT" ||
      booking.status === "PAID" ||
      booking.paymentStatus === "PAID";
    if (!isBatchPaid) return;

    if (
      booking.status === "SELECTED_FOR_BATCH" ||
      booking.status === "AWAITING_PAYMENT"
    ) {
      await this.dataSource
        .getRepository(Booking)
        .update(bookingId, { paymentStatus: "PAID", status: "PAID" });
    } else if (booking.paymentStatus !== "PAID") {
      await this.dataSource
        .getRepository(Booking)
        .update(bookingId, { paymentStatus: "PAID" });
    }

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

    const linked =
      await this.trainScheduleBookingsRepository.existsForBooking(bookingId);
    if (!linked) {
      await this.allocate(booking.trainScheduleId, booking, "paid");
      this.logger.log(
        `Linked PAID booking ${booking.reference ?? bookingId} to schedule ${booking.trainScheduleId}`,
      );
    } 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.remainingWagons(schedule)) <= 0) {
      await this.setWindow(booking.trainScheduleId, "FULL");
    }

    const result = await this.trainSchedulingService.tryAutoWagonAllocation(
      booking.trainScheduleId,
    );
    if (result.assignedBookingIds.length) {
      this.logger.log(
        `Wagon allocation for ${booking.reference ?? bookingId}: ${result.assignedBookingIds.length} assigned`,
      );
    }
    if (
      result.issues.some(
        (i) => i.bookingId === bookingId && i.status !== "ASSIGNED",
      )
    ) {
      const issue = result.issues.find((i) => i.bookingId === bookingId);
      this.logger.warn(
        `Wagon allocation issue for ${booking.reference ?? bookingId}: ${issue?.issue ?? issue?.status}`,
      );
    }
  }

  /** Customer paid — delegate to ensurePaidBookingAllocated. */
  async confirmPaidAndAllocate(bookingId: string): Promise<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 -----------------------------------------------------------

  /**
   * 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> {
    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(),
      );
    if (!candidates.length) {
      throw new ConflictException(
        'No export train is accepting bookings for this day',
      );
    }

    const rules = await this.loadGlobalRules();
    const wagonDims = await this.loadWagonDims();
    const required = need ?? this.needFor(booking, wagonDims);
    let corridorMatched = false;
    for (const candidate of candidates) {
      const schedule = await this.trainSchedulesRepository.findByIdWithFullGraph(
        candidate.id,
      );
      const locomotive = schedule?.trainSet?.locomotive;
      if (!schedule || !locomotive) continue;
      const limits = await this.capacityLimits(locomotive, rules);
      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
      corridorMatched = true;
      if (budget.fits(required, leg)) return schedule.id;
    }
    if (!corridorMatched) {
      throw new ConflictException(
        'No export train is accepting bookings for this day',
      );
    }
    throw new ConflictException('Train is full — no export capacity left for this day');
  }

  /**
   * 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) {
      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: 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> {
    for (const b of bookings) await this.reserve(b, scheduleId);
    this.armSettle(scheduleId);
    const schedule =
      await this.trainSchedulesRepository.findByIdWithFullGraph(scheduleId);
    if (schedule && (await this.remainingWagons(schedule)) <= 0) {
      await this.setWindow(scheduleId, 'FULL');
    }
  }

  /** 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) {
      await this.allocate(scheduleId, booking, "paid");
      this.logger.log(
        `Reconciled PAID booking ${booking.reference ?? booking.id} → schedule ${scheduleId}`,
      );
    }
  }

  // ---- 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 still-relevant schedule (not arrived/
   * cancelled) with its locomotive, capacity usage and its bookings grouped by lifecycle
   * state (allocated / awaiting payment / paid-waiting / pending contract / expired).
   */
  async getBatchBoard(): Promise<BatchBoardSchedule[]> {
    const schedules = await this.trainSchedulesRepository.findAll({
      relations: {
        trainSet: { locomotive: true },
        originStation: true,
        destinationStation: true,
        // Yards supply the route's display name for `routeName` below.
        route: { originYard: true, destinationYard: true },
      },
      order: { scheduledDepartureDate: "ASC" },
    });

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

    const board: BatchBoardSchedule[] = [];
    for (const s of schedules) {
      if (s.status === "ARRIVED" || s.status === "CANCELLED") continue;
      // 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.
      if (s.direction !== "IMPORT") continue;

      const links = await linkRepo.find({ where: { trainScheduleId: s.id } });
      const linkedIds = new Set(links.map((l) => l.bookingId));
      const bookings = await this.bookingsRepository.findAllBySchedule(s.id);

      const items: BatchBoardBooking[] = bookings.map((b) => {
        const need = this.needFor(b, 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, rules));
    }
    return board;
  }

  /** Schedule-level batch board with EAT 3h windows grouped by fullyExecutedAt. */
  async getBatchBoardDetail(
    scheduleId: string,
  ): Promise<BatchBoardScheduleDetail> {
    const s =
      await this.trainSchedulesRepository.findByIdWithFullGraph(scheduleId);
    if (!s)
      throw new NotFoundException(`Train schedule ${scheduleId} not found`);
    if (s.status === "ARRIVED" || s.status === "CANCELLED") {
      throw new BadRequestException("Schedule is no longer active");
    }
    // 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();
    const rules = await this.loadGlobalRules();
    const linkRepo = this.dataSource.getRepository(TrainScheduleBooking);
    const links = await linkRepo.find({ where: { trainScheduleId: s.id } });
    const linkedIds = new Set(links.map((l) => l.bookingId));
    const bookings = await this.bookingsRepository.findAllBySchedule(s.id);

    let allocationPreview: Awaited<
      ReturnType<TrainSchedulingService["previewAllocationForSchedule"]>
    >;
    try {
      allocationPreview =
        await this.trainSchedulingService.previewAllocationForSchedule(s.id);
    } catch {
      allocationPreview = {
        assignedBookingIds: [],
        deferred: [],
        issues: [],
        violations: [],
      };
    }
    const allocationByBooking = new Map(
      allocationPreview.issues.map((i) => [i.bookingId, i]),
    );

    // 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 = s.trainSet?.locomotive ?? null;

    // Display windows are the REAL booking-window cycles this schedule was FROZEN
    // with at creation (import: opens at its stored window time, lasts its rule's
    // duration, reopens per its rule's delay; export: single FCFS lead window) —
    // NOT the live global config. A later global-rules edit only re-derives
    // not-yet-open schedules (restampPendingWindows), so an already-open schedule
    // must keep drawing from its own snapshot, anchored on its stored open time.
    // Legacy rows with no snapshot fall back to the live config.
    const liveCfg = await this.trainSchedulingService.getWindowConfig();
    const num = (v: unknown, fallback: number) => {
      const n = v == null ? NaN : Number(v);
      return Number.isFinite(n) ? n : fallback;
    };
    const windowCfg = {
      windowOpenHour: num(s.ruleWindowOpenHour, liveCfg.windowOpenHour),
      windowCloseHour: num(s.ruleWindowCloseHour, liveCfg.windowCloseHour),
      windowDurationHours: num(
        s.ruleWindowDurationHours,
        liveCfg.windowDurationHours,
      ),
      reopenDelayMinutes: num(s.ruleReopenDelayMinutes, liveCfg.reopenDelayMinutes),
      importWindowLeadDays: num(
        s.ruleImportWindowLeadDays,
        liveCfg.importWindowLeadDays,
      ),
      exportBookingLeadHours: num(
        s.ruleExportBookingLeadHours,
        liveCfg.exportBookingLeadHours,
      ),
    };
    const departureDate = s.scheduledDepartureDate ?? new Date();
    const windowBuckets = groupBookingsIntoBoardWindows(
      items,
      (item) => (item.fullyExecutedAt ? new Date(item.fullyExecutedAt) : null),
      s.direction ?? null,
      departureDate,
      windowCfg,
      undefined,
      s.windowOpensAt ?? null,
    );

    const emptyCounts = () => ({
      allocated: 0,
      selectedForBatch: 0,
      ready: 0,
      waiting: 0,
      expired: 0,
      pendingContract: 0,
    });

    const countFor = (bookingsInWindow: BatchBoardBookingDetail[]) => {
      const counts = emptyCounts();
      for (const b of bookingsInWindow) {
        if (b.state === "ALLOCATED") counts.allocated += 1;
        else if (b.state === "SELECTED_FOR_BATCH") counts.selectedForBatch += 1;
        else if (b.state === "READY") counts.ready += 1;
        else if (b.state === "WAITING") counts.waiting += 1;
        else if (b.state === "EXPIRED") counts.expired += 1;
        else counts.pendingContract += 1;
      }
      return counts;
    };

    const windows: BatchWindowGroup[] = [];
    for (const [key, bucket] of windowBuckets) {
      if (key === "pending-contract" || !bucket.window) continue;
      const w = bucket.window;
      windows.push({
        key: w.key,
        label: w.label,
        date: w.date,
        dateLabel: w.dateLabel,
        start: w.start.toISOString(),
        end: w.end.toISOString(),
        counts: countFor(bucket.items),
        bookings: bucket.items,
      });
    }
    windows.sort(
      (a, b) => new Date(a.start).getTime() - new Date(b.start).getTime(),
    );

    const pendingBookings = windowBuckets.get("pending-contract")?.items ?? [];

    return {
      scheduleId: s.id,
      trainNumber: s.trainNumber ?? null,
      routeName: s.route ? formatRouteLabel(s.route) : null,
      origin: s.originStation?.label ?? s.originStation?.code ?? null,
      destination:
        s.destinationStation?.label ?? s.destinationStation?.code ?? null,
      scheduleDate: s.scheduledDepartureDate
        ? s.scheduledDepartureDate.toISOString()
        : null,
      status: s.status,
      bookingWindowStatus: s.bookingWindowStatus,
      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,
      locomotive: loco
        ? {
          code: loco.code,
          name: loco.name ?? null,
          maxPullWeightTons: Number(loco.maxPullWeightTons),
          maxTrainLengthMeters: Number(loco.maxTrainLengthMeters),
        }
        : null,
      capacity: this.computeBoardCapacity(items, loco, s.maxWagons ?? null, rules),
      counts: {
        allocated: items.filter((i) => i.state === "ALLOCATED").length,
        selectedForBatch: items.filter((i) => i.state === "SELECTED_FOR_BATCH")
          .length,
        ready: items.filter((i) => i.state === "READY").length,
        waiting: items.filter((i) => i.state === "WAITING").length,
        pendingContract: items.filter((i) => i.state === "PENDING_CONTRACT")
          .length,
        expired: items.filter((i) => i.state === "EXPIRED").length,
      },
      windows,
      pendingContract: {
        key: "pending-contract",
        label: "Pending contract",
        date: "",
        dateLabel: "",
        start: "",
        end: "",
        counts: countFor(pendingBookings),
        bookings: pendingBookings,
      },
      allocationViolations: allocationPreview.violations,
    };
  }

  /** Run wagon-level allocation for all eligible linked bookings on a schedule. */
  async runWagonAllocation(scheduleId: string) {
    return this.trainSchedulingService.tryAutoWagonAllocation(scheduleId);
  }

  /**
   * 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 floored
   * by the global rule caps and widened by its overage tolerance. 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: Locomotive | null,
    maxWagons: number | null,
    rules: TrainSchedulingGlobalRules | null,
  ): BatchBoardSchedule["capacity"] {
    const allocated = items.filter((i) => i.state === "ALLOCATED");
    const committed = items.filter(
      (i) => i.state === "ALLOCATED" || i.state === "SELECTED_FOR_BATCH",
    );
    const caps = loco
      ? trainHardCaps(
          {
            maxPullWeightTons: Number(loco.maxPullWeightTons),
            maxTrainLengthMeters: Number(loco.maxTrainLengthMeters),
            overageToleranceTons: Number(loco.overageToleranceTons) || 0,
            overageToleranceMeters: Number(loco.overageToleranceMeters) || 0,
          },
          {
            maxTrainWeightTons: rules?.maxTrainWeightTons
              ? Number(rules.maxTrainWeightTons)
              : undefined,
            maxTrainLengthMeters: rules?.maxTrainLengthMeters
              ? Number(rules.maxTrainLengthMeters)
              : undefined,
          },
        )
      : 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[],
    rules: TrainSchedulingGlobalRules | null,
  ): BatchBoardSchedule {
    const loco = s.trainSet?.locomotive ?? null;

    return {
      scheduleId: s.id,
      trainNumber: s.trainNumber ?? null,
      routeName: s.route ? formatRouteLabel(s.route) : null,
      origin: s.originStation?.label ?? s.originStation?.code ?? null,
      destination:
        s.destinationStation?.label ?? s.destinationStation?.code ?? null,
      scheduleDate: s.scheduledDepartureDate
        ? s.scheduledDepartureDate.toISOString()
        : null,
      status: s.status,
      bookingWindowStatus: s.bookingWindowStatus,
      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,
      locomotive: loco
        ? {
          code: loco.code,
          name: loco.name ?? null,
          maxPullWeightTons: Number(loco.maxPullWeightTons),
          maxTrainLengthMeters: Number(loco.maxTrainLengthMeters),
        }
        : null,
      capacity: this.computeBoardCapacity(items, loco, s.maxWagons ?? null, rules),
      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 = schedule.trainSet?.locomotive;
    if (!schedule.trainSetId || !locomotive) {
      this.logger.warn(
        `Schedule ${scheduleId} has no locomotive/train set — skipped.`,
      );
      return 0;
    }

    const rules = await this.loadGlobalRules();
    const wagonDims = await this.loadWagonDims();
    const limits = await this.capacityLimits(locomotive, rules);
    await this.syncScheduleMaxWagons(schedule, locomotive, rules);
    const budget = await this.remainingBudget(schedule, limits, wagonDims);
    if (budget.maxRemaining().wagons <= 0) {
      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 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,
          );
          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.maxRemaining().wagons <= 0) await this.setWindow(scheduleId, "FULL");
    if (armed) this.armSettle(scheduleId);
    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 rules = await this.loadGlobalRules();
    const wagonDims = await this.loadWagonDims();

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

    for (const t of trains) {
      if (t.budget.maxRemaining().wagons <= 0) await this.setWindow(t.id, "FULL");
      if (t.armed) this.armSettle(t.id);
      void this.triggerWagonAllocation(t.id);
    }

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

  /**
   * A lone commercial IMPORT 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).
   */
  private isSplitEligible(booking: Booking, isPair: boolean): boolean {
    return (
      !isPair &&
      !booking.isGovernment &&
      booking.tradeDirection === "IMPORT" &&
      (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();
    const bulkCapacityTons = await this.loadBulkWagonCapacityTons();

    // 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.
    const perWagon =
      booking.freightType === "BULK" ? wagonDims.bulk : wagonDims.container;
    const partial = sizePartialOfferWagons(budget, need.wagons, perWagon);
    if (!partial) return null;

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

    const offeredNeed: Capacity = {
      wagons: sized.offeredWagons,
      weightTons: bookingGrossWeightTons(
        sized.offeredWeightTons,
        sized.offeredWagons,
        this.tareFor(booking.freightType, wagonDims),
      ),
      lengthMeters: bookingTrainLengthMeters(
        booking.freightType,
        sized.offeredWagons,
        this.lengthsOf(wagonDims),
      ),
    };
    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;
  }

  private async loadBulkWagonCapacityTons(): Promise<number> {
    const cw3 = await this.dataSource
      .getRepository(WagonType)
      .findOne({ where: { code: "CW3" } });
    const capacity = cw3 ? wagonTypeDimensionsFromEntity(cw3).capacityTons : 60;
    return capacity > 0 ? capacity : 60;
  }

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

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

  // ---- staff override actions ----------------------------------------------

  /** Staff "mark paid" override → set PAID and allocate immediately (don't wait for settle). */
  async markPaid(bookingId: string): Promise<void> {
    const booking = await this.dataSource
      .getRepository(Booking)
      .findOne({ where: { id: bookingId } });
    if (!booking) throw new NotFoundException(`Booking ${bookingId} not found`);
    if (!booking.trainScheduleId) {
      throw new BadRequestException(
        "Booking has no target schedule to allocate to",
      );
    }
    await this.dataSource
      .getRepository(Booking)
      .update(bookingId, { paymentStatus: "PAID" });
    await this.allocate(booking.trainScheduleId, booking, "paid");

    const schedule = await this.trainSchedulesRepository.findByIdWithFullGraph(
      booking.trainScheduleId,
    );
    if (schedule && (await this.remainingWagons(schedule)) <= 0) {
      await this.setWindow(booking.trainScheduleId, "FULL");
    }
    void this.triggerWagonAllocation(booking.trainScheduleId!);
  }

  /**
   * Re-point a booking to another OPEN same-route schedule (keeps approval/contract + priority).
   * Used for EXPIRED or full-schedule bookings — no re-approval.
   */
  async moveToSchedule(
    bookingId: string,
    newScheduleId: string,
  ): Promise<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",
      );
    }

    await this.dataSource.transaction(async (manager) => {
      if (booking.trainScheduleId) {
        await this.trainScheduleBookingsRepository.deleteByScheduleAndBooking(
          booking.trainScheduleId,
          bookingId,
          manager,
        );
      }
      const restoredStatus =
        booking.status === "EXPIRED"
          ? booking.isGovernment
            ? "APPROVED"
            : "FULLY_EXECUTED"
          : booking.status;
      await manager.getRepository(Booking).update(bookingId, {
        trainScheduleId: newScheduleId,
        status: restoredStatus,
        schedulingStatus: "ELIGIBLE",
        paymentDeadline: null,
        selectedForBatchAt: null,
      } as never);
    });
  }

  /** Staff "expire" override → free a reservation now (booking becomes EXPIRED). */
  async expireReservation(bookingId: string): Promise<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);
      }
    }
  }

  // ---- 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 = schedule?.trainSet?.locomotive;
    if (!schedule || !locomotive) return null;
    const rules = await this.loadGlobalRules();
    const wagonDims = await this.loadWagonDims();
    const limits = await this.capacityLimits(locomotive, rules);
    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');
      return;
    }
    await this.reserve(booking, scheduleId);
    this.armSettle(scheduleId);
  }

  // ---- 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();
    const deadline = new Date(now.getTime() + (await this.paymentWindowMs()));
    await this.bookingsRepository.update(booking.id, {
      trainScheduleId: scheduleId,
      status: "SELECTED_FOR_BATCH",
      selectedForBatchAt: now,
      paymentDeadline: deadline,
    } as never);
    booking.trainScheduleId = scheduleId;
    // The invoice was generated at booking creation/approval, before this pay
    // window opened — refresh its printed due date to the real deadline.
    await this.billing.syncPayableDueDate(
      Freight.InvoiceSource.Booking,
      booking.id,
      deadline,
      "PREPAID",
    );
    await this.notifier.payNow(booking, deadline);
    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). */
  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);
    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.
   */
  private async expire(booking: Booking): Promise<void> {
    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");
    this.notifier.expired(booking);
    this.logger.log(
      `[BATCH] EXPIRED ${booking.reference} — payment window passed; freed its ` +
        `wagons back to the pool for top-up`,
    );
  }

  /**
   * 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}`,
      );
    }
    for (const booking of unaccepted) {
      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`,
      );
    }
  }

  /**
   * 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 =
      Number(primary.cargoTotalWeightVgm ?? 0) + Number(partner.cargoTotalWeightVgm ?? 0);

    // 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.
    const capacityTons = this.capacityFor(primary.freightType, wagonDims);
    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,
        this.tareFor(primary.freightType, wagonDims),
      ),
      lengthMeters: bookingTrainLengthMeters(
        primary.freightType,
        sharedWagons,
        this.lengthsOf(wagonDims),
      ),
    };
  }

  /** Per-wagon tare for the wagon type this freight rides on. */
  private tareFor(freightType: string | null | undefined, wagonDims: WagonDims): number {
    return freightType === 'BULK'
      ? wagonDims.bulk.tareWeightTons
      : wagonDims.container.tareWeightTons;
  }

  private lengthsOf(wagonDims: WagonDims): { container: number; bulk: number } {
    return {
      container: wagonDims.container.lengthMeters,
      bulk: wagonDims.bulk.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 (wagonsPerUnit = 0.5). The old fallback
    // summed raw container QUANTITY, so 20×20ft counted as 20 wagons, not 10.
    const byLength = containerWagonsForLines(booking.bookingContainers ?? []);

    const capacityTons = this.capacityFor(booking.freightType, wagonDims);
    const cargoTons = Number(booking.cargoTotalWeightVgm ?? 0);
    const byWeight =
      cargoTons > 0 && capacityTons > 0 ? Math.ceil(cargoTons / capacityTons) : 0;

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

  private capacityFor(
    freightType: string | null | undefined,
    wagonDims: WagonDims,
  ): number {
    return freightType === "BULK"
      ? wagonDims.bulk.capacityTons
      : wagonDims.container.capacityTons;
  }

  /**
   * 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);
    return {
      wagons,
      weightTons: bookingGrossWeightTons(
        Number(booking.cargoTotalWeightVgm ?? 0),
        wagons,
        this.tareFor(booking.freightType, wagonDims),
      ),
      lengthMeters: bookingTrainLengthMeters(
        booking.freightType,
        wagons,
        this.lengthsOf(wagonDims),
      ),
    };
  }

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

  /**
   * Hard 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
   * these via {@link needFor}, whose weight axis is gross.
   */
  private async capacityLimits(
    locomotive: Locomotive,
    rules: TrainSchedulingGlobalRules | null,
  ): Promise<Capacity> {
    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,
      {
        maxTrainWeightTons: rules?.maxTrainWeightTons
          ? Number(rules.maxTrainWeightTons)
          : undefined,
        maxTrainLengthMeters: rules?.maxTrainLengthMeters
          ? Number(rules.maxTrainLengthMeters)
          : undefined,
      },
    );
    return {
      wagons: derived.maxWagonSlots,
      weightTons: derived.maxWeightTons,
      lengthMeters: derived.maxLengthMeters,
    };
  }

  /** Keep schedule.max_wagons aligned with locomotive physical limits. */
  private async syncScheduleMaxWagons(
    schedule: TrainSchedule,
    locomotive: Locomotive,
    rules: TrainSchedulingGlobalRules | null,
  ): Promise<void> {
    const limits = await this.capacityLimits(locomotive, rules);
    if ((schedule.maxWagons ?? 0) !== limits.wagons) {
      await this.dataSource
        .getRepository(TrainSchedule)
        .update(schedule.id, { maxWagons: limits.wagons });
      schedule.maxWagons = limits.wagons;
    }
  }

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

  /** Representative wagon per freight type: NW5 flat for containers, CW3 gondola for bulk. */
  private async loadWagonDims(): Promise<WagonDims> {
    const types = await this.dataSource.getRepository(WagonType).find({
      where: [{ code: "NW5" }, { code: "CW3" }],
    });
    const byCode = new Map(
      types.map((t) => [t.code, 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;
    return {
      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),
      },
    };
  }

  private async loadGlobalRules(): Promise<TrainSchedulingGlobalRules | null> {
    return this.dataSource
      .getRepository(TrainSchedulingGlobalRules)
      .findOne({ where: {} });
  }

  /**
   * 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.
   */
  private async remainingBudget(
    schedule: TrainSchedule,
    limits: Capacity,
    wagonDims: WagonDims,
  ): Promise<CorridorBudget> {
    const stops = await this.stopsForSchedule(schedule);
    const budget = new CorridorBudget(stops, limits);
    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;
  }

  /**
   * Wagon slots still boardable somewhere on the corridor (most-open edge).
   * ≤ 0 means no leg can take another booking — the train-wide FULL signal.
   */
  private async remainingWagons(schedule: TrainSchedule): Promise<number> {
    const wagonDims = await this.loadWagonDims();
    const budget = await this.remainingBudget(
      schedule,
      {
        wagons: schedule.maxWagons ?? 0,
        weightTons: Number.POSITIVE_INFINITY,
        lengthMeters: Number.POSITIVE_INFINITY,
      },
      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,
    );
  }

  /** No wagon slots left for allocated + reserved bookings. */
  async isScheduleFull(scheduleId: string): Promise<boolean> {
    const schedule =
      await this.trainSchedulesRepository.findByIdWithFullGraph(scheduleId);
    if (!schedule) return false;
    return (await this.remainingWagons(schedule)) <= 0;
  }

  /**
   * 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;
    if ((await this.remainingWagons(schedule)) <= 0) 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
    }
  }
}