changes export flow

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

@@ -9,6 +9,17 @@
 
 export const BATCH_TIMEZONE = 'Africa/Addis_Ababa';
 
+/**
+ * Slack after a booking's paymentDeadline before the settle sweep expires it.
+ * Covers gateway/webhook lag for a payment STARTED inside the window —
+ * initiation itself is hard-blocked at the deadline (BillingService.payInvoice),
+ * so this never extends the time a customer has to begin paying.
+ */
+export const PAYMENT_GRACE_MS = 5 * 60_000;
+
+/** How long before the pay deadline the one reminder notification goes out. */
+export const PAYMENT_REMINDER_LEAD_MS = 10 * 60_000;
+
 /** Fallback wagons-per-booking when a booking has no computed `wagonsRequired`. */
 export const DEFAULT_WAGONS_PER_BOOKING = 1;
 

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

@@ -109,6 +109,7 @@ describe('BookingBatchService — PAID reconcile', () => {
         windowDurationHours: 3,
         docReviewMinutes: 30,
         paymentWindowMinutes: 60,
+        exportPaymentWindowMinutes: 60,
       }),
     };
 
@@ -835,7 +836,8 @@ describe('BookingBatchService — PAID reconcile', () => {
       // One reservation whose pay window lapsed, and one booking on the waiting list.
       const lapsed = booking('lapsed', 50, {
         status: 'SELECTED_FOR_BATCH',
-        paymentDeadline: new Date(Date.now() - 60_000),
+        // Past deadline + the 5-minute webhook grace, so the settle expires it.
+        paymentDeadline: new Date(Date.now() - 6 * 60_000),
       });
       const waiting = booking('waiting', 10, { trainScheduleId: null });
 
@@ -870,7 +872,8 @@ describe('BookingBatchService — PAID reconcile', () => {
     it('serialises concurrent settles so the same reservation is not settled twice', async () => {
       const lapsed = booking('lapsed', 50, {
         status: 'SELECTED_FOR_BATCH',
-        paymentDeadline: new Date(Date.now() - 60_000),
+        // Past deadline + the 5-minute webhook grace, so the settle expires it.
+        paymentDeadline: new Date(Date.now() - 6 * 60_000),
       });
       // Both callers read the reservation; the lock must stop the second from
       // acting on rows the first already expired. (The PAYMENT transition and the
@@ -901,7 +904,8 @@ describe('BookingBatchService — PAID reconcile', () => {
     it('never expires a reservation whose payment landed — allocates it instead', async () => {
       const latePaid = booking('late-paid', 50, {
         status: 'SELECTED_FOR_BATCH',
-        paymentDeadline: new Date(Date.now() - 60_000),
+        // Past deadline + the 5-minute webhook grace, so the settle expires it.
+        paymentDeadline: new Date(Date.now() - 6 * 60_000),
       });
       bookingsRepository.findReservedForSchedule
         .mockResolvedValueOnce([latePaid])
@@ -1015,7 +1019,8 @@ describe('BookingBatchService — PAID reconcile', () => {
         ...(waiting as unknown as Record<string, unknown>),
         status: 'SELECTED_FOR_BATCH',
         trainScheduleId: exportScheduleId,
-        paymentDeadline: new Date(Date.now() - 1_000),
+        // Past deadline + the 5-minute webhook grace, so the settle expires it.
+        paymentDeadline: new Date(Date.now() - 6 * 60_000),
         originYardId: 'yard-a',
         destinationYardId: 'yard-b',
         priorityScore: 0,

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

@@ -61,11 +61,14 @@ import {
   DEFAULT_CONTAINER_WAGON_LENGTH_METERS,
   DEFAULT_CONTAINER_WAGON_TARE_TONS,
   DEFAULT_WAGONS_PER_BOOKING,
+  PAYMENT_GRACE_MS,
+  PAYMENT_REMINDER_LEAD_MS,
 } from "./booking-batch.constants";
 import {
   LocomotiveLimits,
   WagonTypeDimensions,
   bookingCargoTons,
+  bulkItemWagonsRequired,
   bookingGrossWeightTons,
   deriveTrainCapacityFromLocomotive,
   sizePartialOfferWagons,
@@ -115,6 +118,32 @@ export interface ExportSpaceReport {
   fullMessage: string | null;
 }
 
+/**
+ * One export train the customer can pick for a shipment day: live free-wagon
+ * space measured against THE BOOKING'S allowed wagon types (so the per-type
+ * list doubles as "what cargo this train can take for you"). Unpaid holds
+ * count as taken; lapsed holds free up via the lazy-expiry capacity filter.
+ */
+export interface ExportTrainOption {
+  scheduleId: string;
+  departure: Date;
+  /** Booking cutoff for this train (windowClosesAt), null on legacy rows. */
+  bookingClosesAt: Date | null;
+  /** Whether the export FCFS window is open for booking right now. */
+  isOpen: boolean;
+  /** Best bookable wagons across the booking's allowed types. */
+  freeWagons: number;
+  /** Wagons this booking needs — `fits` = freeWagons >= neededWagons. */
+  neededWagons: number;
+  fits: boolean;
+  byWagonType: Array<{
+    wagonTypeId: string | null;
+    code: string | null;
+    name: string | null;
+    freeWagons: number;
+  }>;
+}
+
 /** A day-level pool key: all trains on this route departing on this EAT day. */
 interface RouteDayGroup {
   originYardId: string;
@@ -663,12 +692,16 @@ export class BookingBatchService implements OnModuleInit {
         { status: TrainScheduleStatusEnum.Scheduled },
       ],
     });
+    // A customer-picked train narrows the scan to that ONE schedule: export
+    // FCFS honors the pick or fails loudly (exportFullMessage names it).
+    const requestedId = booking.requestedTrainScheduleId ?? null;
     const candidates = corridor
       .filter(
         (s) =>
           s.scheduledDepartureDate != null &&
           eatDay(s.scheduledDepartureDate) === day &&
-          this.isFillable(s),
+          this.isFillable(s) &&
+          (!requestedId || s.id === requestedId),
       )
       .sort(
         (a, b) =>
@@ -755,13 +788,18 @@ export class BookingBatchService implements OnModuleInit {
 
   /** Customer-facing "train is full" copy carrying the bookable leftover. */
   private exportFullMessage(booking: Booking, report: ExportSpaceReport): string {
+    const picked = Boolean(booking.requestedTrainScheduleId);
     if (!report.trainsForDay || !report.corridorMatched) {
-      return 'No export train is accepting bookings for this day';
+      return picked
+        ? 'The selected train is no longer accepting bookings — pick another train or day.'
+        : 'No export train is accepting bookings for this day';
     }
     const best = report.bestAvailable;
-    const base =
-      'Not enough train space — an export booking must ride a single train whole, ' +
-      'and no open train on this day can carry it. ';
+    const base = picked
+      ? 'Not enough space left on the selected train — an export booking must ' +
+        'ride one train whole. '
+      : 'Not enough train space — an export booking must ride a single train whole, ' +
+        'and no open train on this day can carry it. ';
     if (!best || best.wagons <= 0) {
       return base + 'No capacity is left on this day — pick another shipment day.';
     }
@@ -864,6 +902,88 @@ export class BookingBatchService implements OnModuleInit {
     return out;
   }
 
+  /**
+   * The export train picker: every export train on the booking's corridor/day
+   * with its live space, measured per allowed wagon type so the customer sees
+   * what each train can still take for THEIR cargo. Includes full/not-yet-open
+   * trains (freeWagons 0 / isOpen false) so the UI can show them disabled —
+   * the request-time gate (exportSpaceReport) stays the enforcement point.
+   */
+  async exportTrainOptionsForDay(
+    booking: Booking,
+    day: string,
+  ): Promise<ExportTrainOption[]> {
+    const corridor = await this.trainSchedulesRepository.findAll({
+      where: [
+        { status: TrainScheduleStatusEnum.Draft },
+        { status: TrainScheduleStatusEnum.Scheduled },
+      ],
+    });
+    const candidates = corridor
+      .filter(
+        (s) =>
+          s.scheduledDepartureDate != null &&
+          eatDay(s.scheduledDepartureDate) === day &&
+          s.direction === 'EXPORT',
+      )
+      .sort(
+        (a, b) =>
+          a.scheduledDepartureDate!.getTime() -
+          b.scheduledDepartureDate!.getTime(),
+      );
+
+    const wagonDims = await this.loadWagonDims();
+    const allowed = this.allowedDimsWithTypes(booking, wagonDims);
+    const neededWagons = this.wagonsFor(booking, wagonDims);
+    const typeIds = allowed
+      .map((a) => a.wagonTypeId)
+      .filter((id): id is string => Boolean(id));
+    const types = typeIds.length
+      ? await this.dataSource
+          .getRepository(WagonType)
+          .find({ where: { id: In(typeIds) } })
+      : [];
+    const typeById = new Map(types.map((t) => [t.id, t]));
+
+    const out: ExportTrainOption[] = [];
+    for (const candidate of candidates) {
+      const schedule = await this.trainSchedulesRepository.findByIdWithFullGraph(
+        candidate.id,
+      );
+      const locomotive = trainSetLocomotiveLimits(schedule?.trainSet);
+      if (!schedule || !locomotive) continue;
+      const limits = await this.capacityLimits(locomotive);
+      const budget = await this.remainingBudget(schedule, limits, wagonDims);
+      const leg = budget.legOf(booking.originYardId, booking.destinationYardId);
+      if (!leg) continue; // this train's route doesn't carry the booking's leg
+      const room = budget.remainingFor(leg);
+      const byWagonType = allowed.map(({ wagonTypeId, dims }) => {
+        const type = wagonTypeId ? typeById.get(wagonTypeId) : undefined;
+        return {
+          wagonTypeId,
+          code: type?.code ?? null,
+          name: type?.name ?? null,
+          freeWagons: this.bookableWithin(room, dims).wagons,
+        };
+      });
+      const freeWagons = byWagonType.reduce(
+        (best, t) => Math.max(best, t.freeWagons),
+        0,
+      );
+      out.push({
+        scheduleId: schedule.id,
+        departure: schedule.scheduledDepartureDate!,
+        bookingClosesAt: schedule.windowClosesAt ?? null,
+        isOpen: this.isFillable(schedule),
+        freeWagons,
+        neededWagons,
+        fits: freeWagons >= neededWagons,
+        byWagonType,
+      });
+    }
+    return out;
+  }
+
   /**
    * Advisory free-wagon count for an IMPORT/DOMESTIC booking on a given day,
    * summed across every train on the booking's corridor that day. Unlike the
@@ -2207,7 +2327,10 @@ export class BookingBatchService implements OnModuleInit {
     };
     if (!this.fits(offeredNeed, budget)) return null;
 
-    const deadline = new Date(Date.now() + (await this.paymentWindowMs()));
+    const deadline = new Date(
+      Date.now() +
+        (await this.paymentWindowMsFor(await this.scheduleById(scheduleId))),
+    );
     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.
@@ -2216,6 +2339,7 @@ export class BookingBatchService implements OnModuleInit {
       status: "SELECTED_FOR_BATCH",
       selectedForBatchAt: new Date(),
       paymentDeadline: deadline,
+      paymentReminderSentAt: null,
     } as never);
     booking.trainScheduleId = scheduleId;
     return offeredNeed;
@@ -2244,9 +2368,11 @@ export class BookingBatchService implements OnModuleInit {
 
     const isPaid = (b: Booking) =>
       b.paymentStatus === "PAID" || b.status === "PAID";
+    // Grace: a payment started inside the window may land minutes late via the
+    // gateway webhook — don't expire until the slack has passed too.
     const isExpired = (b: Booking) =>
       b.paymentDeadline
-        ? b.paymentDeadline.getTime() <= now
+        ? b.paymentDeadline.getTime() + PAYMENT_GRACE_MS <= now
         : expireUnpaidUnknownDeadline;
 
     for (const booking of reserved) {
@@ -2544,6 +2670,39 @@ export class BookingBatchService implements OnModuleInit {
     this.notifyBoardChanged(newScheduleId, "booking_moved");
   }
 
+  /**
+   * One reminder per hold, shortly before its pay deadline (the window tick
+   * calls this every pass; `payment_reminder_sent_at` dedups). Skips paid
+   * bookings — a landed payment the settle hasn't processed yet needs no nag.
+   */
+  async sendPaymentReminders(): Promise<void> {
+    const now = new Date();
+    const due = await this.dataSource
+      .getRepository(Booking)
+      .createQueryBuilder("b")
+      .leftJoinAndSelect("b.company", "company")
+      .where(`b.status IN ('SELECTED_FOR_BATCH', 'AWAITING_PAYMENT')`)
+      .andWhere(`b.payment_status != 'PAID'`)
+      .andWhere("b.payment_reminder_sent_at IS NULL")
+      .andWhere("b.payment_deadline > :now", { now })
+      .andWhere("b.payment_deadline <= :soon", {
+        soon: new Date(now.getTime() + PAYMENT_REMINDER_LEAD_MS),
+      })
+      .getMany();
+    for (const booking of due) {
+      // Stamp BEFORE sending so a slow notifier can't double-send next tick.
+      await this.bookingsRepository.update(booking.id, {
+        paymentReminderSentAt: new Date(),
+      } as never);
+      if (booking.paymentDeadline) {
+        await this.notifier.payDeadlineApproaching(
+          booking,
+          booking.paymentDeadline,
+        );
+      }
+    }
+  }
+
   /** Staff "expire" override → free a reservation now (booking becomes EXPIRED). */
   async expireReservation(bookingId: string): Promise<void> {
     const booking = await this.dataSource
@@ -2564,6 +2723,52 @@ export class BookingBatchService implements OnModuleInit {
     }
   }
 
+  /**
+   * Customer cancel of an unpaid hold: the same immediate release as
+   * expireReservation, but the booking ends CANCELLED (the customer chose to
+   * walk away — "payment window missed" copy would be wrong). Consolidated
+   * pairs are rejected by the caller: the shared wagon is both-or-neither.
+   */
+  async cancelReservation(bookingId: string): Promise<void> {
+    const booking = await this.dataSource
+      .getRepository(Booking)
+      .findOne({ where: { id: bookingId } });
+    if (!booking) throw new NotFoundException(`Booking ${bookingId} not found`);
+    const freedScheduleId = booking.trainScheduleId;
+    await this.bookingsRepository.update(booking.id, {
+      trainScheduleId: null,
+      requestedTrainScheduleId: null,
+      status: "CANCELLED",
+      schedulingStatus: "ELIGIBLE",
+      paymentDeadline: null,
+      selectedForBatchAt: null,
+      paymentReminderSentAt: null,
+    } as never);
+    // An unpaid partial offer dies with the hold — same as expire().
+    if (this.splitService) {
+      await this.splitService.expireOpenOffer(booking.id);
+    }
+    await this.billing.expirePayable(
+      Freight.InvoiceSource.Booking,
+      booking.id,
+      "PREPAID",
+    );
+    if (freedScheduleId) {
+      // Same release choreography as expireReservation: reopen a FULL window,
+      // top up from the waiting list, push one board update with final state.
+      await this.refreshWindowStatus(freedScheduleId);
+      const topUpReserved = await this.topUpFill(freedScheduleId);
+      if (topUpReserved > 0) {
+        await this.extendPaymentPhaseForTopUp(freedScheduleId);
+      }
+      this.notifyBoardChanged(freedScheduleId, "reservation_expired");
+    }
+    this.logger.log(
+      `[BATCH] CANCELLED hold ${booking.reference} — customer released the ` +
+        `reservation before paying; wagons freed`,
+    );
+  }
+
   // ---- intercity ride-along API ---------------------------------------------
 
   /**
@@ -2648,14 +2853,14 @@ export class BookingBatchService implements OnModuleInit {
       return;
     }
     const now = new Date();
-    let deadline = new Date(now.getTime() + (await this.paymentWindowMs()));
+    const targetSchedule = await this.scheduleById(scheduleId);
+    let deadline = new Date(
+      now.getTime() + (await this.paymentWindowMsFor(targetSchedule)),
+    );
     // EXPORT parity: pay windows on an export train never outlive its booking
     // window — export bookings expire at close, so anything reserved onto the
     // same train (FCFS export or an intercity ride-along) must too. Import
     // keeps the plain payment window; its cycles re-fill after settle.
-    const targetSchedule = await this.dataSource
-      .getRepository(TrainSchedule)
-      .findOne({ where: { id: scheduleId } });
     if (targetSchedule?.direction === "EXPORT") {
       const cutoff =
         targetSchedule.windowClosesAt ?? targetSchedule.scheduledDepartureDate;
@@ -2673,6 +2878,7 @@ export class BookingBatchService implements OnModuleInit {
       status: "SELECTED_FOR_BATCH",
       selectedForBatchAt: now,
       paymentDeadline: deadline,
+      paymentReminderSentAt: null,
     } as never);
     booking.trainScheduleId = scheduleId;
     // The invoice was generated DRAFT at booking creation / operation-accept,
@@ -2860,10 +3066,13 @@ export class BookingBatchService implements OnModuleInit {
     const freedScheduleId = booking.trainScheduleId;
     await this.bookingsRepository.update(booking.id, {
       trainScheduleId: null,
+      // The customer's train pick died with the hold — a rebook re-picks.
+      requestedTrainScheduleId: null,
       status: "EXPIRED",
       schedulingStatus: "ELIGIBLE",
       paymentDeadline: null,
       selectedForBatchAt: null,
+      paymentReminderSentAt: null,
     } as never);
     booking.trainScheduleId = null;
     // The wagons this reservation held are back — a schedule parked at FULL
@@ -3406,7 +3615,11 @@ export class BookingBatchService implements OnModuleInit {
     const byWeight =
       cargoTons > 0 && capacityTons > 0 ? Math.ceil(cargoTons / capacityTons) : 0;
 
-    return Math.max(DEFAULT_WAGONS_PER_BOOKING, stored, byLength, byWeight);
+    // Break-bulk (PER_ITEM): indivisible items can need more wagons than raw
+    // tonnage suggests (floor items-per-wagon loses the fractional capacity).
+    const byItems = bulkItemWagonsRequired(booking, capacityTons);
+
+    return Math.max(DEFAULT_WAGONS_PER_BOOKING, stored, byLength, byWeight, byItems);
   }
 
   /**
@@ -3602,6 +3815,18 @@ export class BookingBatchService implements OnModuleInit {
    * representative dims when no allowed type is configured.
    */
   private dimsForAllowed(booking: Booking, wagonDims: WagonDims): PerWagonDims[] {
+    return this.allowedDimsWithTypes(booking, wagonDims).map((p) => p.dims);
+  }
+
+  /**
+   * Same allowed set as {@link dimsForAllowed} but keeping each wagon-type id,
+   * so callers (the export train picker) can label per-type availability.
+   * `wagonTypeId` is null only on the unconfigured fallback entry.
+   */
+  private allowedDimsWithTypes(
+    booking: Booking,
+    wagonDims: WagonDims,
+  ): Array<{ wagonTypeId: string | null; dims: PerWagonDims }> {
     const fallback =
       booking.freightType === "BULK" ? wagonDims.bulk : wagonDims.container;
     const ids =
@@ -3611,19 +3836,23 @@ export class BookingBatchService implements OnModuleInit {
             .flatMap((line) => line.containerType?.wagonTypes ?? [])
             .map((wt) => wt.id);
     const seen = new Set<string>();
-    const dims: PerWagonDims[] = [];
+    const out: Array<{ wagonTypeId: string | null; dims: PerWagonDims }> = [];
     for (const id of ids) {
       if (!id || seen.has(id)) continue;
       seen.add(id);
       const d = wagonDims.byWagonTypeId.get(id);
       if (d) {
-        dims.push({
-          ...d,
-          capacityTons: d.capacityTons > 0 ? d.capacityTons : fallback.capacityTons,
+        out.push({
+          wagonTypeId: id,
+          dims: {
+            ...d,
+            capacityTons:
+              d.capacityTons > 0 ? d.capacityTons : fallback.capacityTons,
+          },
         });
       }
     }
-    return dims.length ? dims : [fallback];
+    return out.length ? out : [{ wagonTypeId: null, dims: fallback }];
   }
 
   /**
@@ -3808,8 +4037,18 @@ export class BookingBatchService implements OnModuleInit {
     const allocated = (schedule.scheduleBookings ?? [])
       .map((sb) => sb.booking)
       .filter((b): b is Booking => Boolean(b));
-    const reserved = await this.bookingsRepository.findReservedForSchedule(
-      schedule.id,
+    // Lazy-expiry guard: a hold whose deadline + grace has lapsed no longer
+    // blocks capacity, even before the 10s sweep flips it to EXPIRED — so
+    // availability shown to the next customer is honest between ticks.
+    const graceCutoff = Date.now() - PAYMENT_GRACE_MS;
+    const reserved = (
+      await this.bookingsRepository.findReservedForSchedule(schedule.id)
+    ).filter(
+      (b) =>
+        b.paymentStatus === "PAID" ||
+        b.status === "PAID" ||
+        b.paymentDeadline == null ||
+        b.paymentDeadline.getTime() > graceCutoff,
     );
     for (const b of [...allocated, ...reserved]) {
       budget.subtract(
@@ -3905,7 +4144,7 @@ export class BookingBatchService implements OnModuleInit {
         b.paymentStatus !== "PAID" &&
         b.status !== "PAID" &&
         b.paymentDeadline != null &&
-        b.paymentDeadline.getTime() > now,
+        b.paymentDeadline.getTime() + PAYMENT_GRACE_MS > now,
     );
   }
 
@@ -4089,10 +4328,33 @@ export class BookingBatchService implements OnModuleInit {
 
   // ---- timer plumbing -------------------------------------------------------
 
-  /** Configured customer pay window in ms (global rules, with defaults). */
-  private async paymentWindowMs(): Promise<number> {
+  /**
+   * Effective customer pay window in ms for a target schedule: the staff
+   * per-schedule override wins, else the global value for the schedule's
+   * direction (export and import pay windows are tuned independently).
+   * No schedule (unknown target) falls back to the import global.
+   */
+  private async paymentWindowMsFor(
+    schedule?: Pick<
+      TrainSchedule,
+      "direction" | "rulePaymentWindowMinutes"
+    > | null,
+  ): Promise<number> {
+    if (schedule?.rulePaymentWindowMinutes != null) {
+      return schedule.rulePaymentWindowMinutes * 60_000;
+    }
     const cfg = await this.trainSchedulingService.getWindowConfig();
-    return cfg.paymentWindowMinutes * 60_000;
+    const minutes =
+      schedule?.direction === "EXPORT"
+        ? cfg.exportPaymentWindowMinutes
+        : cfg.paymentWindowMinutes;
+    return minutes * 60_000;
+  }
+
+  private scheduleById(id: string): Promise<TrainSchedule | null> {
+    return this.dataSource
+      .getRepository(TrainSchedule)
+      .findOne({ where: { id } });
   }
 
   private timeoutName(scheduleId: string): string {
@@ -4104,8 +4366,12 @@ export class BookingBatchService implements OnModuleInit {
    * engine's minute tick calling settleDueReservations off `paymentDeadline`.
    */
   private armSettle(scheduleId: string): void {
-    void this.paymentWindowMs()
-      .then((delayMs) => {
+    void this.scheduleById(scheduleId)
+      .then((schedule) => this.paymentWindowMsFor(schedule))
+      // The timer covers the grace too — firing at the bare deadline would
+      // settle before the sweep's grace cutoff and find nothing to expire.
+      .then((windowMs: number) => {
+        const delayMs = windowMs + PAYMENT_GRACE_MS;
         this.removeTimeout(scheduleId);
         const handle = setTimeout(() => {
           void this.settleBatch(scheduleId).catch((err) =>
@@ -4138,7 +4404,7 @@ export class BookingBatchService implements OnModuleInit {
       .getRepository(TrainSchedule)
       .findOne({ where: { id: scheduleId } });
     if (!schedule || schedule.windowPhase !== "PAYMENT") return;
-    const windowMs = await this.paymentWindowMs();
+    const windowMs = await this.paymentWindowMsFor(schedule);
     let target = new Date(Date.now() + windowMs);
     if (
       schedule.scheduledDepartureDate &&

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

@@ -137,6 +137,25 @@ export class BookingNotifierService {
     });
   }
 
+  /** One warning shortly before the pay window closes (sent once per hold). */
+  async payDeadlineApproaching(b: Booking, deadline: Date): Promise<void> {
+    const minutesLeft = Math.max(
+      1,
+      Math.round((deadline.getTime() - Date.now()) / 60_000),
+    );
+    const eat = deadline.toLocaleString('en-GB', { timeZone: 'Africa/Addis_Ababa' });
+    const msg =
+      `Payment reminder: about ${minutesLeft} minute${minutesLeft === 1 ? '' : 's'} left ` +
+      `to pay for booking ${b.reference ?? b.id}. Deadline: ${eat} EAT — ` +
+      `unpaid reservations are released and the wagons go back on sale.`;
+    await this.notifyContact(b, msg, 'PAY REMINDER');
+    // HIGH: minutes from losing the reserved wagons — must reach SMS/email.
+    this.inApp(b, 'Payment deadline approaching', msg, {
+      type: NotificationType.INVOICE_ISSUED,
+      priority: NotificationPriority.HIGH,
+    });
+  }
+
   /**
    * Partial-capacity offer: only `offeredWagons` of the booking's `totalWagons` fit
    * this train. Paying accepts the split; letting the deadline pass keeps the

apps/edr-freight-api/src/modules/train-scheduling/booking-window.config.ts

@@ -18,7 +18,10 @@ export interface BookingWindowConfig {
   windowDurationHours: number;
   /** Max staff document-review time after the window closes. */
   docReviewMinutes: number;
+  /** Pay window for IMPORT/DOMESTIC bookings (also part of the reopen gap). */
   paymentWindowMinutes: number;
+  /** Pay window for EXPORT bookings — independent of the import value. */
+  exportPaymentWindowMinutes: number;
   /**
    * Minutes before departure the IMPORT/DOMESTIC booking window shuts. When set
    * (> 0), the effective booking cutoff is `departure − this`, capping the first

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

@@ -36,6 +36,7 @@ describe('BookingWindowService — window state machine', () => {
     windowDurationHours: 1,
     docReviewMinutes: 30,
     paymentWindowMinutes: 60,
+    exportPaymentWindowMinutes: 60,
   };
 
   const baseSchedule = (over: Partial<TrainSchedule>): TrainSchedule =>

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

@@ -20,7 +20,7 @@ import {
 import { BookingBatchService } from './booking-batch.service';
 import { BookingWindowGateway } from './booking-window.gateway';
 import { TrainSchedulingService, effectiveWindowConfig } from './train-scheduling.service';
-import { BATCH_TIMEZONE } from './booking-batch.constants';
+import { BATCH_TIMEZONE, PAYMENT_GRACE_MS } from './booking-batch.constants';
 import {
   bookingCloseCutoff,
   clampCloseToOfficeHours,
@@ -141,6 +141,13 @@ export class BookingWindowService implements OnModuleInit {
 
       await this.settleOverdueReservations();
 
+      // One pre-deadline pay reminder per hold (deduped via reminder stamp).
+      await this.bookingBatchService.sendPaymentReminders().catch((err) =>
+        this.logger.warn(
+          `Payment reminder sweep failed: ${(err as Error).message}`,
+        ),
+      );
+
       // Legacy fill (DOMESTIC / pre-migration schedules) every 5 minutes
       // (30 ticks at the 10-second cadence).
       this.tickCount += 1;
@@ -595,7 +602,10 @@ export class BookingWindowService implements OnModuleInit {
       .createQueryBuilder('b')
       .select('DISTINCT b.train_schedule_id', 'scheduleId')
       .where(`b.status IN ('SELECTED_FOR_BATCH', 'AWAITING_PAYMENT')`)
-      .andWhere('b.payment_deadline <= now()')
+      // Deadline + grace: a payment started in-window may webhook in late.
+      .andWhere('b.payment_deadline <= :graceCutoff', {
+        graceCutoff: new Date(Date.now() - PAYMENT_GRACE_MS),
+      })
       .andWhere('b.train_schedule_id IS NOT NULL')
       .getRawMany<{ scheduleId: string }>();
     for (const { scheduleId } of overdue) {

apps/edr-freight-api/src/modules/train-scheduling/dto/update-train-scheduling-global-rules.dto.ts

@@ -61,13 +61,20 @@ export class UpdateTrainSchedulingGlobalRulesDto {
   @Min(0)
   docReviewMinutes?: number;
 
-  @ApiPropertyOptional({ example: 60 })
+  @ApiPropertyOptional({ example: 60, description: 'IMPORT/DOMESTIC customer pay window, minutes' })
   @IsOptional()
   @Type(() => Number)
   @IsInt()
   @Min(1)
   paymentWindowMinutes?: number;
 
+  @ApiPropertyOptional({ example: 60, description: 'EXPORT customer pay window, minutes' })
+  @IsOptional()
+  @Type(() => Number)
+  @IsInt()
+  @Min(1)
+  exportPaymentWindowMinutes?: number;
+
   // Booking-close offsets: minutes before departure the window shuts. The UI
   // enters days/hours/minutes and converts to minutes. 0 or null clears the
   // offset (close at departure). Nullable so it can be explicitly cleared.

apps/edr-freight-api/src/modules/train-scheduling/entities/train-scheduling-global-rules.entity.ts

@@ -77,9 +77,14 @@ export class TrainSchedulingGlobalRules extends BaseEntity {
   @Column({ name: 'doc_review_minutes', type: 'int', default: 30 })
   docReviewMinutes!: number;
 
+  /** Pay window for IMPORT/DOMESTIC bookings (also feeds the window reopen delay). */
   @Column({ name: 'payment_window_minutes', type: 'int', default: 60 })
   paymentWindowMinutes!: number;
 
+  /** Pay window for EXPORT bookings — tunable independently of import. */
+  @Column({ name: 'export_payment_window_minutes', type: 'int', default: 60 })
+  exportPaymentWindowMinutes!: number;
+
   /**
    * Minutes before departure the IMPORT/DOMESTIC booking window shuts. When set,
    * the window's close (first cycle and every reopen) is capped at

apps/edr-freight-api/src/modules/train-scheduling/fleet-plan.util.ts

@@ -1,4 +1,4 @@
-import { bookingCargoTons } from './train-capacity.util';
+import { bookingCargoTons, bulkItemWagonsRequired } from './train-capacity.util';
 import type { Booking } from '../bookings/entities/booking.entity';
 import type { WagonType } from '../wagon-types/entities/wagon-type.entity';
 import {
@@ -51,8 +51,12 @@ export function sortBookingsForScheduling(bookings: Booking[]): Booking[] {
 
 export function wagonsRequiredForBooking(booking: Booking, bulkWagonCapacity?: number): number {
   if (booking.freightType === 'BULK') {
-    const weight = Number(booking.cargoTotalWeightVgm ?? 0);
     const capacity = bulkWagonCapacity && bulkWagonCapacity > 0 ? bulkWagonCapacity : 1;
+    // Break-bulk (PER_ITEM) sizes by indivisible items; `cargoTotalWeightVgm`
+    // holds the item count there, not tons.
+    const byItems = bulkItemWagonsRequired(booking, capacity);
+    if (byItems > 0) return byItems;
+    const weight = Number(booking.cargoTotalWeightVgm ?? 0);
     return Math.max(1, Math.ceil(weight / capacity));
   }
 

apps/edr-freight-api/src/modules/train-scheduling/train-capacity.util.spec.ts

@@ -1,6 +1,8 @@
 import {
+  bookingCargoTons,
   bookingGrossWeightTons,
   bookingTrainLengthMeters,
+  bulkItemWagonsRequired,
   consistUsage,
   consistViolations,
   deriveTrainCapacityFromLocomotive,
@@ -30,6 +32,66 @@ describe('train-capacity.util', () => {
       cargoTons,
     }));
 
+  describe('bulkItemWagonsRequired (break-bulk PER_ITEM)', () => {
+    // cargoTotalWeightVgm carries the ITEM COUNT for PER_ITEM cargo; the real
+    // tonnage rides in bulkTotalWeightTons.
+    const breakBulk = (quantity: number, weightTons: number) => ({
+      freightType: 'BULK',
+      cargoTotalWeightVgm: quantity,
+      bulkTotalWeightTons: weightTons,
+    });
+
+    it('floors items per wagon, then ceils wagons: 400 items / 800T on 69T wagons → 12', () => {
+      // 800/400 = 2T per item; floor(69/2) = 34 per wagon; ceil(400/34) = 12.
+      expect(bulkItemWagonsRequired(breakBulk(400, 800), 69)).toBe(12);
+    });
+
+    it('needs more wagons than raw tonnage suggests when the floor loses capacity', () => {
+      // 3 items × 40T on 69T wagons: by weight ceil(120/69) = 2, but only ONE
+      // whole 40T item fits a wagon → 3 wagons.
+      expect(bulkItemWagonsRequired(breakBulk(3, 120), 69)).toBe(3);
+    });
+
+    it('charges one wagon per item when a single item outweighs a wagon', () => {
+      expect(bulkItemWagonsRequired(breakBulk(2, 200), 69)).toBe(2);
+    });
+
+    it('returns 0 for PER_TON bulk (no stored weight) and container bookings', () => {
+      expect(
+        bulkItemWagonsRequired(
+          { freightType: 'BULK', cargoTotalWeightVgm: 500, bulkTotalWeightTons: null },
+          69,
+        ),
+      ).toBe(0);
+      expect(
+        bulkItemWagonsRequired(
+          { freightType: 'CONTAINER', cargoTotalWeightVgm: 100, bulkTotalWeightTons: 100 },
+          69,
+        ),
+      ).toBe(0);
+    });
+
+    it('returns 0 on zero/invalid capacity or amounts', () => {
+      expect(bulkItemWagonsRequired(breakBulk(400, 800), 0)).toBe(0);
+      expect(bulkItemWagonsRequired(breakBulk(0, 800), 69)).toBe(0);
+      expect(bulkItemWagonsRequired(breakBulk(400, 0), 69)).toBe(0);
+    });
+  });
+
+  describe('bookingCargoTons (break-bulk weight preference)', () => {
+    it('prefers bulkTotalWeightTons over the item-count VGM column', () => {
+      expect(
+        bookingCargoTons({ cargoTotalWeightVgm: 400, bulkTotalWeightTons: 800 }),
+      ).toBe(800);
+    });
+
+    it('falls back to cargoTotalWeightVgm when no break-bulk weight is stored', () => {
+      expect(
+        bookingCargoTons({ cargoTotalWeightVgm: 500, bulkTotalWeightTons: null }),
+      ).toBe(500);
+    });
+  });
+
   describe('deriveTrainCapacityFromLocomotive', () => {
     it('derives wagon slots from train length, not a fixed 53', () => {
       const shortLoco = deriveTrainCapacityFromLocomotive(

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

@@ -91,14 +91,21 @@ function num(value: unknown, fallback = 0): number {
  * its container lines (quantity × VGM per unit). The portal's container flow
  * stores per-line VGM and leaves `cargoTotalWeightVgm` at 0 — reading the
  * total alone made every such booking weigh only its tare.
+ *
+ * Break-bulk (PER_ITEM) bookings overload `cargoTotalWeightVgm` with the ITEM
+ * COUNT, so their real tonnage lives in `bulkTotalWeightTons` — prefer it, or
+ * a 400-item / 800T booking would "weigh" 400T against the pull limit.
  */
 export function bookingCargoTons(booking: {
   cargoTotalWeightVgm?: number | string | null;
+  bulkTotalWeightTons?: number | string | null;
   bookingContainers?: Array<{
     quantity?: number | null;
     vgmPerUnitTons?: number | string | null;
   }> | null;
 }): number {
+  const itemTons = num(booking.bulkTotalWeightTons);
+  if (itemTons > 0) return itemTons;
   const total = num(booking.cargoTotalWeightVgm);
   if (total > 0) return total;
   return (booking.bookingContainers ?? []).reduce(
@@ -107,6 +114,32 @@ export function bookingCargoTons(booking: {
   );
 }
 
+/**
+ * Wagons a break-bulk (PER_ITEM) bulk booking needs. Items are indivisible, so
+ * floor how many whole items fit one wagon, then ceil the wagon count:
+ * 400 items / 800T on 69T wagons → 2T per item → 34 items per wagon → 12 wagons.
+ * Returns 0 when the booking is not item-counted (PER_TON bulk, containers) —
+ * callers then fall back to the pooled-tonnage math.
+ */
+export function bulkItemWagonsRequired(
+  booking: {
+    freightType?: string | null;
+    cargoTotalWeightVgm?: number | string | null;
+    bulkTotalWeightTons?: number | string | null;
+  },
+  capacityTons: number,
+): number {
+  if (booking.freightType !== 'BULK' || !(capacityTons > 0)) return 0;
+  const quantity = num(booking.cargoTotalWeightVgm);
+  const totalWeightTons = num(booking.bulkTotalWeightTons);
+  if (!(quantity > 0) || !(totalWeightTons > 0)) return 0;
+  const perItemTons = totalWeightTons / quantity;
+  // ponytail: an item heavier than a whole wagon still charges 1 wagon per
+  // item; reject such bookings at creation time if the case turns real.
+  const itemsPerWagon = Math.max(1, Math.floor(capacityTons / perItemTons));
+  return Math.max(1, Math.ceil(quantity / itemsPerWagon));
+}
+
 /** Gross weight of one loaded wagon: it hauls itself plus its cargo. */
 export function grossWagonWeightTons(slot: Pick<ConsistSlot, 'tareWeightTons' | 'cargoTons'>): number {
   return num(slot.tareWeightTons) + num(slot.cargoTons);

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

@@ -213,6 +213,7 @@ export function effectiveWindowConfig(
     ruleWindowCloseHour?: number | null;
     ruleWindowDurationHours?: number | null;
     ruleReopenDelayMinutes?: number | null;
+    rulePaymentWindowMinutes?: number | null;
     ruleImportWindowLeadDays?: number | null;
     ruleExportBookingLeadHours?: number | null;
     ruleImportCloseOffsetMinutes?: number | null;
@@ -232,7 +233,14 @@ export function effectiveWindowConfig(
         ? Number(schedule.ruleWindowDurationHours)
         : liveCfg.windowDurationHours,
     docReviewMinutes: liveCfg.docReviewMinutes,
-    paymentWindowMinutes: liveCfg.paymentWindowMinutes,
+    // Pay windows read live values unless staff explicitly overrode this ONE
+    // schedule (rule_payment_window_minutes is only ever written by that
+    // override, never stamped at creation). The override wins for whichever
+    // direction the schedule runs.
+    paymentWindowMinutes:
+      schedule.rulePaymentWindowMinutes ?? liveCfg.paymentWindowMinutes,
+    exportPaymentWindowMinutes:
+      schedule.rulePaymentWindowMinutes ?? liveCfg.exportPaymentWindowMinutes,
     // The close offset is frozen per-schedule: a snapshot value of null means
     // "created with no offset" and must NOT inherit a later live offset (that
     // would retro-shrink an open train's window). Only a truly legacy row that
@@ -696,6 +704,8 @@ export class TrainSchedulingService {
     if (dto.windowDurationHours != null) row.windowDurationHours = dto.windowDurationHours;
     if (dto.docReviewMinutes != null) row.docReviewMinutes = dto.docReviewMinutes;
     if (dto.paymentWindowMinutes != null) row.paymentWindowMinutes = dto.paymentWindowMinutes;
+    if (dto.exportPaymentWindowMinutes != null)
+      row.exportPaymentWindowMinutes = dto.exportPaymentWindowMinutes;
     // Store 0 as null so "no offset" is a single canonical value.
     if (dto.importCloseOffsetMinutes !== undefined)
       row.importCloseOffsetMinutes = dto.importCloseOffsetMinutes || null;
@@ -790,7 +800,14 @@ export class TrainSchedulingService {
       // The reopen gap is doc review + payment; keep the config values unless the
       // override changes them, so the derived snapshot delay stays consistent.
       docReviewMinutes: dto.docReviewMinutes ?? liveCfg.docReviewMinutes,
-      paymentWindowMinutes: dto.paymentWindowMinutes ?? liveCfg.paymentWindowMinutes,
+      paymentWindowMinutes:
+        dto.paymentWindowMinutes ??
+        schedule.rulePaymentWindowMinutes ??
+        liveCfg.paymentWindowMinutes,
+      exportPaymentWindowMinutes:
+        dto.paymentWindowMinutes ??
+        schedule.rulePaymentWindowMinutes ??
+        liveCfg.exportPaymentWindowMinutes,
       // A per-schedule override isn't a close-offset control, so inherit the
       // offset already frozen on the schedule (null = none), or the live one for
       // legacy rows — the override must not silently drop the global offset.
@@ -853,11 +870,17 @@ export class TrainSchedulingService {
       }
     }
 
+    // The pay-window override persists only when staff actually sent it (or the
+    // schedule already had one) — windowRuleSnapshot never stamps it, so NULL
+    // keeps meaning "follow the live global value for my direction".
+    const rulePaymentWindowMinutes =
+      dto.paymentWindowMinutes ?? schedule.rulePaymentWindowMinutes ?? null;
     for (const t of targets) {
       await repo.update(t.id, {
         windowOpensAt: cap(times.windowOpensAt, t.departure),
         windowClosesAt: cap(times.windowClosesAt, t.departure),
         ...ruleFields,
+        rulePaymentWindowMinutes,
       });
     }
     this.logger.log(
@@ -1199,6 +1222,7 @@ export class TrainSchedulingService {
       windowDurationHours: num(row?.windowDurationHours, 3),
       docReviewMinutes: num(row?.docReviewMinutes, 30),
       paymentWindowMinutes: num(row?.paymentWindowMinutes, 60),
+      exportPaymentWindowMinutes: num(row?.exportPaymentWindowMinutes, 60),
       importCloseOffsetMinutes: offset(row?.importCloseOffsetMinutes),
       exportCloseOffsetMinutes: offset(row?.exportCloseOffsetMinutes),
     };
@@ -1931,8 +1955,14 @@ export class TrainSchedulingService {
       removedAt: new Date(),
     });
 
-    console.log(
-      `[NOTIFY] Booking ${bookingReference} removed from schedule ${scheduleId} by user ${userId ?? 'unknown'} — customer should be notified to reschedule or cancel.`,
+    // Ops decision, so the customer hears about it: SMS/email + inbox telling
+    // them to rebook or pick a new schedule (the removal log above is the record).
+    const removedBooking = await this.dataSource
+      .getRepository(Booking)
+      .findOne({ where: { id: bookingId }, relations: { company: true } });
+    if (removedBooking) this.bookingNotifier.removedFromTrain(removedBooking);
+    this.logger.log(
+      `Booking ${bookingReference} removed from schedule ${scheduleId} by user ${userId ?? 'unknown'} — customer notified to reschedule or cancel.`,
     );
 
     return this.getTrainScheduleById(scheduleId);
@@ -6778,7 +6808,14 @@ export class TrainSchedulingService {
         importWindowLeadDays: schedule.ruleImportWindowLeadDays ?? null,
         exportBookingLeadHours: schedule.ruleExportBookingLeadHours ?? null,
         docReviewMinutes: windowCfg.docReviewMinutes,
-        paymentWindowMinutes: windowCfg.paymentWindowMinutes,
+        // Editor prefill: this schedule's own override when staff set one,
+        // else the live global for the schedule's direction (import/export
+        // pay windows are tuned separately).
+        paymentWindowMinutes:
+          schedule.rulePaymentWindowMinutes ??
+          (schedule.direction === 'EXPORT'
+            ? windowCfg.exportPaymentWindowMinutes
+            : windowCfg.paymentWindowMinutes),
       },
       route: schedule.route
         ? { id: schedule.route.id, name: formatRouteLabel(schedule.route) }

apps/edr-freight-api/src/modules/train-scheduling/wagon-plan.util.ts

@@ -3,7 +3,7 @@ import { AllocationLoadType } from '@edr/types';
 import { Booking } from '../bookings/entities/booking.entity';
 import { containersPerWagonForSize, wagonsPerUnitForSize } from '../rule-engine/container-type.util';
 import { WagonType } from '../wagon-types/entities/wagon-type.entity';
-import { consistViolations } from './train-capacity.util';
+import { bookingCargoTons, bulkItemWagonsRequired, consistViolations } from './train-capacity.util';
 
 export const MAX_TRAIN_WEIGHT_TONS = 3500;
 export const MAX_TRAIN_LENGTH_METERS = 760;
@@ -171,11 +171,21 @@ export function buildBulkWagonPlan(
   bookings: Booking[],
   wagonType: WagonType,
 ): WagonPlanSlot[] {
-  const totalWeight = roundTons(
-    bookings.reduce((sum, b) => sum + Number(b.cargoTotalWeightVgm ?? 0), 0),
-  );
   const capacity = Number(wagonType.capacityTons);
-  const slots = Math.max(1, Math.ceil(totalWeight / capacity));
+  // Break-bulk (PER_ITEM) bookings size by indivisible items per booking —
+  // their tonnage must NOT pool with PER_TON cargo (an item can't split
+  // across wagons the way loose tonnage can).
+  const itemSlotsByBooking = bookings.map((b) => bulkItemWagonsRequired(b, capacity));
+  const itemSlots = itemSlotsByBooking.reduce((sum, n) => sum + n, 0);
+  const totalWeight = roundTons(
+    bookings.reduce(
+      (sum, b, i) =>
+        itemSlotsByBooking[i] > 0 ? sum : sum + Number(b.cargoTotalWeightVgm ?? 0),
+      0,
+    ),
+  );
+  const tonSlots = totalWeight > 0 ? Math.ceil(totalWeight / capacity) : 0;
+  const slots = Math.max(1, tonSlots + itemSlots);
 
   const basePlan: WagonPlanSlot[] = Array.from({ length: slots }, (_, index) => ({
     sequenceNo: index + 1,
@@ -293,7 +303,9 @@ function allocateBookingsToSlots(
   const remaining = bookings.map((booking) => ({
     bookingId: booking.id,
     bookingReference: booking.reference,
-    remainingWeightTons: roundTons(Number(booking.cargoTotalWeightVgm ?? 0)),
+    // bookingCargoTons, not the raw VGM column: for break-bulk (PER_ITEM)
+    // bookings that column is an item COUNT, not tons.
+    remainingWeightTons: roundTons(bookingCargoTons(booking)),
   }));
 
   let bookingIndex = 0;