changes

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

@@ -4,6 +4,7 @@
   SchedulingStatus,
   TrainCheckpointKind,
   TrainScheduleStatus as TrainScheduleStatusEnum,
+  WagonMovementKind,
   WagonStatus,
 } from '@edr/types';
 import {
@@ -27,6 +28,7 @@ import { Container } from '../container-management/entities/container.entity';
 import { Locomotive } from '../locomotives/entities/locomotive.entity';
 import { LocomotivesRepository } from '../locomotives/locomotives.repository';
 import { formatRouteLabel, Route } from '../routes/entities/route.entity';
+import { WagonMovement } from '../wagons/entities/wagon-movement.entity';
 import { TrainSetLocomotive } from '../train-sets/entities/train-set-locomotive.entity';
 import { TrainSetWagon } from '../train-sets/entities/train-set-wagon.entity';
 import { TrainSet } from '../train-sets/entities/train-set.entity';
@@ -113,6 +115,7 @@ import {
   eatDay,
 } from './batch-window.util';
 import { TrainCheckpointEvent } from './entities/train-checkpoint-event.entity';
+import { BookingJourneyService } from './booking-journey.service';
 import { TrainCheckpointEventsRepository } from './train-checkpoint-events.repository';
 import { RecordCheckpointDto } from './dto/record-checkpoint.dto';
 import { RouteMilestone } from '../routes/entities/route-milestone.entity';
@@ -276,6 +279,7 @@ export class TrainSchedulingService {
     private readonly warehouseInventoryService: WarehouseInventoryService,
     private readonly pdfDocuments: WarehouseReleaseDocumentService,
     private readonly bookingWindowGateway: BookingWindowGateway,
+    private readonly bookingJourneyService: BookingJourneyService,
     @Optional() private readonly milestoneService?: ClearanceMilestoneService,
     private readonly configService?: ConfigService,
   ) {}
@@ -290,15 +294,26 @@ export class TrainSchedulingService {
   private async completeMilestonesForScheduleBookings(
     scheduleId: string,
     codes: string[],
+    filter?: { originYardId?: string; destinationYardId?: string },
   ): Promise<void> {
     if (!this.milestoneService || codes.length === 0) return;
     try {
+      const conditions = ['tsb.train_schedule_id = $1', 'tsb.deleted_at IS NULL'];
+      const params: unknown[] = [scheduleId];
+      if (filter?.originYardId) {
+        params.push(filter.originYardId);
+        conditions.push(`b.origin_yard_id = $${params.length}`);
+      }
+      if (filter?.destinationYardId) {
+        params.push(filter.destinationYardId);
+        conditions.push(`b.destination_yard_id = $${params.length}`);
+      }
       const rows: Array<{ booking_id: string }> = await this.dataSource.query(
         `SELECT tsb.booking_id
            FROM freight.train_schedule_bookings tsb
-          WHERE tsb.train_schedule_id = $1
-            AND tsb.deleted_at IS NULL`,
-        [scheduleId],
+           JOIN freight.bookings b ON b.id = tsb.booking_id
+          WHERE ${conditions.join(' AND ')}`,
+        params,
       );
       for (const { booking_id } of rows) {
         for (const code of codes) {
@@ -1457,6 +1472,24 @@ export class TrainSchedulingService {
           manager,
         );
       }
+      // Per-booking journey fallback: bookings boarding at the TRAIN's origin
+      // that the operator didn't load individually are auto-loaded now — the
+      // train is leaving with them. Mid-corridor boarders stay PAID until the
+      // operator loads them at their own yard.
+      await manager.query(
+        `UPDATE freight.bookings b
+            SET status = 'IN_TRANSIT',
+                loaded_at = COALESCE(b.loaded_at, $3)
+           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.origin_yard_id = $2
+            AND b.loaded_at IS NULL
+            AND (b.status = 'PAID' OR (b.is_government = true AND b.status = 'APPROVED'))`,
+        [scheduleId, schedule.originStationId, now],
+      );
       // Close the booking window; any still-pending (unallocated) reservations don't ride this train.
       await manager
         .getRepository(TrainSchedule)
@@ -1488,18 +1521,24 @@ export class TrainSchedulingService {
     // Dispatch closed the window — drop it from portal/GL cards right away.
     void this.emitWindowState(scheduleId);
     // Customer tracking: cargo is on the departing train — loading milestones
-    // plus the direction's "departed" handoff milestone.
+    // plus the direction's "departed" handoff milestone. Restricted to bookings
+    // that BOARD at the train's origin; mid-corridor boarders get their loading
+    // milestones from their own operator load at their own yard.
     if (schedule.direction === 'IMPORT' || schedule.direction === 'EXPORT') {
-      void this.completeMilestonesForScheduleBookings(scheduleId, [
-        // CARGO_ARRIVED is export-only (cargo reached the origin yard) — the
-        // doc-trigger path no-ops it for import bookings.
-        'CARGO_ARRIVED',
-        'READY_FOR_LOADING',
-        'LOADED',
-        schedule.direction === 'IMPORT'
-          ? 'DEPARTED_FROM_DJIBOUTI'
-          : 'DEPARTED_TO_DJIBOUTI',
-      ]);
+      void this.completeMilestonesForScheduleBookings(
+        scheduleId,
+        [
+          // CARGO_ARRIVED is export-only (cargo reached the origin yard) — the
+          // doc-trigger path no-ops it for import bookings.
+          'CARGO_ARRIVED',
+          'READY_FOR_LOADING',
+          'LOADED',
+          schedule.direction === 'IMPORT'
+            ? 'DEPARTED_FROM_DJIBOUTI'
+            : 'DEPARTED_TO_DJIBOUTI',
+        ],
+        { originYardId: schedule.originStationId },
+      );
     }
     return this.getTrainScheduleById(scheduleId);
   }
@@ -2426,18 +2465,11 @@ export class TrainSchedulingService {
         });
       }
 
-      await manager.query(
-        `UPDATE freight.bookings b
-            SET status = $2,
-                scheduling_status = $3
-           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.status NOT IN ('DRAFT', 'REJECTED', 'CANCELLED', 'COMPLETED')`,
-        [scheduleId, 'IN_TRANSIT', SchedulingStatus.Dispatched],
-      );
+      // Per-booking journey: bookings destined for the FINAL yard that the
+      // operator didn't unload individually get their arrival stamped now as a
+      // bulk fallback. Mid-corridor bookings are NOT touched — their arrival is
+      // their own unload (possibly already done while the train kept rolling).
+      await this.bookingJourneyService.autoArriveAtFinalYard(manager, schedule, now);
 
       // Release every locomotive of the set (not just the legacy primary) and move it
       // to the destination yard where it physically arrived.
@@ -2455,12 +2487,33 @@ export class TrainSchedulingService {
           .getRepository(Wagon)
           .findOne({ where: { id: slot.physicalWagonId } });
         if (!wagon) continue;
+        // A wagon that already alighted mid-route (unload released it, possibly
+        // re-pinned elsewhere since) is no longer this schedule's to move.
+        if (wagon.currentTrainScheduleId !== scheduleId) continue;
+        // Dynamic consist: the wagon settles at its slot's alight yard, not
+        // blanket at the train's destination.
+        const settleYardId = slot.alightYardId ?? schedule.destinationStationId;
         await manager.getRepository(Wagon).update(wagon.id, {
           currentTrainScheduleId: null,
           trainSetWagonId: null,
           status: WagonStatus.Available,
-          currentYardId: schedule.destinationStationId,
+          currentYardId: settleYardId,
         });
+        // Ledger: the wagon rode this schedule to its settle yard.
+        const slotAllocations = slot.allocations ?? [];
+        await manager.getRepository(WagonMovement).save(
+          manager.getRepository(WagonMovement).create({
+            wagonId: wagon.id,
+            fromYardId: slot.boardYardId ?? schedule.originStationId,
+            toYardId: settleYardId,
+            trainScheduleId: scheduleId,
+            bookingId: slotAllocations[0]?.bookingId ?? null,
+            kind: slotAllocations.length
+              ? WagonMovementKind.Loaded
+              : WagonMovementKind.EmptyReposition,
+            occurredAt: now,
+          }),
+        );
       }
 
       // Ensure a destination checkpoint exists so the timeline shows ARRIVED.
@@ -2482,11 +2535,15 @@ export class TrainSchedulingService {
       }
     });
 
-    // Customer tracking: the train reached the corridor's far end.
+    // Customer tracking: the train reached the corridor's far end. Restricted
+    // to bookings destined for the FINAL yard — mid-corridor bookings get their
+    // arrival milestone from their own operator unload at their own yard.
     if (schedule.direction === 'IMPORT' || schedule.direction === 'EXPORT') {
-      void this.completeMilestonesForScheduleBookings(scheduleId, [
-        schedule.direction === 'IMPORT' ? 'ARRIVED_ETHIOPIA' : 'ARRIVED_AT_DJIBOUTI',
-      ]);
+      void this.completeMilestonesForScheduleBookings(
+        scheduleId,
+        [schedule.direction === 'IMPORT' ? 'ARRIVED_ETHIOPIA' : 'ARRIVED_AT_DJIBOUTI'],
+        { destinationYardId: schedule.destinationStationId },
+      );
     }
 
     const detail = await this.getTrainScheduleById(scheduleId);
@@ -2635,17 +2692,26 @@ export class TrainSchedulingService {
     }
 
     if (
-      bookings.some((b) => {
-        if (targetScheduleId && b.trainScheduleId === targetScheduleId) {
-          return false;
+      await (async () => {
+        // Corridor-aware: a booking belongs on this train when its origin and
+        // destination lie on the schedule's stop list in order — sub-corridor
+        // bookings (Dire→Djibouti on an Addis→…→Djibouti train) are valid.
+        let stops = [dto.originStationId, dto.destinationStationId];
+        if (targetScheduleId) {
+          const target = await this.trainSchedulesRepository.findById(targetScheduleId);
+          if (target) stops = await this.stopYardsForSchedule(target);
         }
-        return (
-          b.originYardId !== dto.originStationId ||
-          b.destinationYardId !== dto.destinationStationId
-        );
-      })
+        return bookings.some((b) => {
+          if (targetScheduleId && b.trainScheduleId === targetScheduleId) {
+            return false;
+          }
+          const fromIdx = stops.indexOf(b.originYardId);
+          const toIdx = stops.indexOf(b.destinationYardId);
+          return fromIdx < 0 || toIdx < 0 || fromIdx >= toIdx;
+        });
+      })()
     ) {
-      violations.push('Selected bookings must share the same origin and destination as the schedule');
+      violations.push('Selected bookings must lie on the schedule route (origin before destination)');
     }
 
     if (!forceAssign) {
@@ -2695,7 +2761,37 @@ export class TrainSchedulingService {
     }
 
     const originYardId = dto.originStationId;
-    const fleetCounts = await this.countFleetAvailability(originYardId, targetScheduleId);
+    // Dynamic consist: a slot's physical wagon may ride from the train's origin
+    // OR already sit at the booking's own boarding yard and attach there — so
+    // the usable fleet is the union across the origin and every boarding yard.
+    const boardYardIds = [
+      ...new Set(
+        [originYardId, ...bookings.map((b) => b.originYardId)].filter(Boolean),
+      ),
+    ];
+    const fleetCountsByYard = await Promise.all(
+      boardYardIds.map((yardId) =>
+        this.countFleetAvailability(yardId, targetScheduleId),
+      ),
+    );
+    const mergedFleet = new Map<string, { code: string; available: number }>();
+    for (const rows of fleetCountsByYard) {
+      for (const row of rows) {
+        const existing = mergedFleet.get(row.wagonTypeId) ?? {
+          code: row.wagonTypeCode,
+          available: 0,
+        };
+        existing.available += row.available;
+        mergedFleet.set(row.wagonTypeId, existing);
+      }
+    }
+    const fleetCounts = [...mergedFleet.entries()].map(
+      ([wagonTypeId, value]) => ({
+        wagonTypeId,
+        wagonTypeCode: value.code,
+        available: value.available,
+      }),
+    );
     const fleetByTypeId = new Map(fleetCounts.map((row) => [row.wagonTypeId, row.available]));
     fleetAvailability = computeFleetAvailability(
       demandPlan,
@@ -2720,6 +2816,12 @@ export class TrainSchedulingService {
       containerWagonType,
       bulkWagonType,
     });
+    this.stampSlotLegs(
+      wagonPlan,
+      fittingBookings,
+      dto.originStationId,
+      dto.destinationStationId,
+    );
 
     violations.push(
       ...(await this.validatePhysicalFleetForPlan(
@@ -3047,6 +3149,7 @@ export class TrainSchedulingService {
         wagonTypeId: slot.wagonTypeId,
         wagonTypeCode: typeCodeById.get(slot.wagonTypeId) ?? slot.wagonTypeId,
         trainSetWagonId: slot.id,
+        boardYardId: slot.boardYardId ?? null,
       }));
 
     const unpinnable = this.findUnpinnableWagonSlots(
@@ -3100,6 +3203,7 @@ export class TrainSchedulingService {
         sequenceNo: slot.sequenceNo,
         wagonTypeId: slot.wagonTypeId,
         wagonTypeCode: slot.wagonTypeCode,
+        boardYardId: slot.boardYardId ?? null,
       })),
       wagons,
       targetScheduleId,
@@ -3108,7 +3212,12 @@ export class TrainSchedulingService {
   }
 
   private findUnpinnableWagonSlots(
-    slots: Array<{ sequenceNo: number; wagonTypeId: string; wagonTypeCode: string }>,
+    slots: Array<{
+      sequenceNo: number;
+      wagonTypeId: string;
+      wagonTypeCode: string;
+      boardYardId?: string | null;
+    }>,
     wagons: Wagon[],
     scheduleId: string | undefined,
     originYardId: string,
@@ -3136,22 +3245,35 @@ export class TrainSchedulingService {
     return violations;
   }
 
+  /**
+   * Dynamic consist: a slot's wagon may either ride from the train's origin
+   * yard (attaching there, possibly empty until the slot's board yard) or
+   * already sit AT the slot's board yard and hook on when the train arrives.
+   */
   private pickPhysicalWagonForSlot(
-    slot: { wagonTypeId: string },
+    slot: { wagonTypeId: string; boardYardId?: string | null },
     wagons: Wagon[],
     scheduleId: string | undefined,
     originYardId: string,
     assignedPhysicalIds: Set<string>,
   ): Wagon | undefined {
-    return wagons.find((wagon) => {
+    const usable = (wagon: Wagon): boolean => {
       if (wagon.wagonTypeId !== slot.wagonTypeId) return false;
       if (assignedPhysicalIds.has(wagon.id)) return false;
       const pinnedOnSchedule = scheduleId
         ? wagon.currentTrainScheduleId === scheduleId
         : false;
-      if (wagon.status !== WagonStatus.Available && !pinnedOnSchedule) return false;
-      return wagon.currentYardId === originYardId;
-    });
+      return wagon.status === WagonStatus.Available || pinnedOnSchedule;
+    };
+    // Prefer a wagon already waiting at the slot's board yard (no empty haul);
+    // fall back to one riding from the train's origin.
+    if (slot.boardYardId) {
+      const atBoardYard = wagons.find(
+        (w) => usable(w) && w.currentYardId === slot.boardYardId,
+      );
+      if (atBoardYard) return atBoardYard;
+    }
+    return wagons.find((w) => usable(w) && w.currentYardId === originYardId);
   }
 
   private positiveNumber(value: number | undefined, fallback: number): number {
@@ -3303,6 +3425,42 @@ export class TrainSchedulingService {
     return containerType?.wagonType?.isActive ? containerType.wagonType : null;
   }
 
+  /**
+   * Stamp each plan slot with the leg it occupies (dynamic consist): the
+   * boarding/alighting yards of the bookings it carries. Null means the
+   * schedule's own endpoint (whole-route slot, legacy behavior). A slot
+   * carrying bookings with mixed corridors stays whole-route (conservative).
+   */
+  private stampSlotLegs(
+    wagonPlan: WagonPlanSlot[],
+    bookings: Booking[],
+    scheduleOriginYardId: string,
+    scheduleDestinationYardId: string,
+  ): void {
+    const bookingById = new Map(bookings.map((b) => [b.id, b]));
+    for (const slot of wagonPlan) {
+      const slotBookings = [
+        ...new Set(slot.allocations.map((a) => a.bookingId)),
+      ]
+        .map((id) => bookingById.get(id))
+        .filter((b): b is Booking => Boolean(b));
+      if (!slotBookings.length) continue;
+      const [first] = slotBookings;
+      const sameCorridor = slotBookings.every(
+        (b) =>
+          b.originYardId === first.originYardId &&
+          b.destinationYardId === first.destinationYardId,
+      );
+      if (!sameCorridor) continue;
+      slot.boardYardId =
+        first.originYardId === scheduleOriginYardId ? null : first.originYardId;
+      slot.alightYardId =
+        first.destinationYardId === scheduleDestinationYardId
+          ? null
+          : first.destinationYardId;
+    }
+  }
+
   private async persistTrainSetWagons(
     manager: EntityManager,
     trainSetId: string,
@@ -3318,6 +3476,8 @@ export class TrainSchedulingService {
         lengthMeters: slot.lengthMeters,
         assignedWeightTons: slot.assignedWeightTons,
         status: 'PLANNED',
+        boardYardId: slot.boardYardId ?? null,
+        alightYardId: slot.alightYardId ?? null,
       }),
     );
     return manager.getRepository(TrainSetWagon).save(wagons);
@@ -4012,26 +4172,44 @@ export class TrainSchedulingService {
 
     // How many wagons of that type the cargo needs.
     const slotsNeeded = this.wagonsNeededForCargo(input, requiredType);
+    void slotsNeeded; // TEMP: unused while the wagon-availability filter is off.
 
-    // AVAILABLE wagons of the required type, counted once per origin yard.
-    const availableByYard = new Map<string, number>();
-    const availableAt = async (yardId: string): Promise<number> => {
-      const cached = availableByYard.get(yardId);
-      if (cached !== undefined) return cached;
-      const counts = await this.countFleetAvailability(yardId);
-      const n =
-        counts.find((c) => c.wagonTypeId === requiredType.id)?.available ?? 0;
-      availableByYard.set(yardId, n);
-      return n;
-    };
+    // TEMP (per request): wagon-availability filtering is DISABLED. A day is now
+    // offered whenever a bookable schedule that day has remaining train capacity
+    // — regardless of whether matching wagons are actually available at the
+    // origin / boarding yard. This surfaces days even when no wagon is on hand.
+    // Restore the block below to bring back the "enough matching wagons" gate.
+    //
+    // // AVAILABLE wagons of the required type, counted once per origin yard.
+    // const availableByYard = new Map<string, number>();
+    // const availableAt = async (yardId: string): Promise<number> => {
+    //   const cached = availableByYard.get(yardId);
+    //   if (cached !== undefined) return cached;
+    //   const counts = await this.countFleetAvailability(yardId);
+    //   const n =
+    //     counts.find((c) => c.wagonTypeId === requiredType.id)?.available ?? 0;
+    //   availableByYard.set(yardId, n);
+    //   return n;
+    // };
 
     const days = new Set<string>();
     for (const s of schedules) {
       const hasCapacity =
         Math.max(0, (s.maxWagons ?? 0) - (s.trainSet?.wagonCount ?? 0)) > 0;
       if (!hasCapacity) continue;
-      const enoughWagons = (await availableAt(s.originStationId)) >= slotsNeeded;
-      if (!enoughWagons) continue;
+      // TEMP (per request): wagon-availability check commented out — see note
+      // above. Dynamic consist: wagons may ride from the train's origin OR
+      // already sit at the booking's own boarding yard and attach when the train
+      // arrives — either pool can serve a sub-corridor booking.
+      // let enoughWagons = (await availableAt(s.originStationId)) >= slotsNeeded;
+      // if (
+      //   !enoughWagons &&
+      //   input.originYardId &&
+      //   input.originYardId !== s.originStationId
+      // ) {
+      //   enoughWagons = (await availableAt(input.originYardId)) >= slotsNeeded;
+      // }
+      // if (!enoughWagons) continue;
       if (s.scheduledDepartureDate)
         days.add(eatDay(new Date(s.scheduledDepartureDate)));
     }
@@ -4063,6 +4241,33 @@ export class TrainSchedulingService {
     return Math.max(1, Math.ceil(teu / 2));
   }
 
+  /**
+   * Ordered stop yards of a schedule's route: origin → milestones → destination,
+   * de-duplicated. Falls back to the two-endpoint pseudo-route when the schedule
+   * has no route milestones. Shared by corridor (sub-leg) validation everywhere.
+   */
+  async stopYardsForSchedule(schedule: TrainSchedule): Promise<string[]> {
+    let milestoneYards: string[] = [];
+    if (schedule.route?.milestones?.length) {
+      milestoneYards = [...schedule.route.milestones]
+        .sort((a, b) => a.sequenceNo - b.sequenceNo)
+        .map((m) => m.yardId);
+    } else if (schedule.routeId) {
+      const milestones = await this.dataSource
+        .getRepository(RouteMilestone)
+        .find({ where: { routeId: schedule.routeId }, order: { sequenceNo: 'ASC' } });
+      milestoneYards = milestones.map((m) => m.yardId);
+    }
+    const raw = milestoneYards.length >= 2
+      ? milestoneYards
+      : [schedule.originStationId, ...milestoneYards, schedule.destinationStationId];
+    const unique: string[] = [];
+    for (const yardId of raw) {
+      if (yardId && !unique.includes(yardId)) unique.push(yardId);
+    }
+    return unique;
+  }
+
   /** Whether a route has ≥1 OPEN bookable departure on a given EAT day. */
   async existsOpenScheduleOnRouteDay(
     originYardId: string,