import {
  BadRequestException,
  Injectable,
  Logger,
  NotFoundException,
  Optional,
} from '@nestjs/common';
import { EventEmitter2, OnEvent } from '@nestjs/event-emitter';
import { InjectDataSource } from '@nestjs/typeorm';
import { DataSource, EntityManager, In } from 'typeorm';
import { Freight } from '@edr/types';

import { YardFacilitiesService } from '../rule-engine/services/yard-facilities.service';
import { FacilityHandlingService } from './facility-handling.service';
import { Booking } from '../bookings/entities/booking.entity';
import { BookingContainer } from '../bookings/entities/booking-container.entity';
import { WagonAllocationContainerItem } from '../train-schedules/entities/wagon-allocation-container-item.entity';
import { ClearanceMilestoneService } from '../contracts/clearance-milestone.service';
import { Yard } from '../rule-engine/entities/yard.entity';
import { TrainSetWagon } from '../train-sets/entities/train-set-wagon.entity';
import {
  StationWorkLog,
  StationWorkPhaseLog,
  TrainSchedule,
} from '../train-schedules/entities/train-schedule.entity';
import { TrainScheduleBooking } from '../train-schedules/entities/train-schedule-booking.entity';
import { WagonBookingAllocation } from '../train-schedules/entities/wagon-booking-allocation.entity';
import { Wagon } from '../wagons/entities/wagon.entity';
import { WagonMovement } from '../wagons/entities/wagon-movement.entity';
import { TrainCheckpointEvent } from './entities/train-checkpoint-event.entity';
import { assertExportReceivedWithGrn, DIRECT_TO_TRAIN } from '../../common/export-received-gate';
import { NotificationsService } from '../notifications/notifications.service';
import { NotificationInboxService } from '../notification-inbox/notification-inbox.service';
import {
  notifyCarriageAcceptanceReady,
  notifyLoadManifest,
} from '../notifications/notify-company.util';
import { FREIGHT_PERMS } from '../../seed/freight-permissions.registry';

/**
 * Per-booking journey along a train's corridor — for EVERY trade direction.
 *
 * A booking rides only its own origin→destination leg, so "dispatched" and
 * "arrived" are per-booking facts confirmed by the yard operator, not train
 * facts: load at the booking's origin yard (PAID → IN_TRANSIT, loadedAt) and
 * unload at its destination yard (IN_TRANSIT → ARRIVED for import/export,
 * → COMPLETED for intercity), possibly long before the train's final arrival.
 * Both are gated on the train's latest recorded checkpoint being at that yard.
 * Unload also fires automatically: recording a checkpoint at a yard auto-
 * unloads every booking destined there (autoUnloadAtYard), so the manual
 * unload endpoint remains only a fallback.
 *
 * Unloading also settles the physical wagons: each wagon that alights with the
 * booking is released at that yard and the move is written to the
 * wagon_movements ledger.
 */
@Injectable()
export class BookingJourneyService {
  private readonly logger = new Logger(BookingJourneyService.name);

  constructor(
    @InjectDataSource() private readonly dataSource: DataSource,
    private readonly yardFacilities: YardFacilitiesService,
    private readonly facilityHandling: FacilityHandlingService,
    private readonly events: EventEmitter2,
    private readonly notifications: NotificationsService,
    private readonly inbox: NotificationInboxService,
    @Optional() private readonly milestoneService?: ClearanceMilestoneService,
  ) {}

  /** Statuses from which a booking may be loaded (gov bookings don't prepay). */
  private canLoad(booking: Booking): boolean {
    if (booking.status === 'PAID') return true;
    return booking.isGovernment && booking.status === 'APPROVED';
  }

  async loadBooking(scheduleId: string, bookingId: string, userId?: string | null) {
    const { schedule, booking } = await this.getScheduleBooking(scheduleId, bookingId);
    await this.assertBookingLoadable(schedule, booking);
    return this.completeLoad(schedule, booking, userId ?? null);
  }

  /**
   * Confirm ONE wagon of the booking loaded (per-wagon loading). The booking
   * stays PAID while wagons remain; loading the last remaining wagon runs the
   * whole-booking completion (IN_TRANSIT, warehouse inventory, GRN,
   * milestones) exactly as the one-shot load does. Wagons that will NOT ride
   * must be cancelled via the at-loading cancellation before the booking can
   * complete (and before the train may dispatch).
   */
  async loadWagon(
    scheduleId: string,
    bookingId: string,
    allocationId: string,
    userId?: string | null,
  ) {
    const { schedule, booking } = await this.getScheduleBooking(scheduleId, bookingId);
    await this.assertBookingLoadable(schedule, booking);
    const allocations = await this.allocationsForBooking(
      this.dataSource.manager,
      scheduleId,
      bookingId,
    );
    if (!allocations.length) {
      throw new BadRequestException(
        'This booking has no wagon allocations on the schedule — use the whole-booking load.',
      );
    }
    const target = allocations.find((a) => a.id === allocationId);
    if (!target) {
      throw new NotFoundException('Wagon allocation not found on this booking/schedule');
    }
    if (target.status === 'LOADED' || target.status === 'DEPARTED') {
      throw new BadRequestException('This wagon is already loaded');
    }

    const now = new Date();
    await this.dataSource.transaction(async (manager) => {
      await manager.getRepository(WagonBookingAllocation).update(target.id, {
        status: 'LOADED',
        loadedAt: now,
        loadedByUserId: userId ?? null,
      });
      if (!booking.loadingStartedAt) {
        await manager
          .getRepository(Booking)
          .update(bookingId, { loadingStartedAt: now } as never);
      }
      // PARTIAL keeps the dispatch-readiness badge honest; completion below
      // flips it to LOADED.
      await manager
        .getRepository(TrainScheduleBooking)
        .update({ trainScheduleId: scheduleId, bookingId }, { loadingStatus: 'PARTIAL' });
    });

    const remaining = allocations.filter(
      (a) => a.id !== target.id && a.status !== 'LOADED' && a.status !== 'DEPARTED',
    ).length;
    if (remaining === 0) {
      const done = await this.completeLoad(schedule, booking, userId ?? null);
      return {
        ...done,
        allocationId,
        loadedWagons: allocations.length,
        totalWagons: allocations.length,
        completed: true,
      };
    }
    return {
      bookingId,
      allocationId,
      status: booking.status,
      loadedWagons: allocations.length - remaining,
      totalWagons: allocations.length,
      completed: false,
    };
  }

  /**
   * The at-loading cancel shrank the booking to its loaded wagons — if every
   * wagon left on it is LOADED, the load is complete: run the whole-booking
   * completion. Fired by BookingWagonCancellationService.cancelRemainingAtLoading.
   */
  @OnEvent('booking.wagonsCancelledAtLoading')
  async onWagonsCancelledAtLoading(payload: {
    bookingId: string;
    scheduleId: string;
    userId?: string | null;
  }): Promise<void> {
    try {
      const allocations = await this.allocationsForBooking(
        this.dataSource.manager,
        payload.scheduleId,
        payload.bookingId,
      );
      const loaded = allocations.filter(
        (a) => a.status === 'LOADED' || a.status === 'DEPARTED',
      ).length;
      if (!allocations.length || loaded < allocations.length) return;
      await this.loadBooking(payload.scheduleId, payload.bookingId, payload.userId);
    } catch (err) {
      this.logger.error(
        `Post-cancel load completion failed for booking ${payload.bookingId}: ${err instanceof Error ? err.message : String(err)}`,
      );
    }
  }

  /** The pre-load gates shared by whole-booking and per-wagon loading. */
  private async assertBookingLoadable(schedule: TrainSchedule, booking: Booking): Promise<void> {
    if (booking.loadedAt || booking.status === 'IN_TRANSIT') {
      throw new BadRequestException('Booking is already loaded');
    }
    if (!this.canLoad(booking)) {
      throw new BadRequestException(
        `Booking must be paid before loading (currently ${booking.status})`,
      );
    }
    await this.assertTrainAtYard(schedule, booking.originYardId, 'origin');
    this.assertStationWorkStarted(schedule, booking.originYardId, 'loading');
    await this.assertYardCanHandleCargo(booking, booking.originYardId, 'origin');
    // Export cargo must be in the warehouse with a GRN before it can be loaded,
    // however it arrived and whatever it is allocated to.
    await assertExportReceivedWithGrn(this.dataSource, booking);
  }

  /** The whole-booking load side effects — gates already passed. */
  private async completeLoad(
    schedule: TrainSchedule,
    booking: Booking,
    userId: string | null,
  ) {
    const scheduleId = schedule.id;
    const bookingId = booking.id;
    // Direct truck-to-train cargo never sees the warehouse, so loading IS its
    // handover moment — the carriage acceptance sheet must go out to the
    // customer right here, not on a receive event that will never fire.
    if (
      booking.tradeDirection === 'EXPORT' &&
      booking.exportHandoverMode === DIRECT_TO_TRAIN
    ) {
      await notifyCarriageAcceptanceReady(
        this.dataSource,
        this.notifications,
        this.inbox,
        booking.id,
        this.logger,
      );
    }

    const now = new Date();
    await this.dataSource.transaction(async (manager) => {
      await manager.getRepository(Booking).update(bookingId, {
        status: 'IN_TRANSIT',
        loadedAt: now,
        loadedByUserId: userId ?? null,
      } as never);
      // Intercity cargo rides the wagons freed by earlier unloads along the
      // corridor — place it before the status flip so it boards with a wagon.
      if (booking.tradeDirection === 'DOMESTIC') {
        await this.autoPlaceOnFreedWagons(manager, schedule, booking);
      }
      await this.setAllocationStatuses(manager, scheduleId, bookingId, 'LOADED');
      // Keep the schedule↔booking link's tracking flag in sync — the dispatch
      // readiness warnings and workspace badges read loading_status, not loadedAt.
      await manager
        .getRepository(TrainScheduleBooking)
        .update({ trainScheduleId: scheduleId, bookingId }, { loadingStatus: 'LOADED' });
      // Warehouse cargo may be loaded either from the warehouse Load-to-Train
      // queue or from the schedule itself. Loading here must move its inventory
      // too, otherwise the goods read as still sitting in the shed while the
      // train leaves with them. No-ops for direct truck-to-train (no inventory).
      // ponytail: no WarehouseLoading record on this path — those are only read
      // back as per-inventory loading history, never billed. Create them here if
      // that history ever has to be complete.
      await manager.query(
        `UPDATE freight.warehouse_inventory
            SET status = 'LOADED', loaded_at = COALESCE(loaded_at, $2), updated_at = NOW()
          WHERE booking_id = $1 AND deleted_at IS NULL
            AND status NOT IN ('LOADED', 'DISPATCHED')`,
        [bookingId, now],
      );
      // The facility handed the cargo over — raise its GRN. No-ops for yards
      // without a facility (import/export terminals), which keep their own flow.
      await this.facilityHandling.recordHandling(manager, {
        booking,
        yardId: booking.originYardId,
        trainScheduleId: scheduleId,
        eventType: 'LOAD',
        performedBy: userId ?? null,
        occurredAt: now,
      });
    });

    // What actually boarded, and what did not. A booking is routinely loaded in
    // parts; the customer is told both halves, and the warehouse desk is told
    // about the leftovers so somebody owns placing them. After the transaction:
    // the lists are read back from the allocation statuses it just wrote.
    void notifyLoadManifest(
      this.dataSource,
      this.notifications,
      this.inbox,
      bookingId,
      scheduleId,
      FREIGHT_PERMS.warehouseInventory.getNotification,
      this.logger,
    );

    // Customer tracking: cargo is on the train — loading milestones plus the
    // direction's "departed" handoff. Doc-trigger path no-ops non-customs
    // bookings (intercity) and already-completed codes.
    void this.completeMilestones(booking, [
      'CARGO_ARRIVED',
      'READY_FOR_LOADING',
      'LOADED',
      ...(booking.tradeDirection === 'IMPORT'
        ? ['DEPARTED_FROM_DJIBOUTI']
        : booking.tradeDirection === 'EXPORT'
          ? ['DEPARTED_TO_DJIBOUTI']
          : []),
    ]);

    return { bookingId, status: 'IN_TRANSIT' as const, loadedAt: now.toISOString() };
  }

  async unloadBooking(scheduleId: string, bookingId: string, userId?: string | null) {
    const { schedule, booking } = await this.getScheduleBooking(scheduleId, bookingId);
    await this.assertBookingUnloadable(schedule, booking);
    return this.completeUnload(schedule, booking, userId ?? null);
  }

  /**
   * Confirm ONE wagon of the booking unloaded (per-wagon unloading). Tracking
   * only while wagons remain on the train — the booking stays IN_TRANSIT;
   * unloading the last wagon runs the whole-booking completion (ARRIVED/
   * COMPLETED, wagon settlement, events) exactly as the one-shot unload does.
   */
  async unloadWagon(
    scheduleId: string,
    bookingId: string,
    allocationId: string,
    userId?: string | null,
  ) {
    const { schedule, booking } = await this.getScheduleBooking(scheduleId, bookingId);
    await this.assertBookingUnloadable(schedule, booking);
    const allocations = await this.allocationsForBooking(
      this.dataSource.manager,
      scheduleId,
      bookingId,
    );
    if (!allocations.length) {
      throw new BadRequestException(
        'This booking has no wagon allocations on the schedule — use the whole-booking unload.',
      );
    }
    const target = allocations.find((a) => a.id === allocationId);
    if (!target) {
      throw new NotFoundException('Wagon allocation not found on this booking/schedule');
    }
    if (target.status === 'DEPARTED') {
      throw new BadRequestException('This wagon is already unloaded');
    }

    const now = new Date();
    await this.dataSource.getRepository(WagonBookingAllocation).update(target.id, {
      status: 'DEPARTED',
      unloadedAt: now,
      unloadedByUserId: userId ?? null,
    });

    const remaining = allocations.filter(
      (a) => a.id !== target.id && a.status !== 'DEPARTED',
    ).length;
    if (remaining === 0) {
      const done = await this.completeUnload(schedule, booking, userId ?? null);
      return {
        ...done,
        allocationId,
        unloadedWagons: allocations.length,
        totalWagons: allocations.length,
        completed: true,
      };
    }
    return {
      bookingId,
      allocationId,
      status: booking.status,
      unloadedWagons: allocations.length - remaining,
      totalWagons: allocations.length,
      completed: false,
    };
  }

  /** The pre-unload gates shared by whole-booking and per-wagon unloading. */
  private async assertBookingUnloadable(
    schedule: TrainSchedule,
    booking: Booking,
  ): Promise<void> {
    if (booking.status !== 'IN_TRANSIT') {
      throw new BadRequestException(
        `Booking must be loaded/in transit before unloading (currently ${booking.status})`,
      );
    }
    await this.assertTrainAtYard(schedule, booking.destinationYardId, 'destination');
    this.assertStationWorkStarted(schedule, booking.destinationYardId, 'unloading');
    await this.assertYardCanHandleCargo(booking, booking.destinationYardId, 'destination');
  }

  /** The whole-booking unload side effects — gates already passed. */
  private async completeUnload(
    schedule: TrainSchedule,
    booking: Booking,
    userId: string | null,
  ) {
    const scheduleId = schedule.id;
    const bookingId = booking.id;
    // Intercity has no clearance/delivery tail — unloading completes it. Import/
    // export continue into clearance, keyed on the booking's own arrival.
    const nextStatus = booking.tradeDirection === 'DOMESTIC' ? 'COMPLETED' : 'ARRIVED';
    const now = new Date();
    await this.dataSource.transaction(async (manager) => {
      await manager.getRepository(Booking).update(bookingId, {
        status: nextStatus,
        arrivedAt: now,
        arrivedByUserId: userId ?? null,
      } as never);
      await this.setAllocationStatuses(manager, scheduleId, bookingId, 'DEPARTED');
      await this.settleWagonsOnUnload(manager, schedule, booking, now, userId ?? null);
      // The facility took the cargo off the train — raise its GRN. Where the
      // facility also stores cargo (Indode), the event links the storage record
      // that storage/demurrage accrue against.
      await this.facilityHandling.recordHandling(manager, {
        booking,
        yardId: booking.destinationYardId,
        trainScheduleId: scheduleId,
        eventType: 'UNLOAD',
        performedBy: userId ?? null,
        occurredAt: now,
      });
    });

    // Intercity ends here — a ONE_TIME contract closes on its shipment being
    // delivered (import/export emit this from booking-transition.complete).
    if (nextStatus === 'COMPLETED') {
      this.events.emit('booking.completed', { bookingId });
    }

    // The cargo is physically off the train at its own yard — mid-corridor or
    // final. WarehouseInventoryService picks this up to create the warehouse
    // record (import/intercity only; export already has one from receive).
    this.events.emit('booking.unloadedAtYard', {
      bookingId,
      tradeDirection: booking.tradeDirection,
    });

    // Customer tracking: THIS booking arrived (train may still be rolling).
    void this.completeMilestones(booking, [
      ...(booking.tradeDirection === 'IMPORT'
        ? ['ARRIVED_ETHIOPIA']
        : booking.tradeDirection === 'EXPORT'
          ? ['ARRIVED_AT_DJIBOUTI']
          : []),
    ]);

    return { bookingId, status: nextStatus, arrivedAt: now.toISOString() };
  }

  /**
   * Per-yard operator worklist for a schedule: which bookings board / alight at
   * each stop, with their journey state, so the yard operator at Dire sees
   * exactly what to load and unload when the train is there.
   */
  async listYardWork(scheduleId: string) {
    const schedule = await this.getSchedule(scheduleId);
    const bookings = await this.dataSource
      .getRepository(Booking)
      .createQueryBuilder('booking')
      .leftJoinAndSelect('booking.company', 'company')
      .leftJoinAndSelect('booking.originYard', 'originYard')
      .leftJoinAndSelect('booking.destinationYard', 'destinationYard')
      .innerJoin(
        TrainScheduleBooking,
        'tsb',
        'tsb.booking_id = booking.id AND tsb.train_schedule_id = :scheduleId AND tsb.deleted_at IS NULL',
        { scheduleId },
      )
      .getMany();

    const latest = await this.latestCheckpoint(scheduleId);
    const yardIds = [
      ...new Set(
        bookings.flatMap((b) => [b.originYardId, b.destinationYardId]).filter(Boolean),
      ),
    ];
    const yards = yardIds.length
      ? await this.dataSource.getRepository(Yard).find({ where: { id: In(yardIds) } })
      : [];
    const yardById = new Map(yards.map((y) => [y.id, y]));
    const yardLabel = (id: string) =>
      yardById.get(id)?.label ?? yardById.get(id)?.code ?? id;

    const mapBooking = (b: Booking) => ({
      id: b.id,
      reference: b.reference,
      status: b.status,
      tradeDirection: b.tradeDirection,
      isGovernment: b.isGovernment,
      customer: b.company?.name ?? 'Unknown customer',
      originYardId: b.originYardId,
      destinationYardId: b.destinationYardId,
      origin: yardLabel(b.originYardId),
      destination: yardLabel(b.destinationYardId),
      loadedAt: b.loadedAt?.toISOString() ?? null,
      arrivedAt: b.arrivedAt?.toISOString() ?? null,
      canLoad: !b.loadedAt && this.canLoad(b),
      canUnload: b.status === 'IN_TRANSIT',
    });

    const byYard = new Map<
      string,
      { yardId: string; yard: string; toLoad: ReturnType<typeof mapBooking>[]; toUnload: ReturnType<typeof mapBooking>[] }
    >();
    const bucket = (yardId: string) => {
      let entry = byYard.get(yardId);
      if (!entry) {
        entry = { yardId, yard: yardLabel(yardId), toLoad: [], toUnload: [] };
        byYard.set(yardId, entry);
      }
      return entry;
    };
    for (const b of bookings) {
      bucket(b.originYardId).toLoad.push(mapBooking(b));
      bucket(b.destinationYardId).toUnload.push(mapBooking(b));
    }

    return {
      scheduleId,
      scheduleStatus: schedule.status,
      // No checkpoint yet ⇒ the train is still at its origin, even just after
      // dispatch — assertTrainAtYard allows origin loading in that state, so
      // the UI position must agree or origin Load buttons grey out wrongly.
      trainAtYardId: latest?.yardId ?? schedule.originStationId,
      // Per-yard loading/unloading time windows — the UI derives its
      // start/end buttons and the load/unload gating from these.
      stationWorkLogs: schedule.stationWorkLogs ?? {},
      yards: [...byYard.values()],
    };
  }

  /**
   * Auto-unload on checkpoint: every IN_TRANSIT booking on this schedule whose
   * destination is the yard the train just reached alights automatically, so
   * the customer's booking flips to ARRIVED (COMPLETED for intercity) the
   * moment the train is recorded at their yard — no separate operator unload.
   * Runs through the same per-booking unload path (wagon settle + ledger +
   * milestones); one booking's failure is logged and never blocks the
   * checkpoint or the other bookings. Returns the unloaded booking ids.
   */
  async autoUnloadAtYard(
    scheduleId: string,
    yardId: string,
    userId?: string | null,
  ): Promise<string[]> {
    const bookings = await this.dataSource
      .getRepository(Booking)
      .createQueryBuilder('booking')
      // Entity-class join: a raw 'freight.table' string is parsed by TypeORM as
      // an alias.property path ("freight" alias was not found) — runtime 500.
      .innerJoin(
        TrainScheduleBooking,
        'tsb',
        'tsb.booking_id = booking.id AND tsb.train_schedule_id = :scheduleId AND tsb.deleted_at IS NULL',
        { scheduleId },
      )
      .where('booking.destination_yard_id = :yardId', { yardId })
      .andWhere(`booking.status = 'IN_TRANSIT'`)
      .getMany();

    const unloaded: string[] = [];
    for (const booking of bookings) {
      try {
        await this.unloadBooking(scheduleId, booking.id, userId);
        unloaded.push(booking.id);
      } catch (err) {
        this.logger.warn(
          `Auto-unload failed for booking ${booking.id} at yard ${yardId}: ${(err as Error).message}`,
        );
      }
    }
    return unloaded;
  }

  /**
   * Bulk fallback at the train's FINAL arrival: any booking destined for the
   * final yard that operators didn't unload individually gets its per-booking
   * arrival stamped now, so nothing stays stuck. Mid-corridor bookings are NOT
   * touched — their arrival is their own unload. Returns the affected ids.
   */
  async autoArriveAtFinalYard(
    manager: EntityManager,
    schedule: TrainSchedule,
    now: Date,
  ): Promise<string[]> {
    const rows: Array<{ id: string; trade_direction: string }> = await manager.query(
      `UPDATE freight.bookings b
          SET status = CASE WHEN b.trade_direction = 'DOMESTIC' THEN 'COMPLETED' ELSE 'ARRIVED' END,
              scheduling_status = 'DISPATCHED',
              arrived_at = COALESCE(b.arrived_at, $3),
              loaded_at = COALESCE(b.loaded_at, b.created_at)
         FROM freight.train_schedule_bookings tsb
        WHERE tsb.booking_id = b.id
          AND tsb.train_schedule_id = $1
          AND tsb.deleted_at IS NULL
          AND b.deleted_at IS NULL
          AND b.destination_yard_id = $2
          AND b.status NOT IN ('DRAFT', 'REJECTED', 'CANCELLED', 'COMPLETED', 'ARRIVED', 'DELIVERED')
        RETURNING b.id, b.trade_direction`,
      [schedule.id, schedule.destinationStationId, now],
    );
    if (rows.length === 0) return [];

    // The facility took the cargo off the train at the final yard — raise its
    // GRN, same as the per-booking unloadBooking() path does. Only when that
    // yard also has a warehouse (or has no facility at all, e.g. Kality) does
    // WarehouseInventoryService additionally get to allocate a warehouse/yard/
    // zone row: a pure facility yard (Dire Dawa, Modjo, Sebeta, Adama) is
    // fully represented by the facility event alone — there is nothing there
    // for warehouse_inventory's NOT NULL warehouse/yard/zone to point at.
    const facility = await this.yardFacilities.facilityForYard(schedule.destinationStationId);
    const bookings = await manager
      .getRepository(Booking)
      .find({ where: { id: In(rows.map((r) => r.id)) }, relations: ['company'] });
    const bookingById = new Map(bookings.map((b) => [b.id, b]));

    for (const row of rows) {
      // Intercity rows just completed — let a ONE_TIME contract close on delivery.
      if (row.trade_direction === 'DOMESTIC') {
        this.events.emit('booking.completed', { bookingId: row.id });
      }

      const booking = bookingById.get(row.id);
      if (booking) {
        await this.facilityHandling.recordHandling(manager, {
          booking,
          yardId: schedule.destinationStationId,
          trainScheduleId: schedule.id,
          eventType: 'UNLOAD',
          occurredAt: now,
        });
      }

      // Same event the per-booking unloadBooking() path emits — WarehouseInventoryService
      // listens for this to auto-create the warehouse_inventory row (import/intercity only,
      // it filters EXPORT itself). The bulk SQL update above skipped this entirely, so
      // bookings caught by this fallback never left "awaiting unload".
      if (row.trade_direction !== 'EXPORT' && (!facility?.hasFacility || facility.hasWarehouse)) {
        this.events.emit('booking.unloadedAtYard', {
          bookingId: row.id,
          tradeDirection: row.trade_direction,
        });
      }
    }
    return rows.map((r) => r.id);
  }

  /**
   * Record a station's loading/unloading time-window click (or edit it — an
   * explicit `at` on an already-set edge overwrites the timestamp under the
   * same permission that set it). Rules: end needs start, start ≤ end, no
   * future times. Stored as ISO strings in train_schedules.station_work_logs.
   * ponytail: read-modify-write on the jsonb — two operators clicking the same
   * schedule in the same instant can clobber one edge; move to jsonb_set if
   * that ever bites.
   */
  async recordStationWork(
    scheduleId: string,
    yardId: string,
    phase: 'loading' | 'unloading',
    edge: 'start' | 'end',
    at?: string,
    userId?: string | null,
  ) {
    const schedule = await this.getSchedule(scheduleId);
    const when = at ? new Date(at) : new Date();
    if (Number.isNaN(when.getTime())) {
      throw new BadRequestException('Invalid timestamp');
    }
    if (when.getTime() > Date.now() + 60_000) {
      throw new BadRequestException(`${phase} ${edge} time cannot be in the future`);
    }

    const logs: Record<string, StationWorkLog> = schedule.stationWorkLogs ?? {};
    const entry: StationWorkLog = logs[yardId] ?? {};
    const ph: StationWorkPhaseLog = entry[phase] ?? {};

    if (edge === 'end') {
      if (!ph.startedAt) {
        throw new BadRequestException(`Start ${phase} at this station first`);
      }
      if (when.getTime() < new Date(ph.startedAt).getTime()) {
        throw new BadRequestException(`${phase} end cannot be before its start`);
      }
      ph.endedAt = when.toISOString();
      ph.endedByUserId = userId ?? null;
    } else {
      if (ph.endedAt && when.getTime() > new Date(ph.endedAt).getTime()) {
        throw new BadRequestException(`${phase} start cannot be after its end`);
      }
      ph.startedAt = when.toISOString();
      ph.startedByUserId = userId ?? null;
    }

    entry[phase] = ph;
    logs[yardId] = entry;
    await this.dataSource
      .getRepository(TrainSchedule)
      .update(scheduleId, { stationWorkLogs: logs });

    return { scheduleId, yardId, phase, ...ph };
  }

  // ---- helpers ---------------------------------------------------------------

  private async getSchedule(scheduleId: string): Promise<TrainSchedule> {
    const schedule = await this.dataSource
      .getRepository(TrainSchedule)
      .findOne({ where: { id: scheduleId } });
    if (!schedule) {
      throw new NotFoundException(`Train schedule ${scheduleId} not found`);
    }
    return schedule;
  }

  private async getScheduleBooking(scheduleId: string, bookingId: string) {
    const schedule = await this.getSchedule(scheduleId);
    const booking = await this.dataSource
      .getRepository(Booking)
      .findOne({ where: { id: bookingId } });
    if (!booking) throw new NotFoundException(`Booking ${bookingId} not found`);
    if (booking.trainScheduleId !== scheduleId) {
      // The schedule↔booking LINK is the same authority the workspace list
      // (listYardWork) renders from — some flows (export train pick) create it
      // with wagon allocations before bookings.train_schedule_id is stamped.
      // Trusting only the column made those rows show a Load button that
      // always 400'd.
      const linked = await this.dataSource.getRepository(TrainScheduleBooking).findOne({
        where: { trainScheduleId: scheduleId, bookingId },
      });
      if (!linked) {
        throw new BadRequestException('Booking is not assigned to this schedule');
      }
    }
    return { schedule, booking };
  }

  private async latestCheckpoint(scheduleId: string): Promise<TrainCheckpointEvent | null> {
    return this.dataSource.getRepository(TrainCheckpointEvent).findOne({
      where: { trainScheduleId: scheduleId },
      order: { occurredAt: 'DESC', createdAt: 'DESC' },
    });
  }

  /**
   * INTERCITY ONLY. Intercity cargo rides a passing train and is handled at the
   * booking's own yards, so those yards need the equipment to do it — a train
   * stopping somewhere is not the same as somewhere being able to load it.
   *
   * Import/export are untouched: their cargo is handled at the route's terminal
   * ports, not at an arbitrary mid-corridor yard, and gating them here would
   * block existing traffic.
   *
   * Lives here rather than in the controller so the checkpoint-driven
   * autoUnloadAtYard path cannot route around it.
   */
  private async assertYardCanHandleCargo(
    booking: Booking,
    yardId: string,
    side: 'origin' | 'destination',
  ): Promise<void> {
    if (booking.tradeDirection !== 'DOMESTIC') return;
    const facility = await this.yardFacilities.facilityForYard(yardId);
    const where = side === 'origin' ? 'loaded at its origin' : 'unloaded at its destination';

    if (!facility?.hasFacility) {
      throw new BadRequestException(
        `${facility?.yardLabel ?? 'This yard'} has no load/unload facility — an intercity booking cannot be ${where} here.`,
      );
    }
    // A facility only handles what its equipment can lift: containers need a
    // reach stacker/gantry, bulk does not.
    if (!this.yardFacilities.canHandleFreight(facility, booking.freightType)) {
      throw new BadRequestException(
        `${facility.yardLabel ?? 'This yard'} does not handle ${String(booking.freightType).toLowerCase()} cargo — ` +
          `an intercity booking cannot be ${where} here.`,
      );
    }
  }

  /**
   * Loading/unloading a booking is only allowed inside a started work window
   * at that yard — the operator must click "Start loading"/"Start unloading"
   * (recordStationWork) before touching cargo. The window's END is not checked:
   * a straggler booking can still be confirmed after the end click, and the
   * operator can push the end time later (it's editable) if that matters.
   * Lives here (not the controller) so the checkpoint-driven autoUnloadAtYard
   * path is gated too — the user wants unloading fully manual.
   */
  private assertStationWorkStarted(
    schedule: TrainSchedule,
    yardId: string,
    phase: 'loading' | 'unloading',
  ): void {
    if (!schedule.stationWorkLogs?.[yardId]?.[phase]?.startedAt) {
      throw new BadRequestException(
        `Start ${phase} at this station first — the ${phase} time window has not been started`,
      );
    }
  }

  /**
   * The train is "at" a yard when the latest recorded checkpoint is that yard,
   * or — for a booking boarding at the train's own origin — when the train has
   * not recorded any checkpoint yet (still sitting at its origin).
   */
  private async assertTrainAtYard(
    schedule: TrainSchedule,
    yardId: string,
    side: 'origin' | 'destination',
  ): Promise<void> {
    const latest = await this.latestCheckpoint(schedule.id);
    if (!latest) {
      if (side === 'origin' && schedule.originStationId === yardId) return;
      throw new BadRequestException(
        'Train has not reached this yard yet — record its checkpoint first',
      );
    }
    if (latest.yardId !== yardId) {
      throw new BadRequestException(
        `Train's last recorded position is not at the booking's ${side} yard`,
      );
    }
  }

  /**
   * INTERCITY ONLY. Intercity cargo does not get its own wagons — it rides the
   * slots freed by cargo already unloaded along the corridor (e.g. import
   * containers uncoupled at Dire Dawa). Staff pinning is a pre-dispatch tool,
   * so a DOMESTIC booking loaded mid-corridor is auto-placed here: greedy over
   * on-train slots (not DEPARTED) with no active cargo (every allocation
   * DEPARTED, or none), in consist order, by capacity. Container numbers are
   * copied onto the first allocation so the marshalling document and its
   * 40ft/20ft tally stay truthful. When nothing is free the load proceeds
   * unallocated — the marshalling document then lists the booking as on board
   * without a recorded wagon.
   * ponytail: remainder over free capacity is dumped on the last used slot
   * (paper overload beats missing cargo); upgrade path is a capacity guard in
   * the intercity accept step.
   */
  private async autoPlaceOnFreedWagons(
    manager: EntityManager,
    schedule: TrainSchedule,
    booking: Booking,
  ): Promise<void> {
    const existing = await this.allocationsForBooking(manager, schedule.id, booking.id);
    if (existing.length) return;

    const slots = await manager
      .getRepository(TrainSetWagon)
      .createQueryBuilder('slot')
      .leftJoinAndSelect('slot.allocations', 'alloc')
      .innerJoin(
        TrainSchedule,
        'schedule',
        'schedule.train_set_id = slot.train_set_id AND schedule.id = :scheduleId',
        { scheduleId: schedule.id },
      )
      .where(`slot.status != 'DEPARTED'`)
      .orderBy('slot.sequence_no', 'ASC')
      .getMany();
    const freed = slots.filter((slot) =>
      (slot.allocations ?? []).every((a) => a.status === 'DEPARTED'),
    );
    if (!freed.length) {
      this.logger.warn(
        `No freed wagon for intercity booking ${booking.reference} on schedule ${schedule.id} — loading without wagon allocation`,
      );
      return;
    }

    let remaining = Number(booking.cargoTotalWeightVgm) || 0;
    const allocRepo = manager.getRepository(WagonBookingAllocation);
    const created: WagonBookingAllocation[] = [];
    for (const slot of freed) {
      const capacity = Number(slot.capacityTons) || remaining || 1;
      const take = Math.min(remaining || capacity, capacity);
      created.push(
        await allocRepo.save(
          allocRepo.create({
            trainSetWagonId: slot.id,
            bookingId: booking.id,
            allocatedWeightTons: take,
            loadType: booking.freightType ?? null,
            status: 'LOADED',
          }),
        ),
      );
      remaining = Math.max(0, remaining - take);
      if (remaining <= 0) break;
    }
    if (remaining > 0 && created.length) {
      await allocRepo.update(created[created.length - 1].id, {
        allocatedWeightTons: () => `allocated_weight_tons + ${remaining}`,
      } as never);
    }

    // Container numbers onto the first allocation, from the booking's container
    // lines (per physical unit when recorded, else per line).
    const lines = await manager
      .getRepository(BookingContainer)
      .find({ where: { bookingId: booking.id }, relations: { units: true } });
    const itemRepo = manager.getRepository(WagonAllocationContainerItem);
    const first = created[0];
    for (const line of lines) {
      const units = line.units?.length ? line.units : [null];
      for (const unit of units) {
        await itemRepo.save(
          itemRepo.create({
            wagonBookingAllocationId: first.id,
            bookingContainerId: line.id,
            containerNumber: unit?.containerNumber ?? line.containerNumber ?? null,
            containerTypeId: line.containerTypeId ?? null,
          }),
        );
      }
    }
  }

  private async setAllocationStatuses(
    manager: EntityManager,
    scheduleId: string,
    bookingId: string,
    status: 'LOADED' | 'DEPARTED',
  ): Promise<void> {
    const allocations = await this.allocationsForBooking(manager, scheduleId, bookingId);
    if (!allocations.length) return;
    await manager
      .getRepository(WagonBookingAllocation)
      .update({ id: In(allocations.map((a) => a.id)) }, { status });
  }

  private async allocationsForBooking(
    manager: EntityManager,
    scheduleId: string,
    bookingId: string,
  ): Promise<Array<WagonBookingAllocation & { trainSetWagon?: TrainSetWagon }>> {
    return manager
      .getRepository(WagonBookingAllocation)
      .createQueryBuilder('alloc')
      .innerJoinAndSelect('alloc.trainSetWagon', 'slot')
      .innerJoin(
        TrainSchedule,
        'schedule',
        'schedule.train_set_id = slot.train_set_id AND schedule.id = :scheduleId',
        { scheduleId },
      )
      .where('alloc.booking_id = :bookingId', { bookingId })
      .getMany();
  }

  /**
   * On unload: write the wagon_movements ledger rows (board yard → unload yard,
   * kind LOADED) for the booking's pinned wagons, and release each wagon whose
   * slot alights here — it detaches, stays at this yard, and becomes Available
   * (dynamic consist). Wagons shared with a still-loaded consolidated partner
   * stay pinned until the last booking on the slot unloads.
   */
  private async settleWagonsOnUnload(
    manager: EntityManager,
    schedule: TrainSchedule,
    booking: Booking,
    now: Date,
    userId: string | null,
  ): Promise<void> {
    const allocations = await this.allocationsForBooking(manager, schedule.id, booking.id);
    for (const alloc of allocations) {
      const slot = alloc.trainSetWagon;
      if (!slot?.physicalWagonId) continue;

      const boardYardId = slot.boardYardId ?? schedule.originStationId;
      await manager.getRepository(WagonMovement).save(
        manager.getRepository(WagonMovement).create({
          wagonId: slot.physicalWagonId,
          fromYardId: boardYardId,
          toYardId: booking.destinationYardId,
          trainScheduleId: schedule.id,
          bookingId: booking.id,
          kind: Freight.WagonMovementKind.Loaded,
          movedByUserId: userId,
          occurredAt: now,
        }),
      );

      // Detach only when this yard is where the slot's leg ends and no other
      // booking on the wagon is still in transit.
      const slotAlightYardId = slot.alightYardId ?? schedule.destinationStationId;
      if (slotAlightYardId !== booking.destinationYardId) continue;
      const siblings = await manager
        .getRepository(WagonBookingAllocation)
        .createQueryBuilder('alloc')
        .innerJoin('alloc.booking', 'b')
        .where('alloc.train_set_wagon_id = :slotId', { slotId: slot.id })
        .andWhere('alloc.booking_id != :bookingId', { bookingId: booking.id })
        .andWhere(`b.status = 'IN_TRANSIT'`)
        .getCount();
      if (siblings > 0) continue;

      await manager.getRepository(TrainSetWagon).update(slot.id, { status: 'DEPARTED' });
      const wagon = await manager
        .getRepository(Wagon)
        .findOne({ where: { id: slot.physicalWagonId } });
      // Only settle a wagon still bound to this schedule (it may have been
      // re-pinned elsewhere already).
      if (wagon && wagon.currentTrainScheduleId === schedule.id) {
        await manager.getRepository(Wagon).update(wagon.id, {
          currentYardId: booking.destinationYardId,
          currentTrainScheduleId: null,
          trainSetWagonId: null,
          // A wagon that belongs to a built train stays coupled to it (ASSIGNED);
          // only loose wagons return to the open AVAILABLE pool. Marking a
          // coupled wagon AVAILABLE made it show up in the train-builder's
          // "available wagons" picker, where attaching it always 409'd.
          status: wagon.trainId
            ? Freight.WagonStatus.Assigned
            : Freight.WagonStatus.Available,
        });
      }
    }
  }

  private async completeMilestones(booking: Booking, codes: string[]): Promise<void> {
    if (!this.milestoneService || !codes.length) return;
    for (const code of codes) {
      try {
        await this.milestoneService.completeByDocTrigger({ bookingId: booking.id }, code);
      } catch (err) {
        this.logger.warn(
          `Milestone ${code} completion failed for booking ${booking.id}: ${(err as Error).message}`,
        );
      }
    }
  }
}