/**
 * Many users, at once. Each test drives one production race through the real
 * HTTP surface and asserts the guard that is supposed to hold:
 *
 *   - two settlements of one invoice   → billing.markInvoiceAsPaid pessimistic lock
 *   - a replayed callback storm        → webhook dedupe on externalEventId
 *   - two tenants, one wagon budget    → reserveOnExport re-verify under lock (H8)
 *   - an invoice-number burst          → pg_advisory_xact_lock in invoice-numbering
 *   - pay after the window closed      → payInvoice dueAt gate
 *
 * These are the tests expected to find things. When one fails, read it as a
 * finding, not as a flaky assertion.
 */
import { afterAll, beforeAll, describe, expect, it } from "vitest";
import {
  closeDb,
  customerA,
  customerB,
  db,
  gateway,
  poll,
  sleep,
} from "./client";
import {
  TIN_B,
  createImportSchedule,
  currentInvoice,
  departureAt,
  ensureCorridorRoute,
  releaseUnpaidHolds,
  forceWindowOpen,
  gatewayIntent,
  invoiceForBooking,
  payInvoice,
  prepareBooking,
  resetCorridorDay,
  runBatch,
  type ReadyBooking,
} from "./flows";

const DEPARTURE = departureAt(8);
const STAMP = String(Date.now());

describe("concurrency and multi-tenant races", () => {
  let scheduleId: string;
  let a: ReadyBooking;
  let b: ReadyBooking;
  let invoiceA: string;
  let invoiceB: string;

  beforeAll(async () => {
    await gateway.reset();
    await releaseUnpaidHolds();
    await ensureCorridorRoute();
    await resetCorridorDay(DEPARTURE);
    const schedule = await createImportSchedule({ departure: DEPARTURE });
    scheduleId = schedule.id;
    await forceWindowOpen(scheduleId, 45);

    // Two DIFFERENT tenants on the same train-day.
    a = await prepareBooking({
      suffix: "CON-A",
      departure: DEPARTURE,
      runStamp: STAMP,
      isoSeed: 200,
      twenty: 2,
    });
    b = await prepareBooking({
      suffix: "CON-B",
      departure: DEPARTURE,
      runStamp: STAMP,
      isoSeed: 210,
      twenty: 2,
      tin: TIN_B,
      as: customerB,
    });
    await runBatch(scheduleId);
    invoiceA = (await invoiceForBooking(a.bookingId)).id;
    invoiceB = (await invoiceForBooking(b.bookingId)).id;
  }, 900_000);

  afterAll(closeDb);

  it("settles once when two callbacks land simultaneously", async () => {
    expect((await payInvoice(invoiceA, { method: "CBE_BIRR" })).status).toBeLessThanOrEqual(201);
    const intent = await gatewayIntent(a.bookingId);

    // Five concurrent deliveries of the SAME provider event.
    const results = await Promise.all(
      Array.from({ length: 5 }, () =>
        gateway.webhook({
          merchantOrderId: intent.merchant_order_id,
          eventId: `RACE-${intent.merchant_order_id}`,
        }),
      ),
    );
    expect(results.every((r) => r.body.delivered === 200)).toBe(true);

    await poll<{ status: string }>(
      "invoice PAID under duplicate delivery",
      `SELECT status FROM freight.invoices WHERE id = $1`,
      [invoiceA],
      (row) => row?.status === "PAID",
      { attempts: 30, intervalMs: 2000 },
    );
    await sleep(3000);

    // Dedupe is at the webhook table: one row, one outbox event, one ledger entry.
    const [{ n: events }] = await db<{ n: string }>(
      `SELECT count(*)::text AS n FROM edr_payment.payment_webhook_event
       WHERE merchant_order_id = $1`,
      [intent.merchant_order_id],
    );
    expect(Number(events)).toBe(1);

    const [{ n: outbox }] = await db<{ n: string }>(
      `SELECT count(*)::text AS n FROM edr_payment.notification_outbox
       WHERE intent_id = $1 AND event_type = 'payment.succeeded'`,
      [intent.id],
    );
    expect(Number(outbox)).toBe(1);

    const invoice = await currentInvoice(invoiceA);
    expect((invoice.payments as unknown[]).length).toBe(1);
    expect(Number(invoice.balance_amount)).toBe(0);
  });

  it("credits an invoice once even when two intents settle for it", async () => {
    // The per-reference unique index was dropped (migration 1782200000000), so
    // two intents on one booking are legal now. Both settling must still not
    // double-credit the invoice.
    const paid = await payInvoice(invoiceB, { method: "CBE_BIRR", as: customerB });
    expect(paid.status, JSON.stringify(paid.body)).toBeLessThanOrEqual(201);
    const first = await gatewayIntent(b.bookingId);

    // A second initiate for the same reference, different provider.
    const second = await payInvoice(invoiceB, { method: "TELEBIRR", as: customerB });
    expect(second.status).toBeLessThanOrEqual(201);

    await Promise.all([
      gateway.webhook({ merchantOrderId: first.merchant_order_id }),
      gateway.webhook({
        merchantOrderId: first.merchant_order_id,
        eventId: `SECOND-${first.merchant_order_id}`,
      }),
    ]);

    await poll<{ status: string }>(
      "invoice PAID once",
      `SELECT status FROM freight.invoices WHERE id = $1`,
      [invoiceB],
      (row) => row?.status === "PAID",
      { attempts: 30, intervalMs: 2000 },
    );
    await sleep(4000);

    const invoice = await currentInvoice(invoiceB);
    expect(Number(invoice.paid_amount)).toBeLessThanOrEqual(Number(invoice.total_amount));
    expect(Number(invoice.balance_amount)).toBe(0);
  });

  it("gives two tenants distinct, gapless invoice numbers under a burst", async () => {
    const numbers = await db<{ invoice_number: string }>(
      `SELECT invoice_number FROM freight.invoices
       WHERE created_at > now() - interval '30 minutes' AND deleted_at IS NULL`,
    );
    const seen = numbers.map((r) => r.invoice_number);
    expect(new Set(seen).size).toBe(seen.length);
  });

  it("never over-reserves the train when both tenants push at once", async () => {
    // The batch already ran for this day. Assert the invariant it must keep:
    // reserved wagons never exceed the consist.
    const [row] = await db<{ max_wagons: number; reserved: string }>(
      `SELECT ts.max_wagons,
              COALESCE(SUM(b.wagons_required), 0)::text AS reserved
       FROM freight.train_schedules ts
       LEFT JOIN freight.bookings b
         ON b.train_schedule_id = ts.id AND b.deleted_at IS NULL
        AND b.status NOT IN ('EXPIRED','CANCELLED','REJECTED')
       WHERE ts.id = $1
       GROUP BY ts.max_wagons`,
      [scheduleId],
    );
    expect(Number(row.reserved)).toBeLessThanOrEqual(Number(row.max_wagons));
  });

  it("rejects a fresh payment once the pay window has closed", async () => {
    // A booking whose deadline has passed must not be able to START a payment
    // (billing.payInvoice dueAt gate) — a payment begun BEFORE the deadline is
    // still honoured later by the expire-time gateway reconcile, which is why
    // the gate lives on initiation and not on settlement.
    const departure = departureAt(9);
    await releaseUnpaidHolds();
    await resetCorridorDay(departure);
    const schedule = await createImportSchedule({ departure });
    await forceWindowOpen(schedule.id, 45);
    const third = await prepareBooking({
      suffix: "CON-C",
      departure,
      runStamp: STAMP,
      isoSeed: 220,
      twenty: 2,
    });
    await runBatch(schedule.id);
    const invoice = await invoiceForBooking(third.bookingId);

    await db(`UPDATE freight.invoices SET due_at = now() - interval '1 minute' WHERE id = $1`, [
      invoice.id,
    ]);
    const res = await payInvoice(invoice.id, { method: "CBE_BIRR", as: customerA });
    expect(res.status).toBe(400);
    expect(JSON.stringify(res.body)).toMatch(/payment window/i);
  }, 600_000);
});