/**
 * Shared helpers for the IMPORT corridor flow specs.
 *
 * Corridor (A→B→C→D→E→T, DJ→ET = IMPORT):
 *   DJIB_PORT → NAGAD → DIRE_DAWA → E2E_AWASH → MOJO → KALITY
 *
 * Philosophy (same as segment_weight.cy.ts): these specs test the
 * scheduling/window/clearance ENGINE, not the contract wizard — contracts are
 * seeded FULLY_EXECUTED in SQL with stamped references; bookings, staff
 * reviews, window phases, payment, dispatch and clearance run through the real
 * API + UI. Window *timestamps* are arranged via db:query (the specs arrange
 * window state, they don't test the wall clock), and every transition is then
 * performed by the app's own 10s window tick or its staff endpoints.
 *
 * No module-level state besides constants: Cypress re-evaluates the spec
 * bundle on cross-origin reloads, so helpers look rows up by stamped-reference
 * SUFFIX + newest row, never by captured ids.
 */

export const customer = "user@gmail.com";
export const companyTin = "0102030405"; // seed-company.sql
export const opsStaff = "operation@edr.local";
/** isSuperAdmin bypasses assertFreightPermission — used for GL endpoints so a
 *  missing preset permission never masks an engine regression. */
export const superAdmin = "superadmin@tria.com";

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

export const apiUrl = () => Cypress.env("apiUrl") as string;

// ---------------------------------------------------------------------------
// small generic plumbing
// ---------------------------------------------------------------------------

export type Row = Record<string, string | number | null>;

export function db<T = Row>(sql: string, params: unknown[] = []) {
  return cy.task<{ rowCount: number; rows: T[] }>("db:query", { sql, params }, { log: false });
}

/** Bearer token for a staff/customer account (portal users use the demo pwd). */
export function tokenFor(email: string): Cypress.Chainable<string> {
  const pass =
    email.endsWith("@gmail.com") ? (Cypress.env("demoPassword") as string) : undefined;
  return cy.apiLogin(email, pass).then(({ token }) => cy.wrap(token, { log: false }));
}

export function apiPost(
  email: string,
  path: string,
  body?: unknown,
  failOnStatusCode = true,
) {
  return tokenFor(email).then((token) =>
    cy.request({
      method: "POST",
      url: `${apiUrl()}${path}`,
      headers: { Authorization: `Bearer ${token}` },
      body: body ?? {},
      failOnStatusCode,
    }),
  );
}

/** Poll a 1-row query until `check` passes (10s window tick ⇒ 3s cadence). */
export function pollDb<T = Row>(
  label: string,
  sql: string,
  params: unknown[],
  check: (row: T | undefined) => boolean,
  attempts = 40,
) {
  const read = (attempt: number): void => {
    db<T>(sql, params).then(({ rows }) => {
      if (check(rows[0])) return;
      expect(attempt, label).to.be.lessThan(attempts);
      cy.wait(3000, { log: false }).then(() => read(attempt + 1));
    });
  };
  read(0);
}

// ---------------------------------------------------------------------------
// 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, // 09:00 UTC = 12:00 EAT
      0,
      0,
    ),
  );
}

/** The EAT calendar day (`YYYY-MM-DD`) of an instant — the booking day key. */
export const eatDayStr = (d: Date) =>
  new Date(d.getTime() + 3 * 3_600_000).toISOString().slice(0, 10);

// ---------------------------------------------------------------------------
// contracts — seeded FULLY_EXECUTED (see file header)
// ---------------------------------------------------------------------------

export interface SeedContractOpts {
  suffix: string;
  reference: string;
  currency?: "ETB" | "USD";
  customs?: boolean;
  direction?: "IMPORT" | "EXPORT" | "DOMESTIC";
  freight?: "CONTAINER" | "BULK";
  originCode?: string;
  destCode?: string;
  /** GENERAL = multi-booking drawdown contract (status CONTRACT_ACTIVE). */
  kind?: "ONE_TIME" | "GENERAL";
  /** GENERAL only: quantity cap on the 20ft scope row (40ft stays uncapped). */
  cap20?: number;
}

export function seedImportContract(opts: SeedContractOpts) {
  const currency = opts.currency ?? "ETB";
  const customs = opts.customs ?? false;
  const direction = opts.direction ?? "IMPORT";
  const freight = opts.freight ?? "CONTAINER";
  const kind = opts.kind ?? "ONE_TIME";
  const status = kind === "GENERAL" ? "CONTRACT_ACTIVE" : "FULLY_EXECUTED";
  // Pre-booking boundary milestone for Path B contracts differs by direction.
  const boundary = direction === "EXPORT" ? "EXPORT_RELEASED" : "DO_COLLECTED";
  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),
              $11::text, $2::text, $9::text,
              (SELECT st.id FROM freight.service_types st ORDER BY st.created_at LIMIT 1),
              $3, $4,
              CASE WHEN $4 THEN 'CLEARANCE_READY_FOR_BOOKING' ELSE 'NOT_APPLICABLE' END,
              $12::text, now(), now() - interval '1 day',
              now() + interval '60 days', 'E2E import-corridor fixture contract'
       FROM freight.companies comp
       WHERE comp.tin = $5
         -- before() re-runs on cross-origin reloads: keep one stable fresh row
         -- per suffix (skip when this run already seeded an unbooked one).
         AND NOT EXISTS (
           SELECT 1 FROM freight.contracts c2
           WHERE c2.reference LIKE 'CTR-IMP-%-' || $8
             AND c2.deleted_at IS NULL
             AND c2.created_at > now() - interval '30 minutes'
             AND NOT EXISTS (
               SELECT 1 FROM freight.bookings b2 WHERE b2.contract_id = c2.id
             )
         )
       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 = $6
       JOIN freight.yards d ON d.code = $7
       RETURNING id
     ), scope_container AS (
       INSERT INTO freight.contract_cargo_scope
         (contract_id, container_size, quantity_cap, cargo_free_text)
       SELECT c.id, v.size,
              CASE WHEN v.size = '20ft' THEN $13::numeric END,
              'E2E import corridor cargo'
       FROM c CROSS JOIN (VALUES ('20ft'), ('40ft')) AS v(size)
       WHERE $9::text = 'CONTAINER'
     ), scope_bulk AS (
       INSERT INTO freight.contract_cargo_scope
         (contract_id, cargo_type_id, cargo_free_text)
       SELECT c.id, ct.id, 'E2E import wheat'
       FROM c JOIN freight.cargo_types ct ON ct.code = 'E2E_IMP_WHEAT'
       WHERE $9::text = 'BULK'
     )
     -- Path B gate: ONE_TIME customs bookings require the pre-booking boundary
     -- milestone (IMPORT → DO_COLLECTED, EXPORT → EXPORT_RELEASED) COMPLETED.
     INSERT INTO freight.clearance_milestones
       (contract_id, milestone_code, milestone_label, status, triggered_at, sort_order)
     SELECT c.id, $10, 'Pre-booking clearance boundary', 'COMPLETED', now(), 0
     FROM c WHERE $4`,
    [
      opts.reference,
      direction,
      currency,
      customs,
      companyTin,
      opts.originCode ?? ORIGIN,
      opts.destCode ?? DEST,
      opts.suffix,
      freight,
      boundary,
      kind,
      status,
      opts.cap20 ?? null,
    ],
  );
  // Backfill the boundary milestone when the insert above was skipped because
  // a prior run's still-unbooked contract row is being reused.
  if (customs) {
    db(
      `INSERT INTO freight.clearance_milestones
         (contract_id, milestone_code, milestone_label, status, triggered_at, sort_order)
       SELECT ct.id, $2::text, 'Pre-booking clearance boundary', 'COMPLETED', now(), 0
       FROM freight.contracts ct
       WHERE ct.reference LIKE 'CTR-IMP-%-' || $1 AND ct.deleted_at IS NULL
         AND NOT EXISTS (
           SELECT 1 FROM freight.clearance_milestones m
           WHERE m.contract_id = ct.id AND m.milestone_code = $2::text
             AND m.deleted_at IS NULL
         )`,
      [opts.suffix, boundary],
    );
  }
}

/** Newest seeded contract for a suffix — stamp-agnostic. */
export function dbContractId(suffix: string) {
  return db<{ id: string }>(
    `SELECT id FROM freight.contracts
     WHERE reference LIKE 'CTR-IMP-%-' || $1
     ORDER BY created_at DESC LIMIT 1`,
    [suffix],
  ).then(({ rows }) => {
    expect(rows, `seeded contract *-${suffix}`).to.have.length(1);
    return cy.wrap(rows[0].id, { log: false });
  });
}

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

export interface BookingRow {
  id: string;
  reference: string;
  status: string;
  scheduling_status: string;
  train_schedule_id: string | null;
  payment_deadline: string | null;
  priority_score: number;
  is_split: boolean;
  contract_id: string;
}

export function dbBooking(suffix: string) {
  return db<BookingRow>(
    `SELECT b.id, b.reference, b.status, b.scheduling_status,
            b.train_schedule_id, b.payment_deadline, b.priority_score,
            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
     ORDER BY b.created_at DESC LIMIT 1`,
    [suffix],
  );
}

export function withBooking(suffix: string, fn: (b: BookingRow) => void) {
  dbBooking(suffix).then(({ rows }) => {
    expect(rows, `booking under *-${suffix}`).to.have.length(1);
    fn(rows[0]);
  });
}

export function expectBookingStatus(suffix: string, status: string | string[]) {
  const want = Array.isArray(status) ? status : [status];
  withBooking(suffix, (b) =>
    expect(b.status, `${suffix} booking status`).to.be.oneOf(want),
  );
}

export function pollBookingStatus(suffix: string, status: string | string[], attempts = 40) {
  const want = Array.isArray(status) ? status : [status];
  pollDb<BookingRow>(
    `${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
     ORDER BY b.created_at DESC LIMIT 1`,
    [suffix],
    (row) => !!row && want.includes(row.status as string),
    attempts,
  );
}

/** 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")}`;
}

/**
 * Customer books containers under a seeded contract via the API (the portal
 * booking form is exercised by export_one_time/intercity specs; a 22-wagon
 * booking means 44 ISO inputs — not a UI journey).
 */
export function bookContainers(opts: {
  suffix: string;
  runStamp: string;
  isoSeed: number;
  twenty?: number;
  forty?: number;
  scheduledDate?: string; // omit for DOMESTIC (intercity)
  vgmTons?: number;
  expectFailure?: string | RegExp; // substring/regex of the expected 4xx error
}) {
  const vgm = opts.vgmTons ?? 10;
  const lines: Array<Record<string, unknown>> = [];
  let unit = 0;
  if (opts.twenty) {
    lines.push({
      containerSize: "20ft",
      quantity: opts.twenty,
      units: Array.from({ length: opts.twenty }, () => ({
        containerNumber: isoNumber(opts.runStamp, opts.isoSeed + unit++),
        vgmTons: vgm,
      })),
    });
  }
  if (opts.forty) {
    lines.push({
      containerSize: "40ft",
      quantity: opts.forty,
      units: Array.from({ length: opts.forty }, () => ({
        containerNumber: isoNumber(opts.runStamp, opts.isoSeed + unit++),
        vgmTons: vgm,
      })),
    });
  }
  db<{ id: string; customs_clearing_enabled: boolean }>(
    `SELECT id, customs_clearing_enabled FROM freight.contracts
     WHERE reference LIKE 'CTR-IMP-%-' || $1
     ORDER BY created_at DESC LIMIT 1`,
    [opts.suffix],
  ).then(({ rows }) => {
    expect(rows, `seeded contract *-${opts.suffix}`).to.have.length(1);
    // Path B: customs-clearance contracts are booked by Global Logistics on
    // behalf of the customer — the portal user is rejected with a 403.
    const actor = rows[0].customs_clearing_enabled ? superAdmin : customer;
    apiPost(
      actor,
      `/api/contracts/${rows[0].id}/bookings`,
      {
        ...(opts.scheduledDate ? { scheduledDate: opts.scheduledDate } : {}),
        containers: lines,
      },
      !opts.expectFailure,
    ).then((res) => {
      if (opts.expectFailure) {
        expect(res.status, `${opts.suffix} booking rejected`).to.be.within(400, 422);
        if (opts.expectFailure instanceof RegExp) {
          expect(JSON.stringify(res.body)).to.match(opts.expectFailure);
        } else {
          expect(JSON.stringify(res.body)).to.include(opts.expectFailure);
        }
      } else {
        expect(res.status, `${opts.suffix} booking created`).to.be.oneOf([200, 201]);
      }
    });
  });
}

/**
 * Book bulk tons under a seeded BULK contract via the API. Wagon demand =
 * ceil(tons / 70) on CW4 covered gondolas.
 */
export function bookBulk(opts: {
  suffix: string;
  tons: number;
  scheduledDate?: string; // omit for DOMESTIC (intercity)
  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
     FROM freight.contracts ct
     WHERE ct.reference LIKE 'CTR-IMP-%-' || $1
     ORDER BY ct.created_at DESC LIMIT 1`,
    [opts.suffix],
  ).then(({ rows }) => {
    expect(rows, `seeded contract *-${opts.suffix}`).to.have.length(1);
    const actor = rows[0].customs_clearing_enabled ? superAdmin : customer;
    apiPost(
      actor,
      `/api/contracts/${rows[0].id}/bookings`,
      {
        ...(opts.scheduledDate ? { scheduledDate: opts.scheduledDate } : {}),
        bulkLines: [{ cargoTypeId: rows[0].cargo_type_id, cargoWeightTons: opts.tons }],
        cargoFreeText: "E2E import wheat",
      },
      !opts.expectFailure,
    ).then((res) => {
      if (opts.expectFailure) {
        expect(res.status, `${opts.suffix} booking rejected`).to.be.within(400, 422);
        if (opts.expectFailure instanceof RegExp) {
          expect(JSON.stringify(res.body)).to.match(opts.expectFailure);
        } else {
          expect(JSON.stringify(res.body)).to.include(opts.expectFailure);
        }
      } else {
        expect(res.status, `${opts.suffix} booking created`).to.be.oneOf([200, 201]);
      }
    });
  });
}

/**
 * Walk a GENERAL booking through its PER-BOOKING clearance chain (Path A):
 * AWAITING_DOCUMENTS → upload one doc → GL approves it → finalize →
 * CLEARANCE_READY → customer proceeds with the shipment day →
 * OPERATION_REQUEST_PENDING → ops accept → FULLY_EXECUTED (pool). The e2e
 * seed configures no required documents, so one ad-hoc doc satisfies the
 * 100%-approved gate.
 */
export function clearGeneralBooking(suffix: string, scheduledDate: string) {
  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");
    apiPost(superAdmin, `/api/bookings/${b.id}/clearance/review`, {
      fileKey: "custom_e2e",
      status: "APPROVED",
    })
      .its("status")
      .should("be.oneOf", [200, 201]);
    apiPost(superAdmin, `/api/bookings/${b.id}/clearance/finalize`)
      .its("status")
      .should("be.oneOf", [200, 201]);
    apiPost(customer, `/api/bookings/${b.id}/clearance/proceed`, { scheduledDate })
      .its("status")
      .should("be.oneOf", [200, 201]);
  });
  acceptOperation(suffix);
}

/** Ops accepts the operation request → FULLY_EXECUTED (enters the day pool). */
export function acceptOperation(suffix: string) {
  withBooking(suffix, (b) => {
    apiPost(opsStaff, `/api/bookings/${b.id}/operation/review`, { decision: "ACCEPT" })
      .its("status")
      .should("be.oneOf", [200, 201]);
  });
  pollBookingStatus(suffix, "FULLY_EXECUTED", 10);
}

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

/** Staff force-pay; polls PAID + SCHEDULED. */
export function markPaid(suffix: string) {
  withBooking(suffix, (b) => {
    apiPost(opsStaff, `/api/train-scheduling/bookings/${b.id}/mark-paid`)
      .its("status")
      .should("be.oneOf", [200, 201]);
  });
  pollDb<BookingRow>(
    `${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
     ORDER BY b.created_at DESC LIMIT 1`,
    [suffix],
    (row) => row?.status === "PAID" && row?.scheduling_status === "SCHEDULED",
    20,
  );
}

/**
 * Settle a reservation through the REAL payment pipeline: seed the gateway
 * intent projection (the payment microservice is absent in e2e), link it to
 * the open invoice, then deliver the `payment.succeeded` event to the public
 * internal endpoint. This drives billing settle → `booking.invoice.paid` →
 * `advanceBookingOnPayment` → `ensurePaidBookingAllocated`, which is the ONLY
 * path that applies a pending split offer (staff mark-paid skips it).
 */
export function settleViaGateway(suffix: string) {
  withBooking(suffix, (b) => {
    db<{ intent_id: string; currency: string; total: string }>(
      `WITH inv AS (
         SELECT id, currency, total_amount FROM freight.invoices
         WHERE source_id = $1 AND deleted_at IS NULL AND paid_at IS NULL
         ORDER BY created_at DESC LIMIT 1
       ), intent AS (
         INSERT INTO freight.payments
           (id, ref_id, type, reference_type, method, currency, amount,
            reason, raw_initiation, merchant_order_id, status)
         SELECT gen_random_uuid(), $1, 'FREIGHT', 'SHIPMENT',
                'telebirr'::freight.payments_method_enum,
                inv.currency::freight.payments_currency_enum, 1,
                'e2e gateway settle', '{}'::jsonb, 'E2E_' || $2,
                'processing'::freight.payments_status_enum
         FROM inv
         RETURNING id
       ), link AS (
         UPDATE freight.invoices SET payment_id = intent.id
         FROM intent WHERE freight.invoices.id = (SELECT id FROM inv)
         RETURNING payment_id
       )
       SELECT intent.id AS intent_id, inv.currency, inv.total_amount AS total
       FROM intent, inv`,
      [b.id, `${suffix}-${b.id.slice(0, 8)}`],
    ).then(({ rows }) => {
      expect(rows, `${suffix} gateway intent`).to.have.length(1);
      cy.request({
        method: "POST",
        url: `${apiUrl()}/api/internal/payments/mark-paid`,
        body: {
          version: 1,
          eventId: crypto.randomUUID(),
          eventType: "payment.succeeded",
          occurredAt: new Date().toISOString(),
          service: "FREIGHT",
          intentId: rows[0].intent_id,
          referenceType: "SHIPMENT",
          referenceId: b.id,
          merchantOrderId: `E2E_${suffix}_${b.id.slice(0, 8)}`,
          provider: "TELEBIRR",
          amountMinor: 1,
          currency: rows[0].currency,
        },
      }).then((res) => {
        expect(res.status, `${suffix} payment event accepted`).to.eq(200);
        // The global response interceptor wraps payloads in { success, data }.
        const raw = res.body as { processed?: boolean; data?: { processed?: boolean } };
        expect(raw.processed ?? raw.data?.processed, "event processed").to.eq(true);
      });
    });
  });
  pollDb<BookingRow>(
    `${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
     ORDER BY b.created_at DESC LIMIT 1`,
    [suffix],
    (row) => row?.status === "PAID",
    20,
  );
}

/** Push a reservation's pay deadline into the past — the 10s tick expires it. */
export function forceReservationExpiry(suffix: string) {
  withBooking(suffix, (b) =>
    db(
      `UPDATE freight.bookings SET payment_deadline = now() - interval '1 second'
       WHERE id = $1`,
      [b.id],
    ),
  );
  pollBookingStatus(suffix, "EXPIRED");
}

/**
 * Type-integrity assert for mixed trains: every wagon slot allocated to the
 * booking is of ONE expected wagon type (containers → NW5, bulk → CW4), and
 * the slot count matches. No wheat on a flat wagon, no box in a gondola.
 */
export function expectWagonType(suffix: string, code: string, wagons: number) {
  withBooking(suffix, (b) =>
    pollDb<{ code: string; n: string }>(
      `${suffix} rides ${wagons}× ${code}`,
      `SELECT wt.code, count(*) 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`,
      [b.id],
      (row) => row?.code === code && Number(row?.n) === wagons,
      20,
    ),
  );
  // A grouped second row would mean mixed wagon types under one booking.
  withBooking(suffix, (b) =>
    db<{ k: string }>(
      `SELECT count(DISTINCT tsw.wagon_type_id) 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`,
      [b.id],
    ).then(({ rows }) =>
      expect(Number(rows[0].k), `${suffix} single wagon type`).to.eq(1),
    ),
  );
}

export function pollAllocations(suffix: string, minWagons = 1) {
  withBooking(suffix, (b) =>
    pollDb<{ n: string }>(
      `${suffix} wagon allocations`,
      `SELECT count(*) AS n FROM freight.wagon_booking_allocations
       WHERE booking_id = $1 AND deleted_at IS NULL`,
      [b.id],
      (row) => Number(row?.n ?? 0) >= minWagons,
      30,
    ),
  );
}

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

export function dbRouteId(originCode = ORIGIN, destCode = DEST) {
  return 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],
  );
}

/** Create the 6-stop corridor route through the API if it doesn't exist yet. */
export function ensureCorridorRoute() {
  dbRouteId().then(({ rows }) => {
    if (rows.length > 0) return;
    db<{ id: string; code: string }>(
      `SELECT id, code FROM freight.yards WHERE code = ANY($1::text[])`,
      [[...CORRIDOR]],
    ).then(({ rows: yards }) => {
      expect(yards, "corridor yards").to.have.length(CORRIDOR.length);
      const byCode = new Map(yards.map((y) => [y.code, y.id]));
      apiPost(opsStaff, "/api/routes", {
        milestones: CORRIDOR.map((code) => ({ yardId: byCode.get(code) })),
      })
        .its("status")
        .should("be.oneOf", [200, 201]);
    });
    // Direction is frozen from the endpoint countries: DJ → ET = IMPORT.
    db<{ direction: string }>(
      `SELECT r.direction 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`,
      [ORIGIN, DEST],
    ).then(({ rows: created }) => {
      expect(created[0]?.direction, "corridor direction").to.eq("IMPORT");
    });
  });
}

/**
 * Make a corridor departure-day re-runnable: soft-delete any schedule a prior
 * run left on that day and expire its leftover fixture bookings (reference
 * scope CTR-IMP-% only — never touches other suites' data). A wiped schedule
 * must never leave bookings pointing at it (ghost refs break assign).
 */
export function resetCorridorDay(departure: Date, destCode = DEST, originCode = ORIGIN) {
  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'
       FROM freight.contracts ct
       WHERE ct.id = b.contract_id AND ct.reference LIKE 'CTR-IMP-%'
         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
     )
     UPDATE freight.train_schedules SET deleted_at = now()
     WHERE id IN (SELECT id FROM stale)`,
    [originCode, destCode, departure.toISOString()],
  );
  // Prior-run pool leftovers (never reserved, so no schedule ref) would
  // contaminate this run's batch — a stale high-priority booking steals the
  // top-up slot from this run's waiting list. Reset runs before this run
  // books anything, so every unpinned fixture booking is debris: expire all.
  db(
    `UPDATE freight.bookings b
     SET status = 'EXPIRED'
     FROM freight.contracts ct
     WHERE ct.id = b.contract_id AND ct.reference LIKE 'CTR-IMP-%'
       AND b.status = 'FULLY_EXECUTED' AND b.train_schedule_id IS NULL`,
    [],
  );
}

/** Create the reversed 6-stop corridor (ET → DJ = EXPORT) if missing. */
export function ensureExportRoute() {
  dbRouteId(EXP_ORIGIN, EXP_DEST).then(({ rows }) => {
    if (rows.length > 0) return;
    const stops = [...CORRIDOR].reverse();
    db<{ id: string; code: string }>(
      `SELECT id, code FROM freight.yards WHERE code = ANY($1::text[])`,
      [stops],
    ).then(({ rows: yards }) => {
      expect(yards, "corridor yards").to.have.length(stops.length);
      const byCode = new Map(yards.map((y) => [y.code, y.id]));
      apiPost(opsStaff, "/api/routes", {
        milestones: stops.map((code) => ({ yardId: byCode.get(code) })),
      })
        .its("status")
        .should("be.oneOf", [200, 201]);
    });
    db<{ direction: string }>(
      `SELECT r.direction 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`,
      [EXP_ORIGIN, EXP_DEST],
    ).then(({ rows: created }) => {
      expect(created[0]?.direction, "export corridor direction").to.eq("EXPORT");
    });
  });
}

/**
 * Ops accepts an EXPORT operation request — FCFS: the accept itself reserves
 * the train slot and opens a pay window clamped to the window close.
 */
export function acceptExport(suffix: string) {
  withBooking(suffix, (b) => {
    apiPost(opsStaff, `/api/bookings/${b.id}/operation/review`, { decision: "ACCEPT" })
      .its("status")
      .should("be.oneOf", [200, 201]);
  });
  pollBookingStatus(suffix, ["SELECTED_FOR_BATCH", "AWAITING_PAYMENT"], 10);
}

export interface ScheduleRow {
  id: string;
  status: string;
  window_phase: string;
  booking_window_status: string;
  booking_cycle_no: number;
  max_wagons: number;
  window_opens_at: string | null;
  window_closes_at: string | null;
  payment_phase_ends_at: string | null;
  scheduled_departure_date: string;
}

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

/** The corridor schedule departing within ±1h of `departure` (12:00 EAT pin). */
export function dbSchedule(departure: Date, destCode = DEST, originCode = ORIGIN) {
  return 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))) < 3600
     ORDER BY ts.created_at DESC LIMIT 1`,
    [originCode, destCode, departure.toISOString()],
  );
}

export function withSchedule(departure: Date, fn: (s: ScheduleRow) => void) {
  dbSchedule(departure).then(({ rows }) => {
    expect(rows, `schedule departing ${departure.toISOString()}`).to.have.length(1);
    fn(rows[0]);
  });
}

/** Export-corridor (KALITY → DJIB_PORT) variant of withSchedule. */
export function withExportSchedule(departure: Date, fn: (s: ScheduleRow) => void) {
  dbSchedule(departure, EXP_DEST, EXP_ORIGIN).then(({ rows }) => {
    expect(rows, `export schedule departing ${departure.toISOString()}`).to.have.length(1);
    fn(rows[0]);
  });
}

/**
 * Ops creates an import schedule on the corridor via the API — loco-pair mode
 * (no built train): capacity comes from maxWagonsPerTrain (54, the corridor
 * standard) and wagon stock is drawn from the origin yard at allocation time.
 */
export function createImportSchedule(opts: {
  departure: Date;
  locoPair: [string, string];
  maxWagons?: number;
  kind?: "container" | "bulk";
  originCode?: string;
  destCode?: string;
}) {
  const originCode = opts.originCode ?? ORIGIN;
  const destCode = opts.destCode ?? DEST;
  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);
      db<{ id: string }>(
        `SELECT id FROM freight.locomotives WHERE code = ANY($1::text[]) ORDER BY code`,
        [opts.locoPair],
      ).then(({ rows: locos }) => {
        expect(locos, `locomotives ${opts.locoPair.join(",")}`).to.have.length(2);
        apiPost(opsStaff, `/api/train-scheduling/${opts.kind ?? "container"}/schedules`, {
          routeId: routes[0].id,
          scheduleDate: opts.departure.toISOString(),
          locomotiveIds: locos.map((l) => l.id),
          maxWagonsPerTrain: opts.maxWagons ?? 54,
        })
          .its("status")
          .should("be.oneOf", [200, 201]);
      });
    });
  });
  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);
  });
}

// ---------------------------------------------------------------------------
// window choreography — arrange timestamps, let the engine do the transition
// ---------------------------------------------------------------------------

function pollSchedulePhase(
  scheduleId: string,
  want: string[],
  label: string,
  attempts = 40,
) {
  pollDb<ScheduleRow>(
    label,
    `SELECT window_phase, booking_window_status FROM freight.train_schedules WHERE id = $1`,
    [scheduleId],
    (row) => !!row && want.includes(row.window_phase as unknown as string),
    attempts,
  );
}

/** Pull the window-open moment into the past; the tick flips PRE_WINDOW→OPEN. */
export function forceWindowOpen(scheduleId: string, closesInMinutes = 45) {
  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)],
  );
  pollSchedulePhase(scheduleId, ["OPEN"], `schedule ${scheduleId} window OPEN`);
  pollDb<ScheduleRow>(
    `schedule ${scheduleId} bookable`,
    `SELECT booking_window_status FROM freight.train_schedules WHERE id = $1`,
    [scheduleId],
    (row) => row?.booking_window_status === "OPEN",
  );
}

/** Pull the close moment into the past; the tick flips OPEN→DOC_REVIEW. */
export function closeBookingWindow(scheduleId: string) {
  db(
    `UPDATE freight.train_schedules
     SET window_closes_at = now() - interval '1 second'
     WHERE id = $1 AND window_phase = 'OPEN'`,
    [scheduleId],
  );
  pollSchedulePhase(scheduleId, ["DOC_REVIEW"], `schedule ${scheduleId} DOC_REVIEW`);
}

/**
 * Staff end document review early → PAYMENT: expires never-accepted bookings,
 * runs the priority batch over the route-day pool, reserves + issues invoices.
 * (Lands on DONE instead when the batch reserved nobody.)
 */
export function completeDocReview(scheduleId: string) {
  apiPost(opsStaff, `/api/train-scheduling/schedules/${scheduleId}/doc-review-complete`)
    .its("status")
    .should("be.oneOf", [200, 201]);
  pollSchedulePhase(
    scheduleId,
    ["PAYMENT", "DONE", "PRE_WINDOW"],
    `schedule ${scheduleId} payment phase`,
  );
}

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

// ---------------------------------------------------------------------------
// train journey + clearance
// ---------------------------------------------------------------------------

export function recordCheckpoint(scheduleId: string, sequenceNo: number, kind: string) {
  apiPost(opsStaff, `/api/train-scheduling/schedules/${scheduleId}/checkpoints`, {
    sequenceNo,
    kind,
  })
    .its("status")
    .should("be.oneOf", [200, 201]);
}

/** GL multipart upload via the Node-side task (cy.request can't send files). */
export function glUpload(
  path: string,
  fields: Record<string, string> = {},
  fileField = "files",
) {
  return tokenFor(superAdmin).then((token) =>
    cy
      .task<{ status: number; body: unknown }>("api:upload", {
        url: `${apiUrl()}${path}`,
        token,
        fields,
        files: [{ field: fileField, fixture: "docs/license.pdf", filename: "e2e-doc.pdf" }],
      })
      .then((res) => {
        expect(res.status, `upload ${path}`).to.be.within(200, 201);
        return cy.wrap(res.body, { log: false });
      }),
  );
}

export function completeBookingMilestone(suffix: string, code: string) {
  withBooking(suffix, (b) => {
    apiPost(superAdmin, `/api/contracts/bookings/${b.id}/milestones/${code}/complete`, {
      note: "e2e",
    })
      .its("status")
      .should("be.oneOf", [200, 201]);
  });
}

export function expectMilestoneDone(suffix: string, code: string) {
  withBooking(suffix, (b) =>
    pollDb<{ n: string }>(
      `${suffix} milestone ${code}`,
      `SELECT count(*) AS n FROM freight.clearance_milestones
       WHERE booking_id = $1 AND milestone_code = $2 AND status = 'COMPLETED'
         AND deleted_at IS NULL`,
      [b.id, code],
      (row) => Number(row?.n ?? 0) > 0,
      10,
    ),
  );
}