feat: integration tests

integration/src/concurrency.it.ts

@@ -0,0 +1,215 @@
+/**
+ * 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);
+});