/**
 * Freight business steps, ported from e2e/freight/cypress/e2e/flows/import-utils.ts.
 *
 * Same philosophy as that file: contracts are seeded FULLY_EXECUTED in SQL
 * (the wizard is the Cypress suite's job), everything after that — bookings,
 * clearance, staff review, window phases, batching, invoicing, payment — runs
 * through the real API.
 *
 * The one deliberate difference: `settleViaGateway` is gone. Where the Cypress
 * suite fakes an intent row and POSTs to freight's own internal webhook, this
 * suite pays through the real payment microservice and lets the gateway mock
 * call back. That seam is the whole point of the suite.
 */
import { join } from "node:path";
import {
  api,
  apiOk,
  customerA,
  db,
  gateway,
  opsStaff,
  poll,
  superAdmin,
  upload,
} from "./client";

export const CORRIDOR = [
  "DJIB_PORT",
  "NAGAD",
  "DIRE_DAWA",
  "E2E_AWASH",
  "MOJO",
  "KALITY",
] as const;
export const ORIGIN = "DJIB_PORT";
export const DEST = "KALITY";
/** The export corridor runs the same six stops reversed (ET → DJ). */
export const EXP_ORIGIN = DEST;
export const EXP_DEST = ORIGIN;

/** Reused from the Cypress fixtures — a tiny PDF that satisfies the doc gate. */
export const DOC_FIXTURE = join(
  process.cwd(),
  "..",
  "e2e",
  "freight",
  "cypress",
  "fixtures",
  "docs",
  "license.pdf",
);

// ---------------------------------------------------------------------------
// time — departures pinned to 12:00 EAT so the EAT day key is unambiguous
// ---------------------------------------------------------------------------

export function departureAt(dayOffset: number): Date {
  const eatNow = new Date(Date.now() + 3 * 3_600_000);
  return new Date(
    Date.UTC(eatNow.getUTCFullYear(), eatNow.getUTCMonth(), eatNow.getUTCDate() + dayOffset, 9, 0, 0),
  );
}

export const eatDayStr = (d: Date) =>
  new Date(d.getTime() + 3 * 3_600_000).toISOString().slice(0, 10);

/** ISO 6346-shaped container number, unique per run+seed (checksum unchecked). */
export function isoNumber(runStamp: string, seed: number): string {
  return `MSCU${String((Number(runStamp.slice(-6)) * 100 + seed) % 10_000_000).padStart(7, "0")}`;
}

// ---------------------------------------------------------------------------
// contracts
// ---------------------------------------------------------------------------

export interface SeedContractOpts {
  suffix: string;
  reference: string;
  currency?: "ETB" | "USD";
  direction?: "IMPORT" | "EXPORT" | "DOMESTIC";
  freight?: "CONTAINER" | "BULK";
  /** Path B: booked by Global Logistics on the customer's behalf, customs tail after arrival. */
  customs?: boolean;
  originCode?: string;
  destCode?: string;
  /** TIN of the owning company. One tenant per booking — see {@link ensureTenant}. */
  tin?: string;
  /**
   * Service type to sell the contract under. Defaults to the oldest one (RAIL,
   * `includes_customs = false`). `RAIL_CUSTOMS` is what makes the rule engine's
   * CUSTOMS priority band apply — see seed-customs-service-type.sql.
   */
  serviceTypeCode?: string;
}

export const TIN_A = "0102030405"; // seed-company.sql
export const TIN_B = "0102030406"; // seed-company-b.sql (this suite)

/**
 * Seed a FULLY_EXECUTED one-time contract, container or bulk, self-clearing or
 * customs (Path B — the pre-booking boundary milestone is stamped COMPLETED so
 * the booking gate opens).
 */
export async function seedContract(opts: SeedContractOpts): Promise<string> {
  const direction = opts.direction ?? "IMPORT";
  const freight = opts.freight ?? "CONTAINER";
  const customs = opts.customs ?? false;
  const boundary = direction === "EXPORT" ? "EXPORT_RELEASED" : "DO_COLLECTED";
  await db(
    `WITH c AS (
       INSERT INTO freight.contracts
         (reference, company_id, company_profile_id, contract_kind,
          trade_direction, freight_type, service_type_id, payment_currency,
          customs_clearing_enabled, clearance_status, status,
          fully_executed_at, contract_valid_from, contract_valid_until,
          contract_summary)
       SELECT $1, comp.id,
              (SELECT p.id FROM freight.company_profiles p
                WHERE p.company_id = comp.id AND p.deleted_at IS NULL
                ORDER BY CASE
                  WHEN $2::text = 'EXPORT' AND p.type = 'exporter' THEN 0
                  WHEN $2::text <> 'EXPORT' AND p.type = 'importer' THEN 0
                  ELSE 1
                END
                LIMIT 1),
              'ONE_TIME', $2::text, $7::text,
              (SELECT st.id FROM freight.service_types st
                WHERE $10::text IS NULL OR st.code = $10::text
                ORDER BY st.created_at LIMIT 1),
              $3, $8,
              CASE WHEN $8 THEN 'CLEARANCE_READY_FOR_BOOKING' ELSE 'NOT_APPLICABLE' END,
              'FULLY_EXECUTED',
              now(), now() - interval '1 day', now() + interval '60 days',
              'IT payment-integration fixture contract'
       FROM freight.companies comp
       WHERE comp.tin = $4
       RETURNING id
     ), r AS (
       INSERT INTO freight.contract_routes
         (contract_id, origin_yard_id, destination_yard_id, sort_order)
       SELECT c.id, o.id, d.id, 0 FROM c
       JOIN freight.yards o ON o.code = $5
       JOIN freight.yards d ON d.code = $6
       RETURNING id
     ), scope_container AS (
       INSERT INTO freight.contract_cargo_scope
         (contract_id, container_size, cargo_free_text)
       SELECT c.id, v.size, 'IT corridor cargo'
       FROM c CROSS JOIN (VALUES ('20ft'), ('40ft')) AS v(size)
       WHERE $7::text = 'CONTAINER'
     ), scope_bulk AS (
       INSERT INTO freight.contract_cargo_scope
         (contract_id, cargo_type_id, cargo_free_text)
       SELECT c.id, ct.id, 'IT corridor wheat'
       FROM c JOIN freight.cargo_types ct ON ct.code = 'E2E_IMP_WHEAT'
       WHERE $7::text = 'BULK'
     )
     -- Path B gate: a customs ONE_TIME booking needs the pre-booking boundary
     -- milestone COMPLETED (IMPORT → DO_COLLECTED, EXPORT → EXPORT_RELEASED).
     INSERT INTO freight.clearance_milestones
       (contract_id, milestone_code, milestone_label, status, triggered_at, sort_order)
     SELECT c.id, $9, 'Pre-booking clearance boundary', 'COMPLETED', now(), 0
     FROM c WHERE $8`,
    [
      opts.reference,
      direction,
      opts.currency ?? "ETB",
      opts.tin ?? TIN_A,
      opts.originCode ?? (direction === "EXPORT" ? DEST : ORIGIN),
      opts.destCode ?? (direction === "EXPORT" ? ORIGIN : DEST),
      freight,
      customs,
      boundary,
      opts.serviceTypeCode ?? null,
    ],
  );
  const rows = await db<{ id: string }>(
    `SELECT id FROM freight.contracts WHERE reference = $1 ORDER BY created_at DESC LIMIT 1`,
    [opts.reference],
  );
  if (!rows[0]) throw new Error(`contract ${opts.reference} was not seeded`);
  return rows[0].id;
}

// ---------------------------------------------------------------------------
// route + schedule
// ---------------------------------------------------------------------------

export async function routeId(originCode = ORIGIN, destCode = DEST): Promise<string | null> {
  const rows = await db<{ id: string }>(
    `SELECT r.id FROM freight.routes r
     JOIN freight.yards o ON o.id = r.origin_yard_id AND o.code = $1
     JOIN freight.yards d ON d.id = r.destination_yard_id AND d.code = $2
     WHERE r.deleted_at IS NULL ORDER BY r.created_at DESC LIMIT 1`,
    [originCode, destCode],
  );
  return rows[0]?.id ?? null;
}

/** Create the 6-stop corridor route through the API if it doesn't exist yet. */
export async function ensureCorridorRoute(): Promise<void> {
  if (await routeId()) return;
  const yards = await db<{ id: string; code: string }>(
    `SELECT id, code FROM freight.yards WHERE code = ANY($1::text[])`,
    [[...CORRIDOR]],
  );
  if (yards.length !== CORRIDOR.length) {
    throw new Error(`corridor yards missing: got ${yards.length}/${CORRIDOR.length}`);
  }
  const byCode = new Map(yards.map((y) => [y.code, y.id]));
  await apiOk(opsStaff, "post", "/api/routes", {
    milestones: CORRIDOR.map((code) => ({ yardId: byCode.get(code) })),
  });
}

/** Create the reversed 6-stop corridor (ET → DJ = EXPORT) if missing. */
export async function ensureExportRoute(): Promise<void> {
  if (await routeId(EXP_ORIGIN, EXP_DEST)) return;
  const stops = [...CORRIDOR].reverse();
  const yards = await db<{ id: string; code: string }>(
    `SELECT id, code FROM freight.yards WHERE code = ANY($1::text[])`,
    [stops],
  );
  if (yards.length !== stops.length) throw new Error("corridor yards missing");
  const byCode = new Map(yards.map((y) => [y.code, y.id]));
  await apiOk(opsStaff, "post", "/api/routes", {
    milestones: stops.map((code) => ({ yardId: byCode.get(code) })),
  });
}

export interface ScheduleRow {
  id: string;
  status: string;
  window_phase: string;
  booking_window_status: string;
  booking_cycle_no: number;
  max_wagons: number;
  scheduled_departure_date: string;
  [k: string]: unknown;
}

const SCHEDULE_COLS = `ts.id, ts.status, ts.window_phase, ts.booking_window_status,
       ts.booking_cycle_no, ts.max_wagons, ts.scheduled_departure_date`;

export async function findSchedule(
  departure: Date,
  originCode = ORIGIN,
  destCode = DEST,
  windowSeconds = 3600,
): Promise<ScheduleRow | undefined> {
  const rows = await db<ScheduleRow>(
    `SELECT ${SCHEDULE_COLS}
     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))) < $4
     ORDER BY ts.created_at DESC LIMIT 1`,
    [originCode, destCode, departure.toISOString(), windowSeconds],
  );
  return rows[0];
}

export const findExportSchedule = (departure: Date) =>
  findSchedule(departure, EXP_ORIGIN, EXP_DEST);

/**
 * Wipe whatever a previous file left on this departure day. Same intent as
 * resetCorridorDay in import-utils.ts: a schedule must never be soft-deleted
 * while bookings still point at it, and leftover allocations keep eating wagons.
 */
export async function resetCorridorDay(
  departure: Date,
  originCode = ORIGIN,
  destCode = DEST,
): Promise<void> {
  await db(
    `WITH stale AS (
       SELECT ts.id 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))) < 43200
     ), unlink AS (
       UPDATE freight.bookings b
       SET train_schedule_id = NULL,
           status = CASE WHEN b.status IN ('FULLY_EXECUTED','SELECTED_FOR_BATCH','AWAITING_PAYMENT')
                         THEN 'EXPIRED' ELSE b.status END,
           scheduling_status = 'NOT_SCHEDULED',
           -- and its shipment DAY, or rescueStrandedPaidForDay re-places it:
           -- that sweep takes any unlinked booking with payment_status PAID and
           -- a scheduled_date on the day, so a previous run's paid fixtures
           -- climb straight back onto the fresh schedule (18 stowaway wagons
           -- the file never booked).
           scheduled_date = NULL
       FROM freight.contracts ct
       WHERE ct.id = b.contract_id AND ct.reference LIKE 'CTR-IT-%'
         AND b.train_schedule_id IN (SELECT id FROM stale)
     ), drop_links AS (
       UPDATE freight.train_schedule_bookings SET deleted_at = now()
       WHERE train_schedule_id IN (SELECT id FROM stale) AND deleted_at IS NULL
     ), free_wagons AS (
       UPDATE freight.wagon_booking_allocations wba SET deleted_at = now()
       FROM freight.bookings b
       JOIN freight.contracts ct ON ct.id = b.contract_id
       WHERE wba.booking_id = b.id AND wba.deleted_at IS NULL
         AND ct.reference LIKE 'CTR-IT-%'
         AND b.train_schedule_id IN (SELECT id FROM stale)
     )
     UPDATE freight.train_schedules SET deleted_at = now()
     WHERE id IN (SELECT id FROM stale)`,
    [originCode, destCode, departure.toISOString()],
  );
  // Unpinned leftovers from earlier files would contaminate this day's batch.
  await db(
    `UPDATE freight.bookings b SET status = 'EXPIRED'
     FROM freight.contracts ct
     WHERE ct.id = b.contract_id AND ct.reference LIKE 'CTR-IT-%'
       AND b.status = 'FULLY_EXECUTED' AND b.train_schedule_id IS NULL`,
  );
}

/**
 * Expire this suite's leftover unpaid holds.
 *
 * A company sitting on a SELECTED_FOR_BATCH / AWAITING_PAYMENT booking cannot
 * create another one (contract-booking.service.ts assertNoUnpaidHold) — a real
 * rule, and the reason each spec file must start from a clean tenant. Scoped to
 * `CTR-IT-%` contracts, so no other suite's data is ever touched.
 */
export async function releaseUnpaidHolds(): Promise<void> {
  await db(
    `UPDATE freight.bookings b
     SET status = 'EXPIRED', train_schedule_id = NULL, scheduling_status = 'NOT_SCHEDULED',
         -- Dropping the shipment day is what makes the retirement stick: the
         -- stranded-PAID sweep re-places any unlinked booking that still has
         -- payment_status PAID and a scheduled_date on the day being filled.
         scheduled_date = NULL
     FROM freight.contracts ct
     WHERE ct.id = b.contract_id AND ct.reference LIKE 'CTR-IT-%'
       AND b.status IN ('SELECTED_FOR_BATCH','AWAITING_PAYMENT','FULLY_EXECUTED','PAID')`,
  );
  // Same treatment for fixtures that were ALREADY terminal: a booking the
  // engine expired (or an earlier run retired) still carries payment_status
  // PAID and its shipment day, which is all rescueStrandedPaidForDay needs to
  // put it back on today's train. The status filter above never sees those
  // rows, so sweep them here.
  await db(
    `UPDATE freight.bookings b SET scheduled_date = NULL
     FROM freight.contracts ct
     WHERE ct.id = b.contract_id AND ct.reference LIKE 'CTR-IT-%'
       AND b.train_schedule_id IS NULL AND b.scheduled_date IS NOT NULL
       AND b.status IN ('EXPIRED','CANCELLED','REJECTED')`,
  );
  // Physical wagon stock is finite and every earlier file's allocations still
  // hold theirs (bookings that PAID and never "arrived" keep their wagons).
  // Freeing them here is safe because this runs at file start, before the file
  // books anything of its own — and without it the fifth or sixth file on the
  // same stack silently gets a short consist.
  await db(
    `UPDATE freight.wagon_booking_allocations wba SET deleted_at = now()
     FROM freight.bookings b
     JOIN freight.contracts ct ON ct.id = b.contract_id
     WHERE wba.booking_id = b.id AND wba.deleted_at IS NULL
       AND ct.reference LIKE 'CTR-IT-%'`,
  );
  // train_schedule_bookings.booking_id is UNIQUE and the constraint ignores
  // deleted_at, so a soft-unlinked booking can never be re-batched: on a warm
  // DB the next batch dies with "already exists for 'Booking Id'". These rows
  // belong to retired fixture bookings, so drop them outright.
  await db(
    `DELETE FROM freight.train_schedule_bookings tsb
     USING freight.bookings b
     JOIN freight.contracts ct ON ct.id = b.contract_id
     WHERE tsb.booking_id = b.id AND ct.reference LIKE 'CTR-IT-%'
       AND b.status IN ('EXPIRED','CANCELLED','REJECTED')`,
  );
}

/**
 * Ops creates a loco-pair schedule — capacity comes from maxWagonsPerTrain
 * (loco-pair mode); wagon stock is drawn from the origin yard at allocation.
 * `kind` picks the wagon family: container → NW5, bulk → CW4.
 */
export async function createSchedule(opts: {
  departure: Date;
  maxWagons?: number;
  locoPair?: [string, string];
  kind?: "container" | "bulk";
  originCode?: string;
  destCode?: string;
}): Promise<ScheduleRow> {
  const originCode = opts.originCode ?? ORIGIN;
  const destCode = opts.destCode ?? DEST;
  const existing = await findSchedule(opts.departure, originCode, destCode);
  if (!existing) {
    const route = await routeId(originCode, destCode);
    if (!route) throw new Error(`route ${originCode}→${destCode} missing`);
    const locos = await db<{ id: string }>(
      `SELECT id FROM freight.locomotives WHERE code = ANY($1::text[]) ORDER BY code`,
      [opts.locoPair ?? ["LOCO-IMP-1", "LOCO-IMP-2"]],
    );
    if (locos.length !== 2) throw new Error("schedule locomotives missing");
    await apiOk(opsStaff, "post", `/api/train-scheduling/${opts.kind ?? "container"}/schedules`, {
      routeId: route,
      scheduleDate: opts.departure.toISOString(),
      locomotiveIds: locos.map((l) => l.id),
      maxWagonsPerTrain: opts.maxWagons ?? 54,
    });
  }
  const schedule = await findSchedule(opts.departure, originCode, destCode);
  if (!schedule) throw new Error("schedule was not created");
  return schedule;
}

/** Back-compat alias for the container-import payment specs. */
export const createImportSchedule = (opts: {
  departure: Date;
  maxWagons?: number;
  locoPair?: [string, string];
}) => createSchedule(opts);

export function scheduleRow(scheduleId: string) {
  return db<ScheduleRow>(
    `SELECT ${SCHEDULE_COLS} , ts.window_opens_at, ts.window_closes_at, ts.payment_phase_ends_at
     FROM freight.train_schedules ts WHERE ts.id = $1`,
    [scheduleId],
  ).then((rows) => rows[0]);
}

async function pollSchedulePhase(scheduleId: string, want: string[], attempts = 40) {
  return poll<{ window_phase: string; booking_window_status: string }>(
    `schedule ${scheduleId} → ${want.join("|")}`,
    `SELECT window_phase, booking_window_status FROM freight.train_schedules WHERE id = $1`,
    [scheduleId],
    (row) => !!row && want.includes(row.window_phase),
    { attempts },
  );
}

/** Pull window_opens_at into the past; the app's 10s tick flips PRE_WINDOW→OPEN. */
export async function forceWindowOpen(scheduleId: string, closesInMinutes = 45): Promise<void> {
  await db(
    `UPDATE freight.train_schedules
     SET window_opens_at = now() - interval '1 minute',
         window_closes_at = now() + ($2 || ' minutes')::interval
     WHERE id = $1`,
    [scheduleId, String(closesInMinutes)],
  );
  await pollSchedulePhase(scheduleId, ["OPEN"]);
  await poll(
    `schedule ${scheduleId} bookable`,
    `SELECT booking_window_status FROM freight.train_schedules WHERE id = $1`,
    [scheduleId],
    (row) => (row as { booking_window_status?: string })?.booking_window_status === "OPEN",
  );
}

export async function closeBookingWindow(scheduleId: string): Promise<void> {
  await db(
    `UPDATE freight.train_schedules SET window_closes_at = now() - interval '1 second'
     WHERE id = $1 AND window_phase = 'OPEN'`,
    [scheduleId],
  );
  await pollSchedulePhase(scheduleId, ["DOC_REVIEW"]);
}

/**
 * Staff end document review early → PAYMENT: the priority batch runs over the
 * route-day pool, reserves wagons and ISSUES THE INVOICES this suite pays.
 */
export async function completeDocReview(scheduleId: string): Promise<void> {
  await apiOk(opsStaff, "post", `/api/train-scheduling/schedules/${scheduleId}/doc-review-complete`);
  await pollSchedulePhase(scheduleId, ["PAYMENT", "DONE", "PRE_WINDOW"]);
}

// ---------------------------------------------------------------------------
// bookings
// ---------------------------------------------------------------------------

export interface BookingRow {
  id: string;
  reference: string;
  status: string;
  scheduling_status: string;
  train_schedule_id: string | null;
  payment_deadline: string | null;
  contract_id: string;
  [k: string]: unknown;
}

export async function bookingFor(contractId: string): Promise<BookingRow> {
  const rows = await db<BookingRow>(
    `SELECT b.id, b.reference, b.status, b.scheduling_status, b.train_schedule_id,
            b.payment_deadline, b.contract_id
     FROM freight.bookings b
     WHERE b.contract_id = $1 AND b.deleted_at IS NULL
     ORDER BY b.created_at DESC LIMIT 1`,
    [contractId],
  );
  if (!rows[0]) throw new Error(`no booking under contract ${contractId}`);
  return rows[0];
}

export function pollBookingStatus(bookingId: string, status: string | string[], attempts = 30) {
  const want = Array.isArray(status) ? status : [status];
  return poll<BookingRow>(
    `booking ${bookingId} → ${want.join("|")}`,
    `SELECT status, scheduling_status FROM freight.bookings WHERE id = $1`,
    [bookingId],
    (row) => !!row && want.includes(row.status),
    { attempts },
  );
}

/** Customer books containers under a seeded contract. */
export async function bookContainers(opts: {
  contractId: string;
  runStamp: string;
  isoSeed: number;
  twenty?: number;
  forty?: number;
  scheduledDate: string;
  vgmTons?: number;
  as?: string;
}) {
  const vgm = opts.vgmTons ?? 10;
  const lines: Array<Record<string, unknown>> = [];
  let unit = 0;
  const line = (size: string, qty: number) => ({
    containerSize: size,
    quantity: qty,
    units: Array.from({ length: qty }, () => ({
      containerNumber: isoNumber(opts.runStamp, opts.isoSeed + unit++),
      vgmTons: vgm,
    })),
  });
  if (opts.twenty) lines.push(line("20ft", opts.twenty));
  if (opts.forty) lines.push(line("40ft", opts.forty));

  return api(opts.as ?? customerA, "post", `/api/contracts/${opts.contractId}/bookings`, {
    scheduledDate: opts.scheduledDate,
    containers: lines,
  });
}

/**
 * Upload one ad-hoc doc → GL approves → finalize → customer proceeds with the
 * shipment day. The e2e seed configures no required documents, so a single doc
 * satisfies the 100%-approved gate.
 */
export async function clearBooking(
  bookingId: string,
  scheduledDate: string,
  as = customerA,
): Promise<void> {
  await upload(superAdmin, `/api/bookings/${bookingId}/clearance/documents`, DOC_FIXTURE, "custom_e2e");
  await apiOk(superAdmin, "post", `/api/bookings/${bookingId}/clearance/review`, {
    fileKey: "custom_e2e",
    status: "APPROVED",
  });
  await apiOk(superAdmin, "post", `/api/bookings/${bookingId}/clearance/finalize`);
  await apiOk(as, "post", `/api/bookings/${bookingId}/clearance/proceed`, { scheduledDate });
  await pollBookingStatus(bookingId, "OPERATION_REQUEST_PENDING", 20);
}

/** Ops accepts the operation request → FULLY_EXECUTED (enters the day pool). */
export async function acceptOperation(bookingId: string): Promise<void> {
  await apiOk(opsStaff, "post", `/api/bookings/${bookingId}/operation/review`, {
    decision: "ACCEPT",
  });
  await pollBookingStatus(bookingId, "FULLY_EXECUTED", 15);
}

// ---------------------------------------------------------------------------
// invoices
// ---------------------------------------------------------------------------

export interface InvoiceRow {
  id: string;
  invoice_number: string;
  status: string;
  total_amount: string;
  balance_amount: string | null;
  paid_amount: string | null;
  currency: string;
  payment_id: string | null;
  source: string;
  source_id: string;
  paid_at: string | null;
  [k: string]: unknown;
}

export function invoiceForBooking(bookingId: string, attempts = 30) {
  return poll<InvoiceRow>(
    `invoice for booking ${bookingId}`,
    `SELECT * FROM freight.invoices
     WHERE source_id = $1 AND source = 'booking' AND deleted_at IS NULL
     ORDER BY created_at DESC LIMIT 1`,
    [bookingId],
    (row) => !!row,
    { attempts },
  );
}

export function currentInvoice(invoiceId: string) {
  return db<InvoiceRow>(`SELECT * FROM freight.invoices WHERE id = $1`, [invoiceId]).then(
    (rows) => rows[0],
  );
}

/** The freight-side projection of a gateway intent. */
export function freightPayment(intentId: string) {
  return db<{ id: string; status: string; merchant_order_id: string; transaction_id: string | null }>(
    `SELECT id, status, merchant_order_id, transaction_id FROM freight.payments WHERE id = $1`,
    [intentId],
  ).then((rows) => rows[0]);
}

// ---------------------------------------------------------------------------
// payment
// ---------------------------------------------------------------------------

/**
 * Customer pays an invoice from the portal. This is the real chain:
 *   portal → billing.payInvoice → PaymentClientService → payment API
 *   → provider (gateway mock) → intent, and the invoice gets `payment_id`.
 */
export function payInvoice(
  invoiceId: string,
  opts: { method?: string; as?: string; payerAccount?: string } = {},
) {
  return api(opts.as ?? customerA, "post", `/api/billing/my-invoices/${invoiceId}/pay`, {
    method: opts.method ?? "CBE_BIRR",
    platform: "web",
    ...(opts.payerAccount ? { payerAccount: opts.payerAccount } : {}),
  });
}

/** The gateway intent the payment service opened for a booking. */
export function gatewayIntent(bookingId: string, attempts = 15) {
  return poll<{
    id: string;
    status: string;
    merchant_order_id: string;
    amount_minor: string;
    currency: string;
    provider: string;
    expires_at: string | null;
  }>(
    `payment intent for booking ${bookingId}`,
    `SELECT id, status, merchant_order_id, amount_minor, currency, provider, expires_at
     FROM edr_payment.payment_intent
     WHERE reference_id = $1 ORDER BY created_at DESC LIMIT 1`,
    [bookingId],
    (row) => !!row,
    { attempts, intervalMs: 1000 },
  );
}

// ---------------------------------------------------------------------------
// the whole arrange chain, in one call
// ---------------------------------------------------------------------------

export interface ReadyBooking {
  contractId: string;
  bookingId: string;
}

/**
 * Contract → booking → clearance → ops accept, ending in the day pool
 * (FULLY_EXECUTED). Invoices are not issued yet: that happens when the batch
 * runs — see {@link runBatch}. Roughly 30–60s of real API work per booking.
 */
export async function prepareBooking(opts: {
  suffix: string;
  departure: Date;
  runStamp: string;
  isoSeed: number;
  twenty?: number;
  forty?: number;
  currency?: "ETB" | "USD";
  tin?: string;
  as?: string;
}): Promise<ReadyBooking> {
  const contractId = await seedContract({
    suffix: opts.suffix,
    reference: `CTR-IT-${opts.runStamp}-${opts.suffix}`,
    currency: opts.currency,
    tin: opts.tin,
  });
  const day = eatDayStr(opts.departure);
  const res = await bookContainers({
    contractId,
    runStamp: opts.runStamp,
    isoSeed: opts.isoSeed,
    twenty: opts.twenty ?? 2,
    forty: opts.forty,
    scheduledDate: day,
    as: opts.as,
  });
  if (res.status > 201) {
    throw new Error(`booking ${opts.suffix} rejected: ${res.status} ${JSON.stringify(res.body)}`);
  }
  const booking = await bookingFor(contractId);
  await clearBooking(booking.id, day, opts.as ?? customerA);
  await acceptOperation(booking.id);
  return { contractId, bookingId: booking.id };
}

/**
 * Close the booking window and end doc review — the batch reserves wagons by
 * priority and issues the invoices. Everything pooled for the day settles here.
 */
export async function runBatch(scheduleId: string): Promise<void> {
  await closeBookingWindow(scheduleId);
  await completeDocReview(scheduleId);
}

// ---------------------------------------------------------------------------
// tenants — one company per booking
// ---------------------------------------------------------------------------

/**
 * A company may hold only ONE unpaid reservation at a time
 * (contract-booking.service.ts assertNoUnpaidHold, company-scoped). Every
 * multi-booking scenario here therefore needs a company per booking: without
 * it the second booking of any train-filling scenario is rejected with
 * "You already have a booking waiting for payment".
 *
 * Staff (`superAdmin`) book and pay on the customer's behalf, so these tenants
 * need no portal user — which is also the real Path B flow for customs.
 */
export async function ensureTenant(slug: string): Promise<string> {
  const tin = tinFor(slug);
  await db(
    `WITH c AS (
       INSERT INTO freight.companies
         (id, name, type, status, tin, fan_number, country, address, phone, email,
          nationality, kind, attributes)
       SELECT gen_random_uuid(), 'IT Tenant ' || $1::text, 'customer', 'active', $2::text,
              $2::text || '000000', 'Ethiopia', 'Addis Ababa',
              '+2519' || substr($2::text, 3, 8),
              'ops+' || lower($1::text) || '@it-tenant.test', 'ethiopian', 'commercial', '{}'::jsonb
       WHERE NOT EXISTS (SELECT 1 FROM freight.companies WHERE tin = $2::text)
       RETURNING id
     ), pick AS (
       SELECT id FROM c
       UNION ALL
       SELECT id FROM freight.companies WHERE tin = $2::text
       LIMIT 1
     )
     INSERT INTO freight.company_profiles (id, company_id, type, status, reference)
     SELECT gen_random_uuid(), pick.id, v.type, 'active', upper(substr(v.type, 1, 3)) || $2::text
     FROM pick CROSS JOIN (VALUES ('importer'), ('exporter')) AS v(type)
     WHERE NOT EXISTS (
       SELECT 1 FROM freight.company_profiles p
       WHERE p.company_id = pick.id AND p.type = v.type
     )`,
    [slug, tin],
  );
  return tin;
}

/** Deterministic 10-digit TIN per tenant slug — stable across runs, unique per slug. */
export function tinFor(slug: string): string {
  let h = 0;
  for (const ch of slug) h = (h * 31 + ch.charCodeAt(0)) % 100_000_000;
  return `20${String(h).padStart(8, "0")}`;
}

/** Seed one tenant + one contract per suffix, and return suffix → contractId. */
export async function seedTenantContracts(
  runStamp: string,
  specs: Array<
    { suffix: string } & Omit<SeedContractOpts, "suffix" | "reference" | "tin">
  >,
): Promise<Map<string, string>> {
  const out = new Map<string, string>();
  for (const spec of specs) {
    const tin = await ensureTenant(`${runStamp}-${spec.suffix}`);
    out.set(
      spec.suffix,
      await seedContract({
        ...spec,
        reference: `CTR-IT-${runStamp}-${spec.suffix}`,
        tin,
      }),
    );
  }
  return out;
}

// ---------------------------------------------------------------------------
// bulk bookings
// ---------------------------------------------------------------------------

/**
 * Book bulk tons under a seeded BULK contract. Wagon demand = ceil(tons / 70)
 * on CW4 covered gondolas. Omit `scheduledDate` for DOMESTIC (intercity)
 * bookings — they never pick a shipment day.
 */
export async function bookBulk(opts: {
  contractId: string;
  tons: number;
  scheduledDate?: string;
  cargoCode?: string;
  as?: string;
}) {
  const [row] = await db<{ cargo_type_id: string }>(
    `SELECT t.id AS cargo_type_id FROM freight.cargo_types t WHERE t.code = $1`,
    [opts.cargoCode ?? "E2E_IMP_WHEAT"],
  );
  if (!row) throw new Error(`cargo type ${opts.cargoCode ?? "E2E_IMP_WHEAT"} not seeded`);
  return api(opts.as ?? superAdmin, "post", `/api/contracts/${opts.contractId}/bookings`, {
    ...(opts.scheduledDate ? { scheduledDate: opts.scheduledDate } : {}),
    bulkLines: [{ cargoTypeId: row.cargo_type_id, cargoWeightTons: opts.tons }],
    cargoFreeText: "IT corridor wheat",
  });
}

/**
 * Book a PER_ITEM break-bulk line (automobiles / machinery from
 * seed-bulk-items.sql). Wagon demand is the greater of the per-item floor and
 * the tonnage math — which is exactly what these scenarios pin down.
 */
export async function bookBulkItems(opts: {
  contractId: string;
  cargoCode: "E2E_IMP_AUTO" | "E2E_IMP_MACHINE";
  items: number;
  tons: number;
  scheduledDate?: string;
  hazardousQuantity?: number;
  reeferQuantity?: number;
  as?: string;
}) {
  const [row] = await db<{ cargo_type_id: string }>(
    `SELECT t.id AS cargo_type_id FROM freight.cargo_types t WHERE t.code = $1`,
    [opts.cargoCode],
  );
  if (!row) throw new Error(`cargo type ${opts.cargoCode} not seeded`);
  return api(opts.as ?? superAdmin, "post", `/api/contracts/${opts.contractId}/bookings`, {
    ...(opts.scheduledDate ? { scheduledDate: opts.scheduledDate } : {}),
    bulkLines: [
      {
        cargoTypeId: row.cargo_type_id,
        itemCount: opts.items,
        cargoWeightTons: opts.tons,
        ...(opts.hazardousQuantity != null ? { hazardousQuantity: opts.hazardousQuantity } : {}),
        ...(opts.reeferQuantity != null ? { reeferQuantity: opts.reeferQuantity } : {}),
      },
    ],
    cargoFreeText: `IT break-bulk ${opts.cargoCode}`,
  });
}

/** Book PER_ITEM cargo and walk it to the pool. */
export async function bookBulkItemsReady(opts: {
  contractId: string;
  cargoCode: "E2E_IMP_AUTO" | "E2E_IMP_MACHINE";
  items: number;
  tons: number;
  scheduledDate: string;
}): Promise<string> {
  const res = await bookBulkItems(opts);
  if (res.status > 201) {
    throw new Error(`break-bulk booking rejected: ${res.status} ${JSON.stringify(res.body)}`);
  }
  const booking = await bookingFor(opts.contractId);
  await clearBooking(booking.id, opts.scheduledDate, superAdmin);
  await acceptOperation(booking.id);
  return booking.id;
}

/**
 * Type-integrity check for mixed trains: every wagon slot under the booking is
 * of ONE expected type (containers → NW5, bulk → CW4) and the count matches.
 * No wheat on a flat wagon, no box in a gondola.
 */
export async function expectWagonType(bookingId: string, code: string, wagons: number) {
  const row = await poll<{ code: string; n: string }>(
    `booking ${bookingId} rides ${wagons}× ${code}`,
    `SELECT wt.code, count(*)::text AS n
     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 = $1 AND wba.deleted_at IS NULL
     GROUP BY wt.code`,
    [bookingId],
    (r) => r?.code === code && Number(r?.n) === wagons,
    { attempts: 20 },
  );
  const [kinds] = await db<{ k: string }>(
    `SELECT count(DISTINCT tsw.wagon_type_id)::text AS k
     FROM freight.wagon_booking_allocations wba
     JOIN freight.train_set_wagons tsw ON tsw.id = wba.train_set_wagon_id
     WHERE wba.booking_id = $1 AND wba.deleted_at IS NULL`,
    [bookingId],
  );
  if (Number(kinds.k) !== 1) {
    throw new Error(`booking ${bookingId} rides ${kinds.k} wagon types, expected 1`);
  }
  return row;
}

/** Book + walk the clearance gate + ops accept, ending FULLY_EXECUTED in the pool. */
export async function bookBulkReady(opts: {
  contractId: string;
  tons: number;
  scheduledDate: string;
  mode?: "import" | "export";
  /** Path B — walk the phased chain instead of the one-shot finalize. */
  customs?: boolean;
}): Promise<string> {
  const res = await bookBulk(opts);
  if (res.status > 201) {
    throw new Error(`bulk booking rejected: ${res.status} ${JSON.stringify(res.body)}`);
  }
  const booking = await bookingFor(opts.contractId);
  if (opts.customs && opts.mode === "export") {
    await clearBookingPhasedCustomsExport(booking.id, opts.scheduledDate);
  } else if (opts.customs) {
    await clearBookingPhasedCustoms(booking.id, opts.scheduledDate);
  } else {
    await clearBooking(booking.id, opts.scheduledDate, superAdmin);
  }
  if (opts.mode === "export") await acceptExport(booking.id);
  else await acceptOperation(booking.id);
  return booking.id;
}

/**
 * Book, walk the clearance gate, and expect the shipment-day request to be
 * REFUSED.
 *
 * The window and export-space gates no longer sit on booking creation — the
 * booking is born in the clearance gate and the day is only validated when the
 * customer picks it (`clearance/proceed` → requestOperation,
 * contract-booking.service.ts). So "rejected at submission" now means rejected
 * at the day request; creation itself succeeds.
 */
export async function expectDayRefused(opts: {
  contractId: string;
  tons: number;
  scheduledDate: string;
}): Promise<{ status: number; body: unknown }> {
  const res = await bookBulk(opts);
  if (res.status > 201) return { status: res.status, body: res.body };

  const booking = await bookingFor(opts.contractId);
  await upload(superAdmin, `/api/bookings/${booking.id}/clearance/documents`, DOC_FIXTURE, "custom_e2e");
  await apiOk(superAdmin, "post", `/api/bookings/${booking.id}/clearance/review`, {
    fileKey: "custom_e2e",
    status: "APPROVED",
  });
  await apiOk(superAdmin, "post", `/api/bookings/${booking.id}/clearance/finalize`);
  const proceed = await api(superAdmin, "post", `/api/bookings/${booking.id}/clearance/proceed`, {
    scheduledDate: opts.scheduledDate,
  });
  return { status: proceed.status, body: proceed.body };
}

/** Ops accepts an EXPORT operation request — FCFS: the accept itself reserves. */
export async function acceptExport(bookingId: string): Promise<void> {
  await apiOk(opsStaff, "post", `/api/bookings/${bookingId}/operation/review`, {
    decision: "ACCEPT",
  });
  await pollBookingStatus(bookingId, ["SELECTED_FOR_BATCH", "AWAITING_PAYMENT"], 15);
}

/**
 * Walk a customs (Path B) IMPORT booking through the PHASED clearance chain.
 *
 * `clearance/finalize` refuses these outright ("General customs bookings use
 * phased clearance…") — the Cypress suite documents that gap and leaves its
 * customs bookings failing. The real chain, GL Ethiopia and GL Djibouti in
 * turn, is:
 *   docs upload → review APPROVED → request transit assignee → assign it →
 *   customs declaration → duty advice (none due) → transit permit →
 *   finalize pre-clearance → delivery order (DO_COLLECTED → ready for
 *   operation) → customer picks the shipment day.
 */
export async function clearBookingPhasedCustoms(
  bookingId: string,
  scheduledDate: string,
): Promise<void> {
  await upload(superAdmin, `/api/bookings/${bookingId}/clearance/documents`, DOC_FIXTURE, "custom_e2e");
  await apiOk(superAdmin, "post", `/api/bookings/${bookingId}/clearance/review`, {
    fileKey: "custom_e2e",
    status: "APPROVED",
  });

  await apiOk(superAdmin, "post", `/api/bookings/${bookingId}/clearance/transit-assignee/request`, {
    note: "it",
  });
  await apiOk(superAdmin, "post", `/api/bookings/${bookingId}/clearance/transit-assignee/assign`, {
    transitAgentId: await ensureTransitAgent(),
  });

  // GL ET sends a priced draft first; the customer accepts it before the real
  // declaration may be filed (DRAFT_DECLARATION_SENT gates pre-clearance).
  await upload(superAdmin, `/api/bookings/${bookingId}/clearance/draft-declaration`, DOC_FIXTURE, "files", {
    price: "1000",
    currency: "ETB",
  });
  await apiOk(superAdmin, "post", `/api/bookings/${bookingId}/clearance/draft-declaration/accept`);

  await upload(superAdmin, `/api/bookings/${bookingId}/clearance/declaration`, DOC_FIXTURE, "declaration");
  // No duty due — the slip step is then skipped by the workflow itself.
  await upload(superAdmin, `/api/bookings/${bookingId}/clearance/duty`, DOC_FIXTURE, "attachment", {
    dutyRequired: "false",
  });
  await upload(superAdmin, `/api/bookings/${bookingId}/clearance/transit-permit`, DOC_FIXTURE, "transit_permit");
  await apiOk(superAdmin, "post", `/api/bookings/${bookingId}/clearance/finalize-pre-clearance`);

  const vesselArrival = new Date(Date.now() - 24 * 3_600_000).toISOString().slice(0, 10);
  await upload(superAdmin, `/api/bookings/${bookingId}/clearance/delivery-order`, DOC_FIXTURE, "file", {
    vesselArrivalDate: vesselArrival,
    doCollectedDate: vesselArrival,
  });

  await apiOk(superAdmin, "post", `/api/bookings/${bookingId}/clearance/proceed`, { scheduledDate });
  await pollBookingStatus(bookingId, "OPERATION_REQUEST_PENDING", 20);
}

/**
 * EXPORT half of the phased customs chain: no transit assignee and no duty —
 * GL ET files the declaration, GL DJ secures the Release Order (which must
 * lead the vessel departure by the configured minimum), and that release is
 * what unlocks the shipment day.
 */
export async function clearBookingPhasedCustomsExport(
  bookingId: string,
  scheduledDate: string,
): Promise<void> {
  await upload(superAdmin, `/api/bookings/${bookingId}/clearance/documents`, DOC_FIXTURE, "custom_e2e");
  await apiOk(superAdmin, "post", `/api/bookings/${bookingId}/clearance/review`, {
    fileKey: "custom_e2e",
    status: "APPROVED",
  });
  await upload(superAdmin, `/api/bookings/${bookingId}/clearance/declaration`, DOC_FIXTURE, "declaration");

  const vesselDeparture = new Date(Date.now() + 7 * 24 * 3_600_000).toISOString().slice(0, 10);
  const ro = await upload(
    superAdmin,
    `/api/bookings/${bookingId}/clearance/release-order`,
    DOC_FIXTURE,
    "file",
    { vesselDepartureDate: vesselDeparture },
  );
  if (ro.status > 201) {
    throw new Error(`release order → ${ro.status}: ${JSON.stringify(ro.body)}`);
  }

  await apiOk(superAdmin, "post", `/api/bookings/${bookingId}/clearance/proceed`, { scheduledDate });
  await pollBookingStatus(bookingId, "OPERATION_REQUEST_PENDING", 20);
}

/** One reusable transit officer — the roster is empty in a fresh stack. */
export async function ensureTransitAgent(): Promise<string> {
  const [existing] = await db<{ id: string }>(
    `SELECT id FROM freight.transit_agents
     WHERE is_active AND deleted_at IS NULL AND valid_from <= now() AND valid_to >= now()
     LIMIT 1`,
  );
  if (existing) return existing.id;
  const [created] = await db<{ id: string }>(
    `INSERT INTO freight.transit_agents (name, valid_from, valid_to, is_active)
     VALUES ('IT Transit Officer', now() - interval '1 day', now() + interval '1 year', true)
     RETURNING id`,
  );
  return created.id;
}

/** DOMESTIC bookings have no shipment day — finalize alone lands FULLY_EXECUTED. */
export async function clearIntercityBooking(bookingId: string): Promise<void> {
  await upload(superAdmin, `/api/bookings/${bookingId}/clearance/documents`, DOC_FIXTURE, "custom_e2e");
  await apiOk(superAdmin, "post", `/api/bookings/${bookingId}/clearance/review`, {
    fileKey: "custom_e2e",
    status: "APPROVED",
  });
  await apiOk(superAdmin, "post", `/api/bookings/${bookingId}/clearance/finalize`);
  await pollBookingStatus(bookingId, "FULLY_EXECUTED", 15);
}

// ---------------------------------------------------------------------------
// batch engine controls
// ---------------------------------------------------------------------------

/** Batch fill reserves by priority DESC — order 1 = first pick. */
export function setPriority(bookingId: string, order: number) {
  return db(`UPDATE freight.bookings SET priority_score = $2 WHERE id = $1`, [
    bookingId,
    1000 - order,
  ]);
}

/**
 * Push the pay window out for the bookings a test intends to PAY.
 *
 * A reservation's deadline is ~60s in this stack. Paying through the real
 * gateway (initiate → provider → signed callback → broker → settle) plus the
 * allocation wait costs more than that when several bookings pay in sequence,
 * so on a loaded stack the last hold expires before its payment lands and the
 * engine strands it ("re-placed stranded PAID booking …"). Widening the window
 * is arrange, not assertion: the tests that are ABOUT expiry never call this.
 */
export async function extendPayWindow(
  scheduleId: string,
  bookingIds: string[],
  minutes = 30,
): Promise<void> {
  await db(
    `UPDATE freight.bookings SET payment_deadline = now() + ($2 || ' minutes')::interval
     WHERE id = ANY($1::uuid[])`,
    [bookingIds, String(minutes)],
  );
  await db(
    `UPDATE freight.train_schedules
     SET payment_phase_ends_at = GREATEST(
           COALESCE(payment_phase_ends_at, now()),
           now() + ($2 || ' minutes')::interval)
     WHERE id = $1`,
    [scheduleId, String(minutes)],
  );
}

/** End the payment phase now — the tick settles (allocate paid / expire unpaid). */
export function endPaymentPhase(scheduleId: string) {
  return db(
    `UPDATE freight.train_schedules
     SET payment_phase_ends_at = now() - interval '1 second'
     WHERE id = $1 AND window_phase = 'PAYMENT'`,
    [scheduleId],
  );
}

/**
 * Push a reservation's pay deadline an hour into the past and wait for the
 * engine to expire it.
 *
 * An hour, not a second: promoting a waiting booking extends the payment phase
 * (extendPaymentPhaseForTopUp), and a deadline only just behind `now()` can end
 * up on the wrong side of that move. Expiry also requires the payment service
 * to answer the reconcile-before-expire question — which it really does here.
 */
export async function forceReservationExpiry(bookingId: string): Promise<void> {
  await db(
    `UPDATE freight.bookings SET payment_deadline = now() - interval '1 hour' WHERE id = $1`,
    [bookingId],
  );
  await pollBookingStatus(bookingId, "EXPIRED", 40);
}

/**
 * Allocation is driven by the schedule's settle tick, not by the payment
 * response — on a busy stack (full-suite run) that tick can be several cycles
 * behind, so wait generously rather than flake.
 */
export function pollAllocations(bookingId: string, minWagons = 1) {
  return poll<{ n: string }>(
    `booking ${bookingId} wagon allocations >= ${minWagons}`,
    `SELECT count(*)::text AS n FROM freight.wagon_booking_allocations
     WHERE booking_id = $1 AND deleted_at IS NULL`,
    [bookingId],
    (row) => Number(row?.n ?? 0) >= minWagons,
    { attempts: 45, intervalMs: 3000 },
  );
}

/** Distinct wagon slots committed to a schedule — the consist occupancy. */
export async function allocatedWagons(scheduleId: string): Promise<number> {
  const [row] = await db<{ n: string }>(
    `SELECT count(DISTINCT wba.train_set_wagon_id)::text 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`,
    [scheduleId],
  );
  return Number(row.n);
}

/** The open partial-split offer the batch made a booking that did not fit whole. */
export function pollPartialOffer(bookingId: string) {
  return poll<{ status: string; offered_wagons: number }>(
    `booking ${bookingId} partial offer`,
    `SELECT status, offered_wagons FROM freight.booking_batch_offers
     WHERE booking_id = $1 AND deleted_at IS NULL
     ORDER BY created_at DESC LIMIT 1`,
    [bookingId],
    (row) => row?.status === "OFFERED",
    { attempts: 15 },
  );
}

/**
 * Wait for the payment phase to settle out.
 *
 * Only a FULL train rests at DONE. A day that ends under-filled CONCLUDES and
 * REOPENS — window_phase goes back to OPEN on the next cycle — so waiting for
 * DONE on an under-filled train waits forever.
 */
export const pollCycleConcluded = (scheduleId: string) =>
  pollWindow(scheduleId, (s) => s.window_phase !== "PAYMENT", "payment phase concluded");

export function pollWindow(scheduleId: string, check: (row: ScheduleRow) => boolean, label: string) {
  return poll<ScheduleRow>(
    `schedule ${scheduleId} ${label}`,
    `SELECT ${SCHEDULE_COLS} FROM freight.train_schedules ts WHERE ts.id = $1`,
    [scheduleId],
    (row) => !!row && check(row),
    { attempts: 40 },
  );
}

// ---------------------------------------------------------------------------
// paying through the REAL gateway
// ---------------------------------------------------------------------------

/**
 * Settle a reservation the way production does: initiate on the invoice
 * (freight → payment service → provider), then have the gateway call back.
 *
 * This replaces the Cypress suite's staff `mark-paid` shortcut everywhere —
 * mark-paid skips `booking.invoice.paid`, so it never applies a pending split
 * offer and never proves the payment seam. Uses the staff initiate endpoint
 * (`/api/payments/initiate`) because these fixtures are booked on the
 * customer's behalf and have no portal user of their own.
 */
export async function payViaGateway(
  bookingId: string,
  opts: { method?: string; expectStatus?: string[] } = {},
): Promise<void> {
  const invoice = await invoiceForBooking(bookingId);
  const res = await api(opsStaff, "post", "/api/payments/initiate", {
    invoiceId: invoice.id,
    method: opts.method ?? "CBE_BIRR",
    platform: "web",
  });
  if (res.status > 201) {
    throw new Error(`initiate for ${bookingId} → ${res.status}: ${JSON.stringify(res.body)}`);
  }
  const intent = await gatewayIntent(bookingId);
  const hook = await gateway.webhook({ merchantOrderId: intent.merchant_order_id });
  if (hook.body?.delivered !== 200) {
    throw new Error(`webhook delivery failed: ${JSON.stringify(hook.body)}`);
  }
  await poll(
    `invoice ${invoice.invoice_number} PAID`,
    `SELECT status FROM freight.invoices WHERE id = $1`,
    [invoice.id],
    (row) => (row as { status?: string })?.status === "PAID",
    { attempts: 30 },
  );
  await pollBookingStatus(bookingId, opts.expectStatus ?? ["PAID", "IN_TRANSIT", "ARRIVED"], 30);
}

// ---------------------------------------------------------------------------
// train journey + customs tail
// ---------------------------------------------------------------------------

export const gatePassGranted = (scheduleId: string) =>
  apiOk(opsStaff, "post", `/api/train-scheduling/schedules/${scheduleId}/import-djibouti/gatepass-granted`);

export const finalizeSchedule = (scheduleId: string) =>
  apiOk(opsStaff, "post", `/api/train-scheduling/schedules/${scheduleId}/finalize`);

export const dispatchSchedule = (scheduleId: string) =>
  apiOk(opsStaff, "post", `/api/train-scheduling/schedules/${scheduleId}/dispatch`);

export const recordCheckpoint = (scheduleId: string, sequenceNo: number, kind: string) =>
  apiOk(opsStaff, "post", `/api/train-scheduling/schedules/${scheduleId}/checkpoints`, {
    sequenceNo,
    kind,
  });

/** Run the corridor: four PASSED checkpoints, then ARRIVED at the terminal. */
export async function runCorridor(scheduleId: string): Promise<void> {
  for (const seq of [1, 2, 3, 4]) await recordCheckpoint(scheduleId, seq, "PASSED");
  await recordCheckpoint(scheduleId, 5, "ARRIVED");
  await pollWindow(scheduleId, (s) => s.status === "ARRIVED", "ARRIVED");
}

export const uploadT1 = (bookingId: string) =>
  upload(superAdmin, `/api/contracts/bookings/${bookingId}/t1-documents`, DOC_FIXTURE);

export const uploadTransportDocument = (bookingId: string) =>
  upload(superAdmin, `/api/contracts/bookings/${bookingId}/transport-document`, DOC_FIXTURE);

export const closeT1 = (bookingId: string) =>
  apiOk(superAdmin, "post", `/api/contracts/bookings/${bookingId}/t1-close`);

/** Post-arrival import tail: T1 close → risk → second duty → final invoice. */
export async function runImportCustomsTail(bookingId: string): Promise<void> {
  await closeT1(bookingId);
  await apiOk(superAdmin, "post", `/api/contracts/bookings/${bookingId}/risk`, {
    riskLevel: "GREEN",
  });
  await upload(superAdmin, `/api/contracts/bookings/${bookingId}/second-duty`, DOC_FIXTURE, "attachment", {
    dutyRequired: "false",
  });
  await upload(superAdmin, `/api/contracts/bookings/${bookingId}/final-invoice`, DOC_FIXTURE, "file", {
    amount: "1000",
    description: "it final invoice",
  });
  // Issue → approve → customer pays → GL confirms. The slip is rejected while
  // the invoice is still a draft.
  await apiOk(superAdmin, "post", `/api/contracts/bookings/${bookingId}/final-invoice/approve`);
  const slip = await upload(
    superAdmin,
    `/api/contracts/bookings/${bookingId}/final-invoice-slip`,
    DOC_FIXTURE,
    "file",
  );
  if (slip.status > 201) {
    throw new Error(`final invoice slip → ${slip.status}: ${JSON.stringify(slip.body)}`);
  }
  await apiOk(superAdmin, "post", `/api/contracts/bookings/${bookingId}/final-invoice/confirm`);
  await completeMilestone(bookingId, "IMPORT_RELEASE_GRANTED");
  await completeMilestone(bookingId, "IMPORT_PROCESS_COMPLETED");
}

export const completeMilestone = (bookingId: string, code: string) =>
  apiOk(superAdmin, "post", `/api/contracts/bookings/${bookingId}/milestones/${code}/complete`, {
    note: "it",
  });

export function expectMilestoneDone(bookingId: string, code: string) {
  return poll<{ n: string }>(
    `booking ${bookingId} milestone ${code}`,
    `SELECT count(*)::text AS n FROM freight.clearance_milestones
     WHERE booking_id = $1 AND milestone_code = $2 AND status = 'COMPLETED' AND deleted_at IS NULL`,
    [bookingId, code],
    (row) => Number(row?.n ?? 0) > 0,
    { attempts: 15 },
  );
}

export async function milestoneCount(bookingId: string, code: string): Promise<number> {
  const [row] = await db<{ n: string }>(
    `SELECT count(*)::text AS n FROM freight.clearance_milestones
     WHERE booking_id = $1 AND milestone_code = $2 AND status = 'COMPLETED' AND deleted_at IS NULL`,
    [bookingId, code],
  );
  return Number(row.n);
}

export const acceptIntercityOnto = (scheduleId: string, bookingId: string) =>
  apiOk(opsStaff, "post", `/api/train-scheduling/schedules/${scheduleId}/intercity/accept`, {
    bookingIds: [bookingId],
  });

// ---------------------------------------------------------------------------
// GROUP 1 — wagon arithmetic on the 53-wagon BUILT container train
// ---------------------------------------------------------------------------

/** The Group 1 built train's consist size — see seed-g1-train.sql. */
export const G1_WAGONS = 53;
export const G1_TRAIN = "TRN-G1-1";

/** seed-government.sql — the kind='government' company POST /api/bookings needs. */
export const GOV_COMPANY_ID = "0a1b0001-0000-4000-8000-000000000001";
export const GOV_PROFILE_ID = "0b1c0001-0000-4000-8000-000000000001";

/**
 * Wagons a container booking needs: two 20ft share a wagon, a 40ft takes one.
 * An odd 20ft count still costs a whole wagon — and `assert20ftPairable`
 * refuses to submit one — so every scenario keeps 20ft quantities even.
 */
export const containerWagons = (twenty = 0, forty = 0) => Math.ceil(twenty / 2) + forty;

/**
 * Schedule the BUILT train rather than a locomotive pair.
 *
 * `maxWagonsPerTrain` is not a real cap on the pair path: syncScheduleMaxWagons
 * recomputes max_wagons from locomotive length (54 on this corridor) every fill
 * pass. A built train's coupled consist wins outright
 * (`physicalWagons ?? capacityLimits(loco)`), so the 53 wagons seed-g1-train.sql
 * couples ARE the capacity — which is the number Group 1's arithmetic is
 * written in. The count is asserted here: a 52-wagon consist would shift every
 * scenario by a slot and fail far from the cause.
 */
export async function createBuiltTrainSchedule(opts: {
  departure: Date;
  trainCode?: string;
  wagons?: number;
}): Promise<ScheduleRow> {
  const trainCode = opts.trainCode ?? G1_TRAIN;
  if (!(await findSchedule(opts.departure))) {
    const route = await routeId();
    if (!route) throw new Error(`route ${ORIGIN}→${DEST} missing`);
    const [train] = await db<{ id: string }>(
      `SELECT id FROM freight.trains WHERE code = $1`,
      [trainCode],
    );
    if (!train) throw new Error(`built train ${trainCode} missing — seed-g1-train.sql`);
    await apiOk(opsStaff, "post", "/api/train-scheduling/container/schedules", {
      routeId: route,
      scheduleDate: opts.departure.toISOString(),
      trainId: train.id,
    });
  }
  const schedule = await findSchedule(opts.departure);
  if (!schedule) throw new Error("built-train schedule was not created");
  const wagons = opts.wagons ?? G1_WAGONS;
  if (Number(schedule.max_wagons) !== wagons) {
    throw new Error(
      `${trainCode} scheduled with max_wagons=${schedule.max_wagons}, expected the ${wagons}-wagon consist`,
    );
  }
  return schedule;
}

/** Book containers on the customer's behalf and walk them into the day pool. */
export async function bookContainersReady(opts: {
  contractId: string;
  runStamp: string;
  isoSeed: number;
  twenty?: number;
  forty?: number;
  scheduledDate: string;
  vgmTons?: number;
  /** Path B contract — walk the phased chain instead of the one-shot finalize. */
  customs?: boolean;
}): Promise<string> {
  const res = await bookContainers({ ...opts, as: superAdmin });
  if (res.status > 201) {
    throw new Error(`container booking rejected: ${res.status} ${JSON.stringify(res.body)}`);
  }
  const booking = await bookingFor(opts.contractId);
  if (opts.customs) await clearBookingPhasedCustoms(booking.id, opts.scheduledDate);
  else await clearBooking(booking.id, opts.scheduledDate, superAdmin);
  await acceptOperation(booking.id);
  return booking.id;
}

/** Containers still on a booking — the number a split REDUCES. */
export async function containerCount(bookingId: string): Promise<number> {
  const [row] = await db<{ q: string | null }>(
    `SELECT sum(quantity)::text AS q FROM freight.booking_container
     WHERE booking_id = $1 AND deleted_at IS NULL`,
    [bookingId],
  );
  return Number(row?.q ?? 0);
}

/** Assert the batch never raised a partial offer — the whole-or-nothing cases. */
export async function expectNoPartialOffer(bookingId: string, label = bookingId): Promise<void> {
  const [row] = await db<{ n: string }>(
    `SELECT count(*)::text AS n FROM freight.booking_batch_offers
     WHERE booking_id = $1 AND deleted_at IS NULL`,
    [bookingId],
  );
  if (Number(row.n) !== 0) throw new Error(`${label} was offered a split (${row.n} offer rows)`);
}

/**
 * Every container on the train mapped to a wagon slot, with a real number on it.
 *
 * Wagon counts alone cannot catch a half-done allocation: a booking whose
 * wagons were reserved but whose units were never placed still reads as a full
 * train. The units live in `wagon_allocation_container_items` — one row per
 * container, and the marshalling sheet is generated from that column.
 */
export async function expectContainersPlaced(scheduleId: string, containers: number) {
  const placed = poll<{ n: string }>(
    `${containers} containers mapped to wagon slots on ${scheduleId}`,
    `SELECT count(*)::text AS n
     FROM freight.wagon_allocation_container_items ci
     JOIN freight.wagon_booking_allocations wba ON wba.id = ci.wagon_booking_allocation_id
     JOIN freight.train_schedule_bookings tsb
       ON tsb.booking_id = wba.booking_id AND tsb.train_schedule_id = $1
     WHERE ci.deleted_at IS NULL AND wba.deleted_at IS NULL AND tsb.deleted_at IS NULL`,
    [scheduleId],
    (row) => Number(row?.n ?? 0) === containers,
    { attempts: 25 },
  );
  await placed;
  const [blank] = await db<{ n: string }>(
    `SELECT count(*)::text AS n
     FROM freight.wagon_allocation_container_items ci
     JOIN freight.wagon_booking_allocations wba ON wba.id = ci.wagon_booking_allocation_id
     JOIN freight.train_schedule_bookings tsb
       ON tsb.booking_id = wba.booking_id AND tsb.train_schedule_id = $1
     WHERE ci.deleted_at IS NULL AND wba.deleted_at IS NULL AND tsb.deleted_at IS NULL
       AND (ci.container_number IS NULL OR ci.container_number = '')`,
    [scheduleId],
  );
  if (Number(blank.n) !== 0) {
    throw new Error(`${blank.n} slot(s) placed without a container number`);
  }
}

/**
 * GROSS tonnage riding a schedule — cargo PLUS the tare of every wagon it
 * occupies, because a locomotive hauls the wagon as well as what is in it.
 *
 * `allocated_weight_tons` holds the CARGO alone, so the tare of each allocated
 * wagon type is added here. Reading the column raw understates a loaded consist
 * by 22.4 T a wagon and makes a weight-bound train look half empty.
 */
export async function allocatedGrossTons(scheduleId: string): Promise<number> {
  const [row] = await db<{ tons: string | null }>(
    `SELECT COALESCE(sum(wba.allocated_weight_tons + wt.tare_weight_tons), 0)::text AS tons
     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
     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`,
    [scheduleId],
  );
  return Number(row.tons ?? 0);
}

/** Wagons a single booking currently holds. */
export async function wagonAllocationCount(bookingId: string): Promise<number> {
  const [row] = await db<{ n: string }>(
    `SELECT count(*)::text AS n FROM freight.wagon_booking_allocations
     WHERE booking_id = $1 AND deleted_at IS NULL`,
    [bookingId],
  );
  return Number(row.n);
}

/** Bookings still linked to a schedule — the seats actually held. */
export async function linkedBookings(scheduleId: string): Promise<number> {
  const [row] = await db<{ n: string }>(
    `SELECT count(*)::text AS n FROM freight.train_schedule_bookings
     WHERE train_schedule_id = $1 AND deleted_at IS NULL`,
    [scheduleId],
  );
  return Number(row.n);
}

/** Invoices still payable against a booking — an expiry must leave none. */
export async function livePayableInvoices(bookingId: string): Promise<number> {
  const [row] = await db<{ n: string }>(
    `SELECT count(*)::text AS n FROM freight.invoices
     WHERE source = 'booking' AND source_id = $1 AND deleted_at IS NULL
       AND paid_at IS NULL
       -- ::text because the enum has no VOID member; comparing the label
       -- directly makes Postgres reject the whole query.
       AND status::text NOT IN ('EXPIRED','CANCELLED','VOID')`,
    [bookingId],
  );
  return Number(row.n);
}

/** The portal's own availability question for a shipment day. */
export async function dayAvailability(bookingId: string, day: string) {
  const res = await apiOk(
    superAdmin,
    "get",
    `/api/bookings/${bookingId}/day-availability?date=${day}`,
  );
  const body = res.body as {
    trainsForDay?: boolean;
    freeWagons?: number;
    data?: { trainsForDay?: boolean; freeWagons?: number };
  };
  return body.data ?? body;
}

/**
 * An EXPIRED booking is recoverable without re-approval: the CONTRACT is still
 * executed, so the customer can rebook a later day. Asserting the contract (not
 * just the booking) is the point — a bug that retired it would strand them.
 */
export async function expectContractStillBookable(bookingId: string): Promise<void> {
  const booking = await bookingRow(bookingId);
  const [contract] = await db<{ status: string }>(
    `SELECT status FROM freight.contracts WHERE id = $1`,
    [booking.contract_id],
  );
  if (!["FULLY_EXECUTED", "CONTRACT_ACTIVE"].includes(contract.status)) {
    throw new Error(`contract of ${bookingId} is ${contract.status}, no longer bookable`);
  }
}

/**
 * Let a partial offer lapse instead of paying it. The offer dies with the
 * booking's pay deadline, so pushing the deadline back is what the wall clock
 * would do — and the settle tick then expires the booking WHOLE.
 */
export async function forceOfferLapse(bookingId: string): Promise<void> {
  await db(
    `UPDATE freight.bookings SET payment_deadline = now() - interval '1 hour' WHERE id = $1`,
    [bookingId],
  );
  await pollBookingStatus(bookingId, "EXPIRED", 40);
}

/**
 * A government booking, created the way the product creates one: staff POST
 * /api/bookings against the seeded kind='government' company (never the
 * contract wizard), then `government-expedite` promotes it to PAID/Eligible —
 * it rides without paying.
 */
export async function createGovernmentBooking(opts: {
  forty: number;
  vgmTons?: number;
  scheduledDate?: string;
}): Promise<string> {
  const vgm = opts.vgmTons ?? 10;
  const [origin] = await db<{ id: string }>(`SELECT id FROM freight.yards WHERE code = $1`, [ORIGIN]);
  const [dest] = await db<{ id: string }>(`SELECT id FROM freight.yards WHERE code = $1`, [DEST]);
  const [service] = await db<{ id: string }>(
    `SELECT id FROM freight.service_types ORDER BY created_at LIMIT 1`,
  );
  const [ctype] = await db<{ id: string }>(
    `SELECT id FROM freight.container_types WHERE size_ft = 40 AND is_active LIMIT 1`,
  );
  await apiOk(superAdmin, "post", "/api/bookings", {
    isGovernment: true,
    companyId: GOV_COMPANY_ID,
    companyProfileId: GOV_PROFILE_ID,
    contractType: "NEW",
    serviceTypeId: service.id,
    equipmentReturn: "WITHOUT_RETURN",
    originYardId: origin.id,
    destinationYardId: dest.id,
    tradeDirection: "IMPORT",
    freightType: "CONTAINER",
    containers: [{ containerTypeId: ctype.id, quantity: opts.forty, vgmPerUnitTons: vgm }],
    cargoTotalWeightVgm: opts.forty * vgm,
    paymentCurrency: "USD",
    ...(opts.scheduledDate ? { scheduledDate: opts.scheduledDate } : {}),
  });
  const [row] = await db<{ id: string }>(
    `SELECT id FROM freight.bookings
     WHERE company_id = $1 AND is_government = true AND deleted_at IS NULL
     ORDER BY created_at DESC LIMIT 1`,
    [GOV_COMPANY_ID],
  );
  if (!row) throw new Error("government booking was not created");
  return row.id;
}

/** Idempotent retry path — create() already expedites, this re-asserts it. */
export const governmentExpedite = (bookingId: string) =>
  apiOk(superAdmin, "post", `/api/bookings/${bookingId}/government-expedite`);

/** Staff pin: fillSchedule's pool is keyed on `booking.train_schedule_id`. */
export const pinToSchedule = (bookingId: string, scheduleId: string) =>
  db(`UPDATE freight.bookings SET train_schedule_id = $1 WHERE id = $2`, [scheduleId, bookingId]);

/** Staff "run batch" button — a fill pass on demand, no phase change. */
export const triggerBatchRun = (scheduleId: string) =>
  apiOk(superAdmin, "post", `/api/train-scheduling/schedules/${scheduleId}/run-batch`);

/** Wipe this suite's leftover government bookings so "newest" is unambiguous. */
export async function releaseGovernmentBookings(): Promise<void> {
  await db(
    `UPDATE freight.wagon_booking_allocations wba SET deleted_at = now()
     FROM freight.bookings b
     WHERE wba.booking_id = b.id AND wba.deleted_at IS NULL AND b.company_id = $1`,
    [GOV_COMPANY_ID],
  );
  await db(
    `DELETE FROM freight.train_schedule_bookings tsb USING freight.bookings b
     WHERE tsb.booking_id = b.id AND b.company_id = $1`,
    [GOV_COMPANY_ID],
  );
  await db(
    `UPDATE freight.bookings SET status = 'CANCELLED', train_schedule_id = NULL,
            scheduling_status = 'NOT_SCHEDULED', deleted_at = now()
     WHERE company_id = $1 AND deleted_at IS NULL`,
    [GOV_COMPANY_ID],
  );
}

/**
 * A priority band, created and dropped by the test that needs it. The rule
 * engine scores each booking additively from the ACTIVE bands, so a band left
 * behind would re-rank every later file's pool.
 */
export async function createPriorityConfig(body: {
  type: "WAGON" | "CURRENCY" | "CUSTOMS";
  label: string;
  currency?: "ETB" | "USD";
  minWagonCount: number;
  maxWagonCount: number;
  scorePoints: number;
}): Promise<string> {
  // Through the API, not SQL: bands must be contiguous from 1 per type
  // (assertNoRangeCollision) and points are capped at 50 — a hand-inserted row
  // could violate both and score a booking the product never would.
  const res = await apiOk(superAdmin, "post", "/api/priority-configs", {
    ...body,
    isActive: true,
  });
  const id = (res.body?.data?.id ?? res.body?.id) as string | undefined;
  if (!id) throw new Error(`priority config not created: ${JSON.stringify(res.body)}`);
  return id;
}

export const dropPriorityConfig = (id: string) =>
  api(superAdmin, "delete", `/api/priority-configs/${id}`);

export async function bookingRow(bookingId: string): Promise<BookingRow> {
  const [row] = await db<BookingRow>(
    `SELECT b.id, b.reference, b.status, b.scheduling_status, b.train_schedule_id,
            b.payment_deadline, b.contract_id, b.is_split, b.pre_split_quantities,
            b.wagons_required, b.cargo_total_weight_vgm, b.priority_score,
            b.is_government
     FROM freight.bookings b WHERE b.id = $1`,
    [bookingId],
  );
  return row;
}