add permissions and fix issues

e2e/freight/cypress/e2e/flows/export_ledger_day.cy.ts

@@ -0,0 +1,449 @@
+/**
+ * EXPORT "LEDGER DAY" (D+19) — one departure day, THREE trains of three
+ * different consist styles, ~22 bookings of every flavour (ONE_TIME +
+ * GENERAL, container + bulk, USD + ETB), every one of them driven to a known
+ * final fate — and a generated day-ledger REPORT at the end.
+ *
+ *   GRAIN  built train TRN-LEDGER-PW2 — 37 × PW2 box wagons (the physical
+ *          consist IS the cap; the only built-train schedule in the suite)
+ *          → bulk only, 37/37 = 2 590 T
+ *   BOX    loco pair, 54 × NW5 → containers only, 54/54
+ *   MIX    loco pair, 54 slots → 24w containers (NW5) + 30w bulk (CW4)
+ *
+ *   145 slots, filled FCFS in wave order (export reserves at ACCEPT):
+ *     – bulk wave fills GRAIN exactly; containers then cascade PAST the full
+ *       PW2 train onto BOX; the mixed wave lands on MIX
+ *     – ONE booking per train never pays → all three EXPIRE in one sweep
+ *     – 2 late bookings are REFUSED at create (export never queues)
+ *     – 1 refused customer rebooks GRAIN's freed 5 wagons, pays, rides
+ *
+ *   The final test classifies every booking of the day and writes
+ *   `cypress/reports/export-ledger-<day>.json`.
+ *
+ * Sequential steps of one journey — retries off.
+ */
+
+import {
+  acceptExport,
+  bookBulk,
+  bookContainers,
+  clearGeneralBooking,
+  createImportSchedule,
+  db,
+  dayLedger,
+  departureAt,
+  eatDayStr,
+  ensureExportRoute,
+  EXP_DEST,
+  EXP_ORIGIN,
+  expectWagonType,
+  forceReservationExpiry,
+  forceWindowOpen,
+  markPaid,
+  pollAllocations,
+  pollBookingStatus,
+  pollDb,
+  resetCorridorDay,
+  seedImportContract,
+  withBooking,
+  writeLedgerReport,
+  type LedgerRow,
+  type ScheduleRow,
+} from "./import-utils";
+
+const GRAIN_AT = departureAt(19);
+const BOX_AT = new Date(GRAIN_AT.getTime() + 2 * 3_600_000);
+const MIX_AT = new Date(GRAIN_AT.getTime() + 4 * 3_600_000);
+const DAY = eatDayStr(GRAIN_AT);
+
+const stamp = String(Date.now());
+const stampedRef = (suffix: string) => `CTR-IMP-${stamp}-${suffix}`;
+
+type Kind = "ONE_TIME" | "GENERAL";
+interface Line {
+  suffix: string;
+  kind: Kind;
+  currency: "ETB" | "USD";
+  freight: "CONTAINER" | "BULK";
+  twenty?: number;
+  tons?: number;
+  wagons: number;
+  wagonType: "PW2" | "NW5" | "CW4";
+  pays: boolean;
+}
+
+/** Wave 1 — fills the built 37 × PW2 GRAIN train exactly. */
+const GRAIN_LINES: Line[] = [
+  { suffix: "LG1", kind: "ONE_TIME", currency: "USD", freight: "BULK", tons: 700, wagons: 10, wagonType: "PW2", pays: true },
+  { suffix: "LG2", kind: "GENERAL", currency: "ETB", freight: "BULK", tons: 560, wagons: 8, wagonType: "PW2", pays: true },
+  { suffix: "LG3", kind: "ONE_TIME", currency: "ETB", freight: "BULK", tons: 560, wagons: 8, wagonType: "PW2", pays: true },
+  { suffix: "LG4", kind: "ONE_TIME", currency: "USD", freight: "BULK", tons: 420, wagons: 6, wagonType: "PW2", pays: true },
+  { suffix: "LG5", kind: "ONE_TIME", currency: "ETB", freight: "BULK", tons: 350, wagons: 5, wagonType: "PW2", pays: false },
+];
+
+/** Wave 2 — 54 × NW5 on the pure container train. */
+const BOX_LINES: Line[] = [
+  { suffix: "LB1", kind: "ONE_TIME", currency: "USD", freight: "CONTAINER", twenty: 40, wagons: 20, wagonType: "NW5", pays: true },
+  { suffix: "LB2", kind: "ONE_TIME", currency: "ETB", freight: "CONTAINER", twenty: 24, wagons: 12, wagonType: "NW5", pays: true },
+  { suffix: "LB3", kind: "GENERAL", currency: "USD", freight: "CONTAINER", twenty: 16, wagons: 8, wagonType: "NW5", pays: true },
+  { suffix: "LB4", kind: "ONE_TIME", currency: "ETB", freight: "CONTAINER", twenty: 12, wagons: 6, wagonType: "NW5", pays: true },
+  { suffix: "LB5", kind: "ONE_TIME", currency: "USD", freight: "CONTAINER", twenty: 8, wagons: 4, wagonType: "NW5", pays: true },
+  { suffix: "LB6", kind: "GENERAL", currency: "ETB", freight: "CONTAINER", twenty: 4, wagons: 2, wagonType: "NW5", pays: true },
+  { suffix: "LB7", kind: "ONE_TIME", currency: "USD", freight: "CONTAINER", twenty: 2, wagons: 1, wagonType: "NW5", pays: true },
+  { suffix: "LB8", kind: "ONE_TIME", currency: "ETB", freight: "CONTAINER", twenty: 2, wagons: 1, wagonType: "NW5", pays: false },
+];
+
+/** Wave 3 — 24w containers + 30w bulk share the MIX train. */
+const MIX_LINES: Line[] = [
+  { suffix: "LM1", kind: "ONE_TIME", currency: "USD", freight: "CONTAINER", twenty: 24, wagons: 12, wagonType: "NW5", pays: true },
+  { suffix: "LM2", kind: "GENERAL", currency: "ETB", freight: "CONTAINER", twenty: 16, wagons: 8, wagonType: "NW5", pays: true },
+  { suffix: "LM3", kind: "ONE_TIME", currency: "ETB", freight: "CONTAINER", twenty: 8, wagons: 4, wagonType: "NW5", pays: true },
+  { suffix: "LM4", kind: "ONE_TIME", currency: "USD", freight: "BULK", tons: 700, wagons: 10, wagonType: "CW4", pays: true },
+  { suffix: "LM5", kind: "GENERAL", currency: "ETB", freight: "BULK", tons: 700, wagons: 10, wagonType: "CW4", pays: true },
+  { suffix: "LM6", kind: "ONE_TIME", currency: "ETB", freight: "BULK", tons: 700, wagons: 10, wagonType: "CW4", pays: false },
+];
+
+const ALL_LINES = [...GRAIN_LINES, ...BOX_LINES, ...MIX_LINES];
+const PAID = ALL_LINES.filter((l) => l.pays);
+const UNPAID = ALL_LINES.filter((l) => !l.pays);
+/** Refused at create once all 145 slots are held; LR2 later redeems itself. */
+const REJECTED = ["LR1", "LR2"];
+
+let isoSeed = 20_000;
+
+/** Book one line and take it to its FCFS reservation. */
+function bookAndAccept(line: Line) {
+  if (line.freight === "CONTAINER") {
+    bookContainers({
+      suffix: line.suffix,
+      runStamp: stamp,
+      isoSeed,
+      twenty: line.twenty,
+      scheduledDate: DAY,
+    });
+    isoSeed += (line.twenty ?? 0) + 5;
+  } else {
+    // PW2 bulk rides the built GRAIN consist → GRAINS cargo; CW4 bulk rides the
+    // loco-pair MIX train → WHEAT. One wagon length per cargo keeps the
+    // loco-pair length budget deterministic (see seed 5b3).
+    bookBulk({
+      suffix: line.suffix,
+      tons: line.tons!,
+      scheduledDate: DAY,
+      cargoCode: line.wagonType === "PW2" ? "E2E_IMP_GRAINS" : "E2E_IMP_WHEAT",
+    });
+  }
+  if (line.kind === "GENERAL") clearGeneralBooking(line.suffix, DAY, "export");
+  else acceptExport(line.suffix);
+}
+
+function seedLine(line: Line) {
+  seedImportContract({
+    suffix: line.suffix,
+    reference: stampedRef(line.suffix),
+    kind: line.kind,
+    currency: line.currency,
+    freight: line.freight,
+    direction: "EXPORT",
+    originCode: EXP_ORIGIN,
+    destCode: EXP_DEST,
+  });
+}
+
+interface DaySchedule {
+  id: string;
+  reference: string;
+  max_wagons: number;
+  booking_window_status: string;
+}
+
+function daySchedule(at: Date) {
+  return db<DaySchedule>(
+    `SELECT ts.id, ts.reference, ts.max_wagons, ts.booking_window_status
+     FROM freight.train_schedules ts
+     JOIN freight.yards o ON o.id = ts.origin_station_id AND o.code = $1
+     JOIN freight.yards d ON d.id = ts.destination_station_id AND d.code = $2
+     WHERE ts.deleted_at IS NULL
+       AND abs(extract(epoch FROM (ts.scheduled_departure_date - $3::timestamptz))) < 3600
+     ORDER BY ts.created_at DESC LIMIT 1`,
+    [EXP_ORIGIN, EXP_DEST, at.toISOString()],
+  ).then(({ rows }) => {
+    expect(rows, `schedule departing ${at.toISOString()}`).to.have.length(1);
+    return cy.wrap(rows[0], { log: false });
+  });
+}
+
+describe("export ledger day: three trains, every fate, one report", { retries: 0 }, () => {
+  before(() => {
+    cy.task("db:seedFile", "seed-import-corridor.sql");
+    ALL_LINES.forEach(seedLine);
+    seedLine({
+      suffix: "LR1",
+      kind: "ONE_TIME",
+      currency: "USD",
+      freight: "CONTAINER",
+      twenty: 2,
+      wagons: 1,
+      wagonType: "NW5",
+      pays: false,
+    });
+    seedLine({
+      suffix: "LR2",
+      kind: "ONE_TIME",
+      currency: "ETB",
+      freight: "BULK",
+      tons: 350,
+      wagons: 5,
+      wagonType: "PW2",
+      pays: false,
+    });
+  });
+
+  it("operations schedules THREE trains for one day: 37×PW2 built GRAIN, 54×NW5 BOX, 54 MIX", () => {
+    ensureExportRoute();
+    resetCorridorDay(GRAIN_AT, EXP_DEST, EXP_ORIGIN);
+
+    // The built train's coupled consist IS its capacity — 37, not the
+    // loco-length-derived 54 every other schedule in the suite gets.
+    createImportSchedule({
+      departure: GRAIN_AT,
+      trainCode: "TRN-LEDGER-PW2",
+      maxWagons: 37,
+      kind: "bulk",
+      originCode: EXP_ORIGIN,
+      destCode: EXP_DEST,
+    });
+    createImportSchedule({
+      departure: BOX_AT,
+      locoPair: ["LOCO-LED-3", "LOCO-LED-4"],
+      originCode: EXP_ORIGIN,
+      destCode: EXP_DEST,
+    });
+    createImportSchedule({
+      departure: MIX_AT,
+      locoPair: ["LOCO-LED-5", "LOCO-LED-6"],
+      originCode: EXP_ORIGIN,
+      destCode: EXP_DEST,
+    });
+
+    [GRAIN_AT, BOX_AT, MIX_AT].forEach((at) =>
+      daySchedule(at).then((s) => forceWindowOpen(s.id, 120)),
+    );
+    daySchedule(GRAIN_AT).then((s) =>
+      expect(s.max_wagons, "built consist = 37 PW2").to.eq(37),
+    );
+    daySchedule(BOX_AT).then((s) => expect(s.max_wagons, "BOX = 54").to.eq(54));
+    daySchedule(MIX_AT).then((s) => expect(s.max_wagons, "MIX = 54").to.eq(54));
+  });
+
+  it("wave 1 — five bulk bookings fill the built PW2 train exactly (37/37 reserved)", () => {
+    GRAIN_LINES.forEach(bookAndAccept);
+    daySchedule(GRAIN_AT).then((s) => {
+      GRAIN_LINES.forEach((l) =>
+        withBooking(l.suffix, (b) =>
+          expect(b.train_schedule_id, `${l.suffix} on GRAIN`).to.eq(s.id),
+        ),
+      );
+    });
+  });
+
+  it("ONE_TIME is single-slot; GENERAL draws down again but is stopped at the FCFS accept", () => {
+    // A second booking on the live ONE_TIME contract LG1 is refused outright…
+    bookBulk({
+      suffix: "LG1",
+      tons: 70,
+      scheduledDate: DAY,
+      cargoCode: "E2E_IMP_GRAINS",
+      expectFailure: "already has an active booking",
+    });
+
+    // …while the GENERAL neighbour LG2 draws again freely. Engine truth: a
+    // GENERAL booking clears PER BOOKING, so it SKIPS the create-time
+    // window/space gate that refused the ONE_TIME latecomers above — a full
+    // day stops it only later, when staff try to accept it onto a train.
+    bookBulk({ suffix: "LG2", tons: 70, scheduledDate: DAY, cargoCode: "E2E_IMP_GRAINS" });
+    withBooking("LG2", (probe) => {
+      expect(probe.status, "GENERAL drawdown accepted at create").to.eq(
+        "AWAITING_DOCUMENTS",
+      );
+      // Dispose of the probe so the day's ledger keeps exactly one live
+      // booking per contract (every lookup skips soft-deleted rows).
+      db(`UPDATE freight.bookings SET deleted_at = now() WHERE id = $1`, [probe.id]);
+    });
+    withBooking("LG2", (b) =>
+      expect(b.status, "LG2's real reservation is untouched").to.be.oneOf([
+        "SELECTED_FOR_BATCH",
+        "AWAITING_PAYMENT",
+      ]),
+    );
+  });
+
+  it("wave 2 — containers cascade PAST the full PW2 train onto BOX (54/54 reserved)", () => {
+    BOX_LINES.forEach(bookAndAccept);
+    daySchedule(BOX_AT).then((s) => {
+      BOX_LINES.forEach((l) =>
+        withBooking(l.suffix, (b) =>
+          expect(b.train_schedule_id, `${l.suffix} on BOX`).to.eq(s.id),
+        ),
+      );
+    });
+  });
+
+  it("wave 3 — containers and bulk share MIX (24w + 30w = 54/54 reserved)", () => {
+    MIX_LINES.forEach(bookAndAccept);
+    daySchedule(MIX_AT).then((s) => {
+      MIX_LINES.forEach((l) =>
+        withBooking(l.suffix, (b) =>
+          expect(b.train_schedule_id, `${l.suffix} on MIX`).to.eq(s.id),
+        ),
+      );
+    });
+  });
+
+  it("every reservation's pay deadline is clamped to its OWN train's window close", () => {
+    const check = (at: Date, lines: Line[]) =>
+      daySchedule(at).then((s) => {
+        db<{ window_closes_at: string }>(
+          `SELECT window_closes_at FROM freight.train_schedules WHERE id = $1`,
+          [s.id],
+        ).then(({ rows }) => {
+          lines.forEach((l) =>
+            withBooking(l.suffix, (b) => {
+              expect(
+                new Date(b.payment_deadline!).getTime(),
+                `${l.suffix} clamped`,
+              ).to.be.at.most(new Date(rows[0].window_closes_at).getTime());
+            }),
+          );
+        });
+      });
+    check(GRAIN_AT, GRAIN_LINES);
+    check(BOX_AT, BOX_LINES);
+    check(MIX_AT, MIX_LINES);
+  });
+
+  it("all 145 slots are held — two late bookings are REFUSED (export never queues)", () => {
+    bookContainers({
+      suffix: "LR1",
+      runStamp: stamp,
+      isoSeed: 29_000,
+      twenty: 2,
+      scheduledDate: DAY,
+      expectFailure: /space|window/i,
+    });
+    bookBulk({
+      suffix: "LR2",
+      tons: 350,
+      scheduledDate: DAY,
+      cargoCode: "E2E_IMP_GRAINS",
+      expectFailure: /space|window/i,
+    });
+  });
+
+  it("16 of the 19 pay — typed allocation on all three trains", () => {
+    PAID.forEach((l) => {
+      markPaid(l.suffix);
+      pollAllocations(l.suffix, l.wagons);
+      expectWagonType(l.suffix, l.wagonType, l.wagons);
+    });
+  });
+
+  it("the three unpaid — one per train — expire in one sweep", () => {
+    UNPAID.forEach((l) => forceReservationExpiry(l.suffix));
+    UNPAID.forEach((l) => pollBookingStatus(l.suffix, "EXPIRED"));
+  });
+
+  it("redemption: the refused bulk customer takes GRAIN's freed 5 PW2 wagons and pays", () => {
+    bookBulk({ suffix: "LR2", tons: 350, scheduledDate: DAY, cargoCode: "E2E_IMP_GRAINS" });
+    acceptExport("LR2");
+    markPaid("LR2");
+    pollAllocations("LR2", 5);
+    expectWagonType("LR2", "PW2", 5);
+    daySchedule(GRAIN_AT).then((s) =>
+      withBooking("LR2", (b) =>
+        expect(b.train_schedule_id, "LR2 rides GRAIN").to.eq(s.id),
+      ),
+    );
+  });
+
+  it("per-train totals: 37 PW2 + 54 NW5 + 54 mixed, links match bookings", () => {
+    const totals: Array<[Date, string, number, number]> = [
+      // [departure, label, wagons, bookings]
+      [GRAIN_AT, "GRAIN", 37, 5], // 4 paid + the redeemed LR2
+      [BOX_AT, "BOX", 53, 7], // one booking expired: 54 − 1
+      [MIX_AT, "MIX", 44, 5], // one 10w booking expired: 54 − 10
+    ];
+    totals.forEach(([at, label, wagons, bookings]) => {
+      daySchedule(at).then((s) => {
+        pollDb<{ n: string }>(
+          `${label} carries ${bookings} bookings`,
+          `SELECT count(*) AS n FROM freight.train_schedule_bookings
+           WHERE train_schedule_id = $1 AND deleted_at IS NULL`,
+          [s.id],
+          (row) => Number(row?.n) === bookings,
+          15,
+        );
+        db<{ n: string }>(
+          `SELECT count(DISTINCT wba.train_set_wagon_id) AS n
+           FROM freight.wagon_booking_allocations wba
+           JOIN freight.train_schedule_bookings tsb
+             ON tsb.booking_id = wba.booking_id AND tsb.train_schedule_id = $1
+           WHERE wba.deleted_at IS NULL AND tsb.deleted_at IS NULL`,
+          [s.id],
+        ).then(({ rows }) =>
+          expect(Number(rows[0].n), `${label} wagons`).to.eq(wagons),
+        );
+      });
+    });
+  });
+
+  it("THE LEDGER — classify every booking of the day and write the report", () => {
+    dayLedger(stamp, { rejected: REJECTED, redeemed: ["LR2"] }).then((rows) => {
+      const ledger = rows as LedgerRow[];
+      const byFate = (fate: string) => ledger.filter((r) => r.fate === fate);
+
+      // Class counts: 16 riding + 1 redeemed + 3 expired + 2 refusals.
+      expect(byFate("PAID_RIDING").length, "paid & riding").to.eq(PAID.length);
+      expect(byFate("REBOOKED_PAID").length, "refused then rebooked").to.eq(1);
+      expect(byFate("EXPIRED_UNPAID").length, "reserved but never paid").to.eq(
+        UNPAID.length,
+      );
+      expect(byFate("REJECTED_NO_SPACE").length, "refused at create").to.eq(
+        REJECTED.length,
+      );
+      expect(byFate("RESERVED_UNPAID").length, "nothing left hanging").to.eq(0);
+
+      // Every engineered line landed in its expected class.
+      PAID.forEach((l) => {
+        const row = ledger.find((r) => r.suffix === l.suffix && r.fate === "PAID_RIDING");
+        expect(row, `${l.suffix} rides`).to.exist;
+        expect(row!.kind, `${l.suffix} kind`).to.eq(l.kind);
+        expect(row!.currency, `${l.suffix} currency`).to.eq(l.currency);
+        expect(row!.wagon_type, `${l.suffix} wagon type`).to.eq(l.wagonType);
+        expect(row!.wagons, `${l.suffix} wagons`).to.eq(l.wagons);
+      });
+      UNPAID.forEach((l) =>
+        expect(
+          ledger.find((r) => r.suffix === l.suffix && r.fate === "EXPIRED_UNPAID"),
+          `${l.suffix} expired`,
+        ).to.exist,
+      );
+      // LR2 appears TWICE: the refusal and the redemption.
+      expect(
+        ledger.filter((r) => r.suffix === "LR2").length,
+        "LR2 refused then rebooked",
+      ).to.eq(2);
+
+      // Currency split of the riding cargo (USD vs ETB invoices on one day).
+      const riding = [...byFate("PAID_RIDING"), ...byFate("REBOOKED_PAID")];
+      expect(riding.filter((r) => r.currency === "USD").length, "USD riders").to.be.greaterThan(0);
+      expect(riding.filter((r) => r.currency === "ETB").length, "ETB riders").to.be.greaterThan(0);
+      // Every riding booking sits on a train, every expired one does not ride.
+      riding.forEach((r) => expect(r.train, `${r.suffix} has a train`).to.be.a("string"));
+
+      writeLedgerReport(`export-ledger-${DAY}`, ledger);
+    });
+  });
+});
+
+export {};

e2e/freight/cypress/e2e/flows/import-utils.ts

@@ -271,7 +271,7 @@ export function dbBooking(suffix: string) {
             b.is_split, b.contract_id
      FROM freight.bookings b
      JOIN freight.contracts ct ON ct.id = b.contract_id
-     WHERE ct.reference LIKE 'CTR-IMP-%-' || $1
+     WHERE ct.reference LIKE 'CTR-IMP-%-' || $1 AND b.deleted_at IS NULL
      ORDER BY b.created_at DESC LIMIT 1`,
     [suffix],
   );
@@ -297,7 +297,7 @@ export function pollBookingStatus(suffix: string, status: string | string[], att
     `${suffix} → ${want.join("|")}`,
     `SELECT b.status FROM freight.bookings b
      JOIN freight.contracts ct ON ct.id = b.contract_id
-     WHERE ct.reference LIKE 'CTR-IMP-%-' || $1
+     WHERE ct.reference LIKE 'CTR-IMP-%-' || $1 AND b.deleted_at IS NULL
      ORDER BY b.created_at DESC LIMIT 1`,
     [suffix],
     (row) => !!row && want.includes(row.status as string),
@@ -389,15 +389,17 @@ export function bookBulk(opts: {
   suffix: string;
   tons: number;
   scheduledDate?: string; // omit for DOMESTIC (intercity)
+  /** Cargo code — WHEAT rides CW4, GRAINS rides PW2. Defaults to wheat. */
+  cargoCode?: "E2E_IMP_WHEAT" | "E2E_IMP_GRAINS";
   expectFailure?: string | RegExp;
 }) {
   db<{ id: string; customs_clearing_enabled: boolean; cargo_type_id: string }>(
     `SELECT ct.id, ct.customs_clearing_enabled,
-            (SELECT t.id FROM freight.cargo_types t WHERE t.code = 'E2E_IMP_WHEAT') AS cargo_type_id
+            (SELECT t.id FROM freight.cargo_types t WHERE t.code = $2) AS cargo_type_id
      FROM freight.contracts ct
      WHERE ct.reference LIKE 'CTR-IMP-%-' || $1
      ORDER BY ct.created_at DESC LIMIT 1`,
-    [opts.suffix],
+    [opts.suffix, opts.cargoCode ?? "E2E_IMP_WHEAT"],
   ).then(({ rows }) => {
     expect(rows, `seeded contract *-${opts.suffix}`).to.have.length(1);
     const actor = rows[0].customs_clearing_enabled ? superAdmin : customer;
@@ -433,7 +435,12 @@ export function bookBulk(opts: {
  * seed configures no required documents, so one ad-hoc doc satisfies the
  * 100%-approved gate.
  */
-export function clearGeneralBooking(suffix: string, scheduledDate: string) {
+export function clearGeneralBooking(
+  suffix: string,
+  scheduledDate: string,
+  /** EXPORT ends at acceptExport (FCFS reserves on accept), not the pool. */
+  mode: "import" | "export" = "import",
+) {
   withBooking(suffix, (b) => {
     expect(b.status, `${suffix} starts in the clearance gate`).to.eq("AWAITING_DOCUMENTS");
     glUpload(`/api/bookings/${b.id}/clearance/documents`, {}, "custom_e2e");
@@ -450,7 +457,8 @@ export function clearGeneralBooking(suffix: string, scheduledDate: string) {
       .its("status")
       .should("be.oneOf", [200, 201]);
   });
-  acceptOperation(suffix);
+  if (mode === "export") acceptExport(suffix);
+  else acceptOperation(suffix);
 }
 
 /** Ops accepts the operation request → FULLY_EXECUTED (enters the day pool). */
@@ -473,6 +481,28 @@ export function setPriority(suffix: string, order: number) {
   );
 }
 
+/**
+ * Customs bookings pay their clearance service fee ON the booking invoice —
+ * there is no separate prepaid clearance invoice. Asserts the booking's
+ * invoice carries a CUSTOMS_CLEARANCE line.
+ */
+export function expectClearanceOnBookingInvoice(suffix: string) {
+  pollDb<{ n: string }>(
+    `${suffix} clearance fee on booking invoice`,
+    `SELECT COUNT(*)::text AS n
+     FROM freight.invoice_lines l
+     JOIN freight.invoices i ON i.id = l.invoice_id
+     JOIN freight.bookings b ON b.id::text = i.source_id::text
+     JOIN freight.contracts ct ON ct.id = b.contract_id
+     WHERE ct.reference LIKE 'CTR-IMP-%-' || $1 AND b.deleted_at IS NULL
+       AND i.source = 'booking'
+       AND l.charge_type LIKE 'CUSTOMS_CLEARANCE%'`,
+    [suffix],
+    (row) => Number(row?.n ?? 0) > 0,
+    10,
+  );
+}
+
 /** Staff force-pay; polls PAID + SCHEDULED. */
 export function markPaid(suffix: string) {
   withBooking(suffix, (b) => {
@@ -484,7 +514,7 @@ export function markPaid(suffix: string) {
     `${suffix} PAID+SCHEDULED`,
     `SELECT b.status, b.scheduling_status FROM freight.bookings b
      JOIN freight.contracts ct ON ct.id = b.contract_id
-     WHERE ct.reference LIKE 'CTR-IMP-%-' || $1
+     WHERE ct.reference LIKE 'CTR-IMP-%-' || $1 AND b.deleted_at IS NULL
      ORDER BY b.created_at DESC LIMIT 1`,
     [suffix],
     (row) => row?.status === "PAID" && row?.scheduling_status === "SCHEDULED",
@@ -557,7 +587,7 @@ export function settleViaGateway(suffix: string) {
     `${suffix} PAID via gateway`,
     `SELECT b.status FROM freight.bookings b
      JOIN freight.contracts ct ON ct.id = b.contract_id
-     WHERE ct.reference LIKE 'CTR-IMP-%-' || $1
+     WHERE ct.reference LIKE 'CTR-IMP-%-' || $1 AND b.deleted_at IS NULL
      ORDER BY b.created_at DESC LIMIT 1`,
     [suffix],
     (row) => row?.status === "PAID",
@@ -808,7 +838,13 @@ export function withExportSchedule(departure: Date, fn: (s: ScheduleRow) => void
  */
 export function createImportSchedule(opts: {
   departure: Date;
-  locoPair: [string, string];
+  /** Loco-pair mode — capacity comes from maxWagonsPerTrain. */
+  locoPair?: [string, string];
+  /**
+   * Built-train mode — the Train-Builder consist IS the capacity (its coupled
+   * wagon count), immune to the loco-length slot recompute.
+   */
+  trainCode?: string;
   maxWagons?: number;
   kind?: "container" | "bulk";
   originCode?: string;
@@ -816,16 +852,32 @@ export function createImportSchedule(opts: {
 }) {
   const originCode = opts.originCode ?? ORIGIN;
   const destCode = opts.destCode ?? DEST;
+  const endpoint = `/api/train-scheduling/${opts.kind ?? "container"}/schedules`;
   dbSchedule(opts.departure, destCode, originCode).then(({ rows }) => {
     if (rows.length > 0) return;
     dbRouteId(originCode, destCode).then(({ rows: routes }) => {
       expect(routes, "corridor route").to.have.length(1);
+      if (opts.trainCode) {
+        db<{ id: string }>(`SELECT id FROM freight.trains WHERE code = $1`, [
+          opts.trainCode,
+        ]).then(({ rows: trains }) => {
+          expect(trains, `built train ${opts.trainCode}`).to.have.length(1);
+          apiPost(opsStaff, endpoint, {
+            routeId: routes[0].id,
+            scheduleDate: opts.departure.toISOString(),
+            trainId: trains[0].id,
+          })
+            .its("status")
+            .should("be.oneOf", [200, 201]);
+        });
+        return;
+      }
       db<{ id: string }>(
         `SELECT id FROM freight.locomotives WHERE code = ANY($1::text[]) ORDER BY code`,
-        [opts.locoPair],
+        [opts.locoPair ?? []],
       ).then(({ rows: locos }) => {
-        expect(locos, `locomotives ${opts.locoPair.join(",")}`).to.have.length(2);
-        apiPost(opsStaff, `/api/train-scheduling/${opts.kind ?? "container"}/schedules`, {
+        expect(locos, `locomotives ${(opts.locoPair ?? []).join(",")}`).to.have.length(2);
+        apiPost(opsStaff, endpoint, {
           routeId: routes[0].id,
           scheduleDate: opts.departure.toISOString(),
           locomotiveIds: locos.map((l) => l.id),
@@ -838,7 +890,111 @@ export function createImportSchedule(opts: {
   });
   dbSchedule(opts.departure, destCode, originCode).then(({ rows }) => {
     expect(rows, "created schedule").to.have.length(1);
-    expect(rows[0].max_wagons, "54-wagon consist").to.eq(opts.maxWagons ?? 54);
+    expect(rows[0].max_wagons, "consist size").to.eq(opts.maxWagons ?? 54);
+  });
+}
+
+// ---------------------------------------------------------------------------
+// day ledger — who booked, who rides, who expired, who was refused
+// ---------------------------------------------------------------------------
+
+export interface LedgerRow {
+  suffix: string;
+  reference: string;
+  kind: string;
+  freight: string;
+  currency: string;
+  status: string;
+  wagons: number;
+  wagon_type: string | null;
+  train: string | null;
+  fate: string;
+}
+
+/**
+ * Classify every booking made under this run's stamped contracts into its
+ * final fate. Bookings the engine REFUSED at create never exist as rows —
+ * the caller passes those suffixes in (they are only observable as 4xx at
+ * request time), plus any suffix that was refused and later rebooked.
+ */
+export function dayLedger(
+  runStamp: string,
+  opts: { rejected?: string[]; redeemed?: string[] } = {},
+) {
+  return db<LedgerRow>(
+    `SELECT split_part(ct.reference, '-', 4) AS suffix,
+            b.reference,
+            ct.contract_kind AS kind,
+            b.freight_type   AS freight,
+            b.payment_currency AS currency,
+            b.status,
+            COALESCE(a.wagons, 0)::int AS wagons,
+            a.wagon_type,
+            ts.reference AS train,
+            CASE
+              WHEN b.status IN ('PAID','IN_TRANSIT','ARRIVED','COMPLETED') THEN 'PAID_RIDING'
+              WHEN b.status = 'EXPIRED' THEN 'EXPIRED_UNPAID'
+              WHEN b.status IN ('SELECTED_FOR_BATCH','AWAITING_PAYMENT') THEN 'RESERVED_UNPAID'
+              ELSE 'PENDING_' || b.status
+            END AS fate
+     FROM freight.bookings b
+     JOIN freight.contracts ct ON ct.id = b.contract_id
+     LEFT JOIN freight.train_schedules ts ON ts.id = b.train_schedule_id
+     LEFT JOIN LATERAL (
+       SELECT count(*)::int AS wagons, max(wt.code) AS wagon_type
+       FROM freight.wagon_booking_allocations wba
+       JOIN freight.train_set_wagons tsw ON tsw.id = wba.train_set_wagon_id
+       JOIN freight.wagon_types wt ON wt.id = tsw.wagon_type_id
+       WHERE wba.booking_id = b.id AND wba.deleted_at IS NULL
+     ) a ON true
+     WHERE ct.reference LIKE 'CTR-IMP-' || $1 || '-%' AND b.deleted_at IS NULL
+     ORDER BY b.created_at`,
+    [runStamp],
+  ).then(({ rows }) => {
+    const rejected = (opts.rejected ?? []).map((suffix) => ({
+      suffix,
+      reference: "—",
+      kind: "—",
+      freight: "—",
+      currency: "—",
+      status: "NOT_CREATED",
+      wagons: 0,
+      wagon_type: null,
+      train: null,
+      fate: "REJECTED_NO_SPACE",
+    })) as LedgerRow[];
+    // A refused customer who later rebooked shows BOTH lines: the refusal
+    // above and the live booking here, re-labelled.
+    const ledger = [
+      ...rows.map((r) =>
+        (opts.redeemed ?? []).includes(r.suffix) && r.fate === "PAID_RIDING"
+          ? { ...r, fate: "REBOOKED_PAID" }
+          : r,
+      ),
+      ...rejected,
+    ];
+    return cy.wrap(ledger, { log: false });
+  });
+}
+
+/** Print the ledger as a table into the Cypress log and write a JSON artifact. */
+export function writeLedgerReport(name: string, ledger: LedgerRow[]) {
+  const counts = ledger.reduce<Record<string, number>>((acc, r) => {
+    acc[r.fate] = (acc[r.fate] ?? 0) + 1;
+    return acc;
+  }, {});
+  cy.log(`**LEDGER ${name}** — ${JSON.stringify(counts)}`);
+  ledger.forEach((r) =>
+    cy.log(
+      `${r.suffix.padEnd(8)} ${r.kind.padEnd(9)} ${r.freight.padEnd(9)} ` +
+        `${r.currency.padEnd(4)} ${String(r.wagons).padStart(2)}w ` +
+        `${(r.wagon_type ?? "-").padEnd(4)} ${(r.train ?? "-").padEnd(14)} ${r.fate}`,
+    ),
+  );
+  cy.writeFile(`cypress/reports/${name}.json`, {
+    generatedAt: new Date().toISOString(),
+    counts,
+    bookings: ledger,
   });
 }
 

e2e/freight/cypress/e2e/flows/import_full_train.cy.ts

@@ -43,6 +43,7 @@ import {
   withBooking,
   withSchedule,
   type ScheduleRow,
+  expectClearanceOnBookingInvoice,
 } from "./import-utils";
 
 const DEPARTURE = departureAt(4);
@@ -147,6 +148,9 @@ describe("import: six bookings fill the 54-wagon corridor train", { retries: 0 }
 
   it("all six pay — allocated onto the train, 54/54 wagons, window FULL and schedule finalized", () => {
     BOOKINGS.forEach((b) => {
+      // Customs bookings carry their clearance service fee ON the booking
+      // invoice (no prepaid clearance invoice) — assert before settling.
+      if (b.customs) expectClearanceOnBookingInvoice(b.suffix);
       markPaid(b.suffix);
       pollAllocations(b.suffix, b.wagons);
     });

e2e/freight/cypress/fixtures/seed-import-corridor.sql

@@ -26,6 +26,24 @@
 -- missing production migration.
 DROP INDEX IF EXISTS freight.uq_one_active_booking_per_one_time_contract;
 
+-- 0b. Schema drift guard: CUSTOMS_CLEARANCE / WITH_RETURN rates became
+-- route-scoped in migrations 2820 + 2830, but an e2e image built before them
+-- still carries the old CK_rates_yard_scope (yards allowed only for ALWAYS
+-- BULK/CONTAINER/INTERCITY rows). Re-state the post-2830 shape — a no-op on an
+-- up-to-date image, and what lets the customs-clearance lane rates below load.
+ALTER TABLE freight.rates DROP CONSTRAINT IF EXISTS "CK_rates_yard_scope";
+ALTER TABLE freight.rates
+  ADD CONSTRAINT "CK_rates_yard_scope" CHECK (
+    deleted_at IS NOT NULL
+    OR status = 'SUPERSEDED'
+    OR CASE
+         WHEN ("trigger" = 'ALWAYS' AND applies_to IN ('BULK', 'CONTAINER', 'INTERCITY'))
+           OR "trigger" IN ('CUSTOMS_CLEARANCE', 'WITH_RETURN')
+           THEN origin_yard_id IS NOT NULL AND destination_yard_id IS NOT NULL
+         ELSE origin_yard_id IS NULL AND destination_yard_id IS NULL
+       END
+  );
+
 -- 1a. Extra Ethiopian mid-corridor yard.
 INSERT INTO freight.yards (id, code, label, country, is_active, display_order)
 SELECT gen_random_uuid(), 'E2E_AWASH', 'E2E Awash Yard', 'Ethiopia', true, 50
@@ -124,7 +142,7 @@ FROM (
   JOIN freight.yards y ON y.id = w2.current_yard_id AND y.code = 'DJIB_PORT'
   WHERE w2.train_id IS NULL AND w2.deleted_at IS NULL
   ORDER BY w2.wagon_number ASC
-  LIMIT 120
+  LIMIT 200
 ) pick
 WHERE w.id = pick.id;
 
@@ -158,22 +176,35 @@ WHERE NOT EXISTS (SELECT 1 FROM freight.locomotives l WHERE l.code = v.code);
 INSERT INTO freight.wagons (id, wagon_number, wagon_type_id, current_yard_id)
 SELECT gen_random_uuid(), 'ECW' || lpad(g::text, 4, '0'), wt.id,
        (SELECT id FROM freight.yards
-        WHERE code = CASE WHEN g <= 80 THEN 'KALITY' ELSE 'DIRE_DAWA' END)
-FROM generate_series(1, 100) AS g
+        WHERE code = CASE WHEN g <= 120 THEN 'KALITY' ELSE 'DIRE_DAWA' END)
+FROM generate_series(1, 160) AS g
 JOIN freight.wagon_types wt ON wt.code = 'CW4'
 WHERE NOT EXISTS (
   SELECT 1 FROM freight.wagons w WHERE w.wagon_number = 'ECW' || lpad(g::text, 4, '0')
 );
 
--- 5b3. Ledger-day rolling stock: wheat may also ride PW2 box wagons, and the
--- GRAIN train is a BUILT Train-Builder consist — 37 PW2 wagons coupled at
--- KALITY behind two dedicated locos. A built train's physical wagon count IS
--- its schedule capacity (37), immune to the loco-length slot recompute.
+-- 5b3. Ledger-day rolling stock: GRAINS ride PW2 box wagons on a BUILT
+-- Train-Builder consist — 37 PW2 wagons coupled at KALITY behind two dedicated
+-- locos. A built train's physical wagon count IS its schedule capacity (37),
+-- immune to the loco-length slot recompute.
+--
+-- GRAINS gets PW2 (17.066 m), WHEAT gets CW4 (13.976 m) below — kept on SEPARATE
+-- cargo types on purpose. `dimsFor`/`needFor` size a bulk booking off its cargo
+-- type's FIRST wagon type, so a cargo carrying two wagon-lengths sizes
+-- inconsistently on loco-pair export trains (one booking as CW4, another as PW2)
+-- and the length budget never matches the 54-wagon CW4 board the export specs
+-- expect. One cargo → one wagon length keeps that math deterministic.
+DELETE FROM freight.cargo_type_wagon_types x
+USING freight.cargo_types ct, freight.wagon_types wt
+WHERE x.cargo_type_id = ct.id AND x.wagon_type_id = wt.id
+  AND ((ct.code = 'E2E_IMP_WHEAT'  AND wt.code = 'PW2')
+    OR (ct.code = 'E2E_IMP_GRAINS' AND wt.code = 'CW4'));
+
 INSERT INTO freight.cargo_type_wagon_types (cargo_type_id, wagon_type_id)
 SELECT ct.id, wt.id
 FROM freight.cargo_types ct
 JOIN freight.wagon_types wt ON wt.code = 'PW2'
-WHERE ct.code IN ('E2E_IMP_GRAINS', 'E2E_IMP_WHEAT')
+WHERE ct.code = 'E2E_IMP_GRAINS'
   AND NOT EXISTS (
     SELECT 1 FROM freight.cargo_type_wagon_types x
     WHERE x.cargo_type_id = ct.id AND x.wagon_type_id = wt.id
@@ -182,7 +213,8 @@ WHERE ct.code IN ('E2E_IMP_GRAINS', 'E2E_IMP_WHEAT')
 INSERT INTO freight.locomotives
   (id, code, max_pull_weight_tons, max_train_length_meters, current_yard_id)
 SELECT gen_random_uuid(), v.code, 9000, 760, y.id
-FROM (VALUES ('LOCO-LED-1'), ('LOCO-LED-2')) AS v(code)
+FROM (VALUES ('LOCO-LED-1'), ('LOCO-LED-2'), ('LOCO-LED-3'), ('LOCO-LED-4'),
+             ('LOCO-LED-5'), ('LOCO-LED-6')) AS v(code)
 JOIN freight.yards y ON y.code = 'KALITY'
 WHERE NOT EXISTS (SELECT 1 FROM freight.locomotives l WHERE l.code = v.code);
 
@@ -269,7 +301,7 @@ INSERT INTO freight.cargo_type_wagon_types (cargo_type_id, wagon_type_id)
 SELECT ct.id, wt.id
 FROM freight.cargo_types ct
 JOIN freight.wagon_types wt ON wt.code = 'CW4'
-WHERE ct.code IN ('E2E_IMP_GRAINS', 'E2E_IMP_WHEAT')
+WHERE ct.code = 'E2E_IMP_WHEAT'
   AND NOT EXISTS (
     SELECT 1 FROM freight.cargo_type_wagon_types x
     WHERE x.cargo_type_id = ct.id AND x.wagon_type_id = wt.id
@@ -289,7 +321,7 @@ WHERE wt.id = w.wagon_type_id AND wt.code = 'CW4'
 -- Re-park the export CW4 pocket every seed (the mint above is insert-guarded).
 UPDATE freight.wagons w
 SET current_yard_id = (SELECT id FROM freight.yards
-                       WHERE code = CASE WHEN substring(w.wagon_number FROM 4)::int <= 80
+                       WHERE code = CASE WHEN substring(w.wagon_number FROM 4)::int <= 120
                                          THEN 'KALITY' ELSE 'DIRE_DAWA' END)
 FROM freight.wagon_types wt
 WHERE wt.id = w.wagon_type_id AND wt.code = 'CW4'
@@ -357,3 +389,62 @@ WHERE NOT EXISTS (
     AND r.origin_yard_id = a.id AND r.destination_yard_id = b.id
     AND r.deleted_at IS NULL
 );
+
+-- 7b. Customs clearance service fees — billed on the booking invoice together
+-- with the freight. Sold per direction + route + cargo kind: container fees
+-- per container type (one row per active 20/40ft type so whichever type the
+-- server resolves for a size always matches), bulk fees per commodity.
+-- Without these, every customs booking hard-blocks at pricing.
+INSERT INTO freight.rates
+  (id, rate_type, applies_to, trigger, currency, rate_value, rate_unit, status,
+   trade_direction, container_type_id, origin_yard_id, destination_yard_id,
+   proposed_by_staff_id)
+SELECT gen_random_uuid(), 'CUSTOMS_CLEARANCE', 'OTHER', 'CUSTOMS_CLEARANCE',
+       'USD', 50, 'PER_CONTAINER', 'LIVE', v.direction, ct.id, a.id, b.id, u.id
+FROM (VALUES
+        ('IMPORT', 'DJIB_PORT', 'KALITY'),
+        ('IMPORT', 'DJIB_PORT', 'MOJO'),
+        ('IMPORT', 'NAGAD',     'KALITY'),
+        ('IMPORT', 'NAGAD',     'MOJO'),
+        ('EXPORT', 'KALITY',    'DJIB_PORT'),
+        ('EXPORT', 'MOJO',      'DJIB_PORT'),
+        ('EXPORT', 'DIRE_DAWA', 'DJIB_PORT')
+     ) AS v(direction, from_code, to_code)
+JOIN freight.yards a ON a.code = v.from_code
+JOIN freight.yards b ON b.code = v.to_code
+JOIN freight.container_types ct ON ct.size_ft IN (20, 40) AND ct.is_active
+JOIN iam.users u ON u.email = 'operation@edr.local'
+WHERE NOT EXISTS (
+  SELECT 1 FROM freight.rates r
+  WHERE r.rate_type = 'CUSTOMS_CLEARANCE'
+    AND r.container_type_id = ct.id
+    AND r.origin_yard_id = a.id AND r.destination_yard_id = b.id
+    AND r.deleted_at IS NULL
+);
+
+INSERT INTO freight.rates
+  (id, rate_type, applies_to, trigger, currency, rate_value, rate_unit, status,
+   trade_direction, cargo_type_id, origin_yard_id, destination_yard_id,
+   proposed_by_staff_id)
+SELECT gen_random_uuid(), 'CUSTOMS_CLEARANCE', 'OTHER', 'CUSTOMS_CLEARANCE',
+       'USD', 2, 'PER_TON', 'LIVE', v.direction, cgt.id, a.id, b.id, u.id
+FROM (VALUES
+        ('IMPORT', 'DJIB_PORT', 'KALITY'),
+        ('IMPORT', 'DJIB_PORT', 'MOJO'),
+        ('IMPORT', 'NAGAD',     'KALITY'),
+        ('IMPORT', 'NAGAD',     'MOJO'),
+        ('EXPORT', 'KALITY',    'DJIB_PORT'),
+        ('EXPORT', 'MOJO',      'DJIB_PORT'),
+        ('EXPORT', 'DIRE_DAWA', 'DJIB_PORT')
+     ) AS v(direction, from_code, to_code)
+JOIN freight.yards a ON a.code = v.from_code
+JOIN freight.yards b ON b.code = v.to_code
+JOIN freight.cargo_types cgt ON cgt.code = 'E2E_IMP_WHEAT'
+JOIN iam.users u ON u.email = 'operation@edr.local'
+WHERE NOT EXISTS (
+  SELECT 1 FROM freight.rates r
+  WHERE r.rate_type = 'CUSTOMS_CLEARANCE'
+    AND r.cargo_type_id = cgt.id
+    AND r.origin_yard_id = a.id AND r.destination_yard_id = b.id
+    AND r.deleted_at IS NULL
+);