import {
  BadRequestException,
  ConflictException,
  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 { Locomotive } from '../locomotives/entities/locomotive.entity';
import { formatRouteLabel } from '../routes/entities/route.entity';
import { TrainSchedule } from '../train-schedules/entities/train-schedule.entity';
import { TrainScheduleBooking } from '../train-schedules/entities/train-schedule-booking.entity';
import { TrainSchedulesRepository } from '../train-schedules/train-schedules.repository';
import { TrainScheduleBookingsRepository } from '../train-schedules/train-schedule-bookings.repository';
import { TrainSchedulingGlobalRules } from './entities/train-scheduling-global-rules.entity';
import { BookingNotifierService } from './booking-notifier.service';
import { TrainSchedulingService } from './train-scheduling.service';
import { eatDay, groupBookingsIntoBoardWindows } from './batch-window.util';
import { Freight, TrainScheduleStatus as TrainScheduleStatusEnum } from "@edr/types";
import { BillingService } from "../billing/billing.service";


import {
  DEFAULT_BULK_WAGON_LENGTH_METERS,
  DEFAULT_CONTAINER_WAGON_LENGTH_METERS,
  DEFAULT_WAGONS_PER_BOOKING,
} from "./booking-batch.constants";
import {
  bookingTrainLengthMeters,
  deriveTrainCapacityFromLocomotive,
  wagonTypeDimensionsFromEntity,
} from './train-capacity.util';
import { WagonType } from '../wagon-types/entities/wagon-type.entity';
import { ClearanceMilestoneService } from '../contracts/clearance-milestone.service';
import { BookingSplitService } from './booking-split.service';
import { MAX_TEU_SLOTS_PER_WAGON } from './wagon-plan.util';

/** A train's remaining capacity along the three physical limits the batch enforces. */
interface Capacity {
  wagons: number;
  weightTons: number;
  lengthMeters: number;
}

/** A day-level pool key: all trains on this route departing on this EAT day. */
interface RouteDayGroup {
  originYardId: string;
  destinationYardId: string;
  /** EAT calendar day, `yyyy-MM-dd`. */
  day: string;
}

type WagonLengths = { container: number; bulk: number };

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

export interface BatchBoardBooking {
  id: string;
  reference: string;
  company: string;
  isGovernment: boolean;
  wagons: number;
  weightTons: number;
  lengthMeters: number;
  paymentDeadline: string | null;
  state: BatchBoardBookingState;
}

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

export interface BatchBoardBookingDetail extends BatchBoardBooking {
  fullyExecutedAt: string | null;
  selectedForBatchAt: string | null;
  allocationStatus: BookingAllocationStatus;
  allocationIssue: string | null;
  /** 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;
  };
  counts: {
    allocated: number;
    selectedForBatch: number;
    ready: number;
    waiting: number;
    pendingContract: number;
    expired: number;
  };
  bookings: BatchBoardBooking[];
}

/**
 * Demand-batching engine: every 3h (EAT) it ranks each OPEN schedule's ready pool
 * by priority, greedily fills the train to capacity (skipping bookings that don't fit),
 * reserves a 1h pay window for commercial customers (government allocated unpaid,
 * preempting lower-priority commercial if needed), then settles each batch 1h later —
 * allocating those who paid and expiring those who didn't, topping up from the waiting list.
 * Capacity is bounded on three axes at once: wagon count (`schedule.maxWagons`), the
 * locomotive's max pull weight, and its max train length (also capped by global rules).
 */
@Injectable()
export class BookingBatchService implements OnModuleInit {
  private readonly logger = new Logger(BookingBatchService.name);

  constructor(
    @InjectDataSource() private readonly dataSource: DataSource,
    private readonly bookingsRepository: BookingsRepository,
    private readonly trainSchedulesRepository: TrainSchedulesRepository,
    private readonly trainScheduleBookingsRepository: TrainScheduleBookingsRepository,
    private readonly notifier: BookingNotifierService,
    private readonly scheduler: SchedulerRegistry,
    private readonly trainSchedulingService: TrainSchedulingService,
    private readonly billing: BillingService,

    @Optional() private readonly milestoneService?: ClearanceMilestoneService,
    @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> {
    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" },
      })
    ).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}`,
      );
    }

    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));
    const corridor = await this.trainSchedulesRepository.findAll({
      where: [
        {
          originStationId: booking.originYardId,
          destinationStationId: booking.destinationYardId,
          status: TrainScheduleStatusEnum.Draft,
        },
        {
          originStationId: booking.originYardId,
          destinationStationId: booking.destinationYardId,
          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 wagonLengths = await this.loadWagonLengths();
    const required = need ?? this.needFor(booking, wagonLengths);
    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.remainingCapacity(schedule, limits, wagonLengths);
      if (this.fits(required, budget)) return schedule.id;
    }
    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 wagonLengths = await this.loadWagonLengths();
    const need = this.combinedNeed(booking, partner, wagonLengths);
    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,
        route: true,
      },
      order: { scheduledDepartureDate: "ASC" },
    });

    const wagonLengths = await this.loadWagonLengths();
    const linkRepo = this.dataSource.getRepository(TrainScheduleBooking);

    const board: BatchBoardSchedule[] = [];
    for (const s of schedules) {
      if (s.status === "ARRIVED" || s.status === "CANCELLED") continue;
      // 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, wagonLengths);
        return {
          id: b.id,
          reference: b.reference ?? b.id.slice(0, 8),
          company: b.isGovernment
            ? (b.governmentInstitution ?? "Government")
            : (b.company?.name ?? "—"),
          isGovernment: Boolean(b.isGovernment),
          wagons: need.wagons,
          weightTons: need.weightTons,
          lengthMeters: need.lengthMeters,
          paymentDeadline: b.paymentDeadline
            ? b.paymentDeadline.toISOString()
            : null,
          state: this.boardState(b, linkedIds.has(b.id)),
        };
      });

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

  /** Schedule-level batch board with EAT 3h windows grouped by fullyExecutedAt. */
  async getBatchBoardDetail(
    scheduleId: string,
  ): Promise<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 wagonLengths = await this.loadWagonLengths();
    const linkRepo = this.dataSource.getRepository(TrainScheduleBooking);
    const links = await linkRepo.find({ where: { trainScheduleId: s.id } });
    const linkedIds = new Set(links.map((l) => l.bookingId));
    const bookings = await this.bookingsRepository.findAllBySchedule(s.id);

    let allocationPreview: Awaited<
      ReturnType<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, wagonLengths);
      const alloc = allocationByBooking.get(b.id);
      return {
        id: b.id,
        reference: b.reference ?? b.id.slice(0, 8),
        company: b.isGovernment
          ? (b.governmentInstitution ?? "Government")
          : (b.company?.name ?? "—"),
        isGovernment: Boolean(b.isGovernment),
        wagons: need.wagons,
        weightTons: need.weightTons,
        lengthMeters: need.lengthMeters,
        paymentDeadline: b.paymentDeadline
          ? b.paymentDeadline.toISOString()
          : null,
        state: this.boardState(b, linkedIds.has(b.id)),
        fullyExecutedAt: b.fullyExecutedAt
          ? b.fullyExecutedAt.toISOString()
          : null,
        selectedForBatchAt: b.selectedForBatchAt
          ? b.selectedForBatchAt.toISOString()
          : null,
        allocationStatus: alloc?.status ?? "NOT_ATTEMPTED",
        allocationIssue: alloc?.issue ?? null,
        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 from the global-rules
    // config (import: opens at windowOpenHour EAT importWindowLeadDays before
    // departure, lasts windowDurationHours, reopens per reopenDelayMinutes;
    // export: single FCFS lead window) — not a fixed clock grid.
    const windowCfg = await this.trainSchedulingService.getWindowConfig();
    const departureDate = s.scheduledDepartureDate ?? new Date();
    const windowBuckets = groupBookingsIntoBoardWindows(
      items,
      (item) => (item.fullyExecutedAt ? new Date(item.fullyExecutedAt) : null),
      s.direction ?? null,
      departureDate,
      windowCfg,
    );

    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),
      counts: {
        allocated: items.filter((i) => i.state === "ALLOCATED").length,
        selectedForBatch: items.filter((i) => i.state === "SELECTED_FOR_BATCH")
          .length,
        ready: items.filter((i) => i.state === "READY").length,
        waiting: items.filter((i) => i.state === "WAITING").length,
        pendingContract: items.filter((i) => i.state === "PENDING_CONTRACT")
          .length,
        expired: items.filter((i) => i.state === "EXPIRED").length,
      },
      windows,
      pendingContract: {
        key: "pending-contract",
        label: "Pending contract",
        date: "",
        dateLabel: "",
        start: "",
        end: "",
        counts: countFor(pendingBookings),
        bookings: pendingBookings,
      },
      allocationViolations: allocationPreview.violations,
    };
  }

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

  private computeBoardCapacity(
    items: Array<{
      state: BatchBoardBookingState;
      wagons: number;
      weightTons: number;
      lengthMeters: number;
    }>,
    loco: Locomotive | null,
  ): BatchBoardSchedule["capacity"] {
    const allocated = items.filter((i) => i.state === "ALLOCATED");
    const committed = items.filter(
      (i) => i.state === "ALLOCATED" || i.state === "SELECTED_FOR_BATCH",
    );
    return {
      allocatedWagons: allocated.reduce((sum, i) => sum + i.wagons, 0),
      allocatedLengthMeters:
        Math.round(
          allocated.reduce((sum, i) => sum + i.lengthMeters, 0) * 100,
        ) / 100,
      maxLengthMeters: loco ? Number(loco.maxTrainLengthMeters) : null,
      usedWeightTons:
        Math.round(committed.reduce((sum, i) => sum + i.weightTons, 0) * 100) /
        100,
      maxWeightTons: loco ? Number(loco.maxPullWeightTons) : null,
    };
  }

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

    return {
      scheduleId: s.id,
      trainNumber: s.trainNumber ?? null,
      routeName: s.route ? formatRouteLabel(s.route) : null,
      origin: s.originStation?.label ?? s.originStation?.code ?? null,
      destination:
        s.destinationStation?.label ?? s.destinationStation?.code ?? null,
      scheduleDate: s.scheduledDepartureDate
        ? s.scheduledDepartureDate.toISOString()
        : null,
      status: s.status,
      bookingWindowStatus: s.bookingWindowStatus,
      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),
      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. */
  async fillSchedule(scheduleId: string): Promise<void> {
    const schedule =
      await this.trainSchedulesRepository.findByIdWithFullGraph(scheduleId);
    if (!schedule || !this.isFillable(schedule)) return;
    const locomotive = schedule.trainSet?.locomotive;
    if (!schedule.trainSetId || !locomotive) {
      this.logger.warn(
        `Schedule ${scheduleId} has no locomotive/train set — skipped.`,
      );
      return;
    }

    const rules = await this.loadGlobalRules();
    const wagonLengths = await this.loadWagonLengths();
    const limits = await this.capacityLimits(locomotive, rules);
    await this.syncScheduleMaxWagons(schedule, locomotive, rules);
    let budget = await this.remainingCapacity(schedule, limits, wagonLengths);
    if (budget.wagons <= 0) {
      await this.setWindow(scheduleId, "FULL");
      return;
    }

    const pool = await this.bookingsRepository.findBatchPool(scheduleId);
    const units = this.groupConsolidatedPool(pool);
    let armed = false;

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

      if (!this.fits(need, budget)) {
        if (isGov) {
          budget = await this.preemptForGovernment(
            scheduleId,
            need,
            budget,
            wagonLengths,
          );
          if (!this.fits(need, budget)) continue; // still doesn't fit even after preempt
        } else {
          continue; // skip a unit that exceeds weight/length/wagons, try the next
        }
      }

      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;
      }
      budget = this.subtract(budget, need);
      if (budget.wagons <= 0) break; // no wagon slots left — nothing more can board
    }

    if (budget.wagons <= 0) await this.setWindow(scheduleId, "FULL");
    if (armed) this.armSettle(scheduleId);
    void this.triggerWagonAllocation(scheduleId);
  }

  /**
   * Distribute one (route, day) pool across ALL of that day's OPEN trains, by
   * priority, filling each train (earliest departure first) until it's full and
   * spilling overflow to the next. Government bookings that fit no train preempt
   * lower-priority commercial; bookings that fit no train at all stay pending and
   * trigger a staff `unplaced` warning. Returns the schedule ids that were touched
   * (or that had remaining pool work) so the caller can settle them per-schedule.
   */
  async fillRouteDay(
    originYardId: string,
    destinationYardId: string,
    day: string,
  ): Promise<string[]> {
    // 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 scheduleIds = corridor
      .filter(
        (s) =>
          s.scheduledDepartureDate != null &&
          eatDay(s.scheduledDepartureDate) === day &&
          this.isFillable(s),
      )
      .sort(
        (a, b) =>
          a.scheduledDepartureDate.getTime() - b.scheduledDepartureDate.getTime(),
      )
      .map((s) => s.id);

    if (scheduleIds.length === 0) return [];

    const rules = await this.loadGlobalRules();
    const wagonLengths = await this.loadWagonLengths();

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

    const pool = await this.bookingsRepository.findBatchPoolByRouteDay(
      originYardId,
      destinationYardId,
      day,
    );
    // 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);

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

      // First train (earliest departure) that fits this unit as-is.
      let target = trains.find((t) => this.fits(need, t.budget));

      if (!target && isGov) {
        // Government fits nowhere on its own — try to preempt commercial
        // on each train (earliest first) until one frees enough room.
        for (const t of trains) {
          t.budget = await this.preemptForGovernment(
            t.id,
            need,
            t.budget,
            wagonLengths,
          );
          if (this.fits(need, t.budget)) {
            target = t;
            break;
          }
        }
      }

      if (!target) {
        // A consolidated pair is placed whole or not at all — never split.
        if (!isPair) {
          // Fits no train whole. Import GENERAL-contract commercial bookings get a
          // partial-capacity offer on the train with the most free wagons.
          const partialTarget = [...trains]
            .filter((t) => t.budget.wagons >= 1)
            .sort((a, b) => b.budget.wagons - a.budget.wagons)[0];
          if (
            partialTarget &&
            !booking.isGovernment &&
            booking.tradeDirection === "IMPORT" &&
            booking.contractKind === "GENERAL" &&
            this.splitService
          ) {
            const offered = await this.tryPartialOffer(
              booking,
              partialTarget.id,
              partialTarget.budget,
              need,
            );
            if (offered) {
              partialTarget.budget = this.subtract(partialTarget.budget, offered);
              partialTarget.armed = true;
              continue;
            }
          }
        }
        // Stays in the pool, retried next batch/window cycle.
        this.notifier.unplaced(booking, day);
        if (partner) this.notifier.unplaced(partner, day);
        continue;
      }

      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;
      }
      target.budget = this.subtract(target.budget, need);
    }

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

    return trains.map((t) => t.id);
  }

  /**
   * 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 wagonLengths = await this.loadWagonLengths();
    const bulkCapacityTons = await this.loadBulkWagonCapacityTons();
    const sized = await this.splitService.sizeOffer(
      booking,
      budget.wagons,
      need.wagons,
      bulkCapacityTons,
    );
    if (!sized) return null;

    const offeredNeed: Capacity = {
      wagons: sized.offeredWagons,
      weightTons: sized.offeredWeightTons,
      lengthMeters: bookingTrainLengthMeters(booking.freightType, sized.offeredWagons, {
        container: wagonLengths.container,
        bulk: wagonLengths.bulk,
      }),
    };
    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;

    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. */
  async settleDueReservations(scheduleId: string): Promise<void> {
    const anySettled = await this.settleReserved(scheduleId, false);
    if (anySettled) await this.fillSchedule(scheduleId);
  }

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

  /** Allocate paid reservations, expire the rest, then top up. */
  async settleBatch(scheduleId: string): Promise<void> {
    this.removeTimeout(scheduleId);
    await this.settleReserved(scheduleId, true);
    await this.fillSchedule(scheduleId);
    void this.triggerWagonAllocation(scheduleId);
  }

  private triggerWagonAllocation(scheduleId: string): void {
    void this.trainSchedulingService
      .tryAutoWagonAllocation(scheduleId)
      .catch((err) =>
        this.logger.warn(
          `Auto wagon allocation failed for ${scheduleId}: ${(err as Error).message}`,
        ),
      );
  }

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

  /** Staff "mark paid" override → set PAID and allocate immediately (don't wait for settle). */
  async markPaid(bookingId: string): Promise<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",
      );
    }
    if (
      schedule.originStationId !== booking.originYardId ||
      schedule.destinationStationId !== booking.destinationYardId
    ) {
      throw new BadRequestException(
        "Target schedule is not on the booking route",
      );
    }

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

  /** Staff "expire" override → free a reservation now (booking becomes EXPIRED). */
  async expireReservation(bookingId: string): Promise<void> {
    const booking = await this.dataSource
      .getRepository(Booking)
      .findOne({ where: { id: bookingId } });
    if (!booking) throw new NotFoundException(`Booking ${bookingId} not found`);
    await this.expire(booking);
    if (booking.trainScheduleId)
      await this.fillSchedule(booking.trainScheduleId);
  }

  // ---- mutations ------------------------------------------------------------

  /**
   * Reserve capacity for a commercial booking 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> {
    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);
  }

  /** 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.notifier.secured(booking, reason);
    void this.triggerWagonAllocation(scheduleId);
    void this.markWagonAllocatedMilestone(booking.id);
  }

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

  /**
   * 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> {
    await this.bookingsRepository.update(booking.id, {
      trainScheduleId: null,
      status: "EXPIRED",
      schedulingStatus: "ELIGIBLE",
      paymentDeadline: null,
      selectedForBatchAt: null,
    } as never);
    booking.trainScheduleId = null;
    // 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);
  }

  /**
   * Free capacity for a government booking by displacing the lowest-priority commercial
   * bookings (reserved first, then allocated — including PAID). Displaced → EXPIRED + notified.
   */
  private async preemptForGovernment(
    scheduleId: string,
    need: Capacity,
    budget: Capacity,
    wagonLengths: WagonLengths,
  ): Promise<Capacity> {
    const reservedCommercial = (
      await this.bookingsRepository.findReservedForSchedule(scheduleId)
    ).filter((b) => !b.isGovernment);
    const allocatedCommercial =
      await this.bookingsRepository.findAllocatedCommercialForSchedule(
        scheduleId,
      );

    // lowest priority first; reserved are cheaper to free than allocated
    const candidates = [...reservedCommercial, ...allocatedCommercial].sort(
      (a, b) => (a.priorityScore ?? 0) - (b.priorityScore ?? 0),
    );

    let freed = budget;
    for (const victim of candidates) {
      if (this.fits(need, freed)) break;
      await this.dataSource.transaction(async (manager) => {
        await this.trainScheduleBookingsRepository.deleteByScheduleAndBooking(
          scheduleId,
          victim.id,
          manager,
        );
        await manager.getRepository(Booking).update(victim.id, {
          status: "EXPIRED",
          schedulingStatus: "ELIGIBLE",
          paymentDeadline: null,
          selectedForBatchAt: null,
        } as never);
        // Displaced → EXPIRED: close its open invoice too, so a dead booking
        // can't still be paid (mirrors `expire()`; enlisted in this txn).
        await this.billing.expirePayable(
          Freight.InvoiceSource.Booking,
          victim.id,
          "PREPAID",
          manager,
        );
      });
      this.notifier.displaced(victim);
      freed = this.add(freed, this.needFor(victim, wagonLengths));
    }
    return freed;
  }

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

  /**
   * 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,
    wagonLengths: WagonLengths,
  ): Capacity {
    const containers = (b: Booking): number =>
      (b.bookingContainers ?? []).reduce((sum, c) => sum + Number(c.quantity ?? 0), 0);
    const totalContainers = containers(primary) + containers(partner);
    const sharedWagons =
      totalContainers > 0
        ? Math.ceil(totalContainers / MAX_TEU_SLOTS_PER_WAGON)
        : this.wagonsFor(primary) + this.wagonsFor(partner);
    const weightTons =
      Number(primary.cargoTotalWeightVgm ?? 0) + Number(partner.cargoTotalWeightVgm ?? 0);
    return {
      wagons: sharedWagons,
      weightTons,
      lengthMeters: bookingTrainLengthMeters(primary.freightType, sharedWagons, {
        container: wagonLengths.container,
        bulk: wagonLengths.bulk,
      }),
    };
  }

  private wagonsFor(booking: Booking): number {
    if (booking.wagonsRequired && booking.wagonsRequired > 0) {
      return Math.ceil(booking.wagonsRequired);
    }
    const fromContainers = (booking.bookingContainers ?? []).reduce(
      (sum, c) => sum + Number(c.quantity ?? 0),
      0,
    );
    return Math.max(
      DEFAULT_WAGONS_PER_BOOKING,
      fromContainers || DEFAULT_WAGONS_PER_BOOKING,
    );
  }

  /** What one booking consumes along all three capacity axes. */
  private needFor(booking: Booking, wagonLengths: WagonLengths): Capacity {
    const wagons = this.wagonsFor(booking);
    return {
      wagons,
      weightTons: Number(booking.cargoTotalWeightVgm ?? 0),
      lengthMeters: bookingTrainLengthMeters(booking.freightType, wagons, {
        container: wagonLengths.container,
        bulk: wagonLengths.bulk,
      }),
    };
  }

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

  private subtract(budget: Capacity, need: Capacity): Capacity {
    return {
      wagons: budget.wagons - need.wagons,
      weightTons: budget.weightTons - need.weightTons,
      lengthMeters: budget.lengthMeters - need.lengthMeters,
    };
  }

  private add(budget: Capacity, freed: Capacity): Capacity {
    return {
      wagons: budget.wagons + freed.wagons,
      weightTons: budget.weightTons + freed.weightTons,
      lengthMeters: budget.lengthMeters + freed.lengthMeters,
    };
  }

  /** Locomotive + wagon-type-derived caps (weight, length, wagon slots — not a fixed 53). */
  private async capacityLimits(
    locomotive: Locomotive,
    rules: TrainSchedulingGlobalRules | null,
  ): Promise<Capacity> {
    const wagonTypes = await this.loadWagonTypeDimensions();
    const derived = deriveTrainCapacityFromLocomotive(
      {
        maxPullWeightTons: Number(locomotive.maxPullWeightTons),
        maxTrainLengthMeters: Number(locomotive.maxTrainLengthMeters),
      },
      wagonTypes,
      {
        maxTrainWeightTons: rules?.maxTrainWeightTons
          ? Number(rules.maxTrainWeightTons)
          : undefined,
        maxTrainLengthMeters: rules?.maxTrainLengthMeters
          ? Number(rules.maxTrainLengthMeters)
          : undefined,
      },
    );
    return {
      wagons: derived.maxWagonSlots,
      weightTons: derived.maxWeightTons,
      lengthMeters: derived.maxLengthMeters,
    };
  }

  /** Keep schedule.max_wagons aligned with locomotive physical limits. */
  private async syncScheduleMaxWagons(
    schedule: TrainSchedule,
    locomotive: Locomotive,
    rules: TrainSchedulingGlobalRules | null,
  ): Promise<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;
    }
  }

  private async loadWagonTypeDimensions(): Promise<
    Array<{ lengthMeters: number; capacityTons: number }>
  > {
    const types = await this.dataSource.getRepository(WagonType).find({
      where: [{ code: "NW5" }, { code: "CW3" }],
    });
    if (types.length) return types.map(wagonTypeDimensionsFromEntity);
    return [
      { lengthMeters: DEFAULT_CONTAINER_WAGON_LENGTH_METERS, capacityTons: 70 },
      { lengthMeters: DEFAULT_BULK_WAGON_LENGTH_METERS, capacityTons: 60 },
    ];
  }

  private async loadWagonLengths(): Promise<WagonLengths> {
    const types = await this.dataSource.getRepository(WagonType).find({
      where: [{ code: "NW5" }, { code: "CW3" }],
    });
    const byCode = new Map(
      types.map((t) => [t.code, wagonTypeDimensionsFromEntity(t)]),
    );
    return {
      container:
        byCode.get("NW5")?.lengthMeters ??
        DEFAULT_CONTAINER_WAGON_LENGTH_METERS,
      bulk: byCode.get("CW3")?.lengthMeters ?? DEFAULT_BULK_WAGON_LENGTH_METERS,
    };
  }

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

  /** Remaining capacity = hard caps minus what allocated + reserved bookings already use. */
  private async remainingCapacity(
    schedule: TrainSchedule,
    limits: Capacity,
    wagonLengths: WagonLengths,
  ): Promise<Capacity> {
    const allocated = (schedule.scheduleBookings ?? [])
      .map((sb) => sb.booking)
      .filter((b): b is Booking => Boolean(b));
    const reserved = await this.bookingsRepository.findReservedForSchedule(
      schedule.id,
    );
    const used = [...allocated, ...reserved].reduce<Capacity>(
      (acc, b) => this.add(acc, this.needFor(b, wagonLengths)),
      { wagons: 0, weightTons: 0, lengthMeters: 0 },
    );
    return this.subtract(limits, used);
  }

  /** maxWagons minus wagons already taken by allocated + reserved bookings. */
  private async remainingWagons(schedule: TrainSchedule): Promise<number> {
    const allocated = (schedule.scheduleBookings ?? [])
      .map((sb) => sb.booking)
      .filter((b): b is Booking => Boolean(b));
    const reserved = await this.bookingsRepository.findReservedForSchedule(
      schedule.id,
    );
    const used =
      allocated.reduce((s, b) => s + this.wagonsFor(b), 0) +
      reserved.reduce((s, b) => s + this.wagonsFor(b), 0);
    return (schedule.maxWagons ?? 0) - used;
  }

  async setWindow(
    scheduleId: string,
    status: "OPEN" | "FULL" | "CLOSED",
  ): Promise<void> {
    await this.dataSource
      .getRepository(TrainSchedule)
      .update(scheduleId, { bookingWindowStatus: status });
  }

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

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

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