Merge pull request #537 from Tria-plc/freight_feature/usermanagement

ensure pessimistic-lock writes run in a transaction for invoice proce…

apps/edr-freight-api/src/modules/billing/billing.service.spec.ts

@@ -21,7 +21,9 @@ function makeManager(savedLines: unknown[]) {
 }
 
 function makeEvents() {
-  return { emit: jest.fn() };
+  // BillingService emits via both emit() and emitAsync() (the post-commit async
+  // listener path) — the mock must provide both.
+  return { emit: jest.fn(), emitAsync: jest.fn().mockResolvedValue([]) };
 }
 
 function generateInput(overrides: Record<string, unknown> = {}) {
@@ -162,7 +164,7 @@ describe("BillingService.markInvoiceAsPaid", () => {
         ],
       },
     );
-    expect(events.emit).toHaveBeenCalledWith(
+    expect(events.emitAsync).toHaveBeenCalledWith(
       "booking.invoice.paid",
       expect.objectContaining({
         invoiceId: "inv-1",
@@ -197,6 +199,7 @@ describe("BillingService.markInvoiceAsPaid", () => {
 
     expect(mg.update).not.toHaveBeenCalled();
     expect(events.emit).not.toHaveBeenCalled();
+    expect(events.emitAsync).not.toHaveBeenCalled();
   });
 });
 
@@ -257,6 +260,7 @@ describe("BillingService.recordPayment", () => {
       }),
     );
     expect(events.emit).not.toHaveBeenCalled();
+    expect(events.emitAsync).not.toHaveBeenCalled();
   });
 
   it("settles to PAID, stamps paidAt, and emits ${source}.invoice.paid when the balance clears", async () => {
@@ -268,7 +272,7 @@ describe("BillingService.recordPayment", () => {
     expect(updated.balanceAmount).toBe(0);
     expect(updated.paidAt).toBeInstanceOf(Date);
     expect(mg.update).toHaveBeenCalled();
-    expect(events.emit).toHaveBeenCalledWith(
+    expect(events.emitAsync).toHaveBeenCalledWith(
       "warehouse.invoice.paid",
       expect.objectContaining({ invoiceId: "inv-1", status: Freight.InvoiceStatus.Paid }),
     );
@@ -296,3 +300,80 @@ describe("BillingService.recordPayment", () => {
     expect(mg.update).not.toHaveBeenCalled();
   });
 });
+
+/**
+ * Regression: `expirePayable` (batch settle path, called when a payment window
+ * lapses) transitions the invoice to EXPIRED, which locks the row FOR UPDATE.
+ * The bug passed `dataSource.manager` (the non-transactional default) into the
+ * transition, so runTransition skipped opening a transaction and the lock threw
+ * `An open transaction is required for pessimistic lock` — aborting the whole
+ * settle pass (the "settle/reserve one booking at a time" symptom). The locked
+ * write MUST run inside dataSource.transaction.
+ */
+describe("BillingService.expirePayable — locked write runs in a transaction", () => {
+  const openInvoice = {
+    id: "inv-1",
+    status: Freight.InvoiceStatus.Pending,
+    source: "booking",
+    sourceId: "booking-1",
+  };
+
+  const build = (lookupResult: Record<string, unknown> | null) => {
+    const defaultManager = {
+      findOne: jest.fn().mockResolvedValue(lookupResult),
+      update: jest.fn().mockResolvedValue(undefined),
+    };
+    const txManager = {
+      findOne: jest.fn().mockResolvedValue(openInvoice),
+      update: jest.fn().mockResolvedValue(undefined),
+    };
+    const transaction = jest
+      .fn()
+      .mockImplementation((cb: (mg: unknown) => unknown) => cb(txManager));
+    const events = makeEvents();
+    const service = new BillingService(
+      { manager: defaultManager, transaction } as never,
+      {} as never,
+      {} as never,
+      events as never,
+      {} as never,
+      {} as never,
+      {} as never,
+    );
+    return { service, defaultManager, txManager, transaction };
+  };
+
+  it("opens a transaction and runs the pessimistic-lock read on the tx manager", async () => {
+    const { service, transaction, txManager, defaultManager } = build(openInvoice);
+
+    await service.expirePayable(
+      Freight.InvoiceSource.Booking,
+      "booking-1",
+      "prepaid",
+    );
+
+    expect(transaction).toHaveBeenCalledTimes(1);
+    expect(txManager.findOne).toHaveBeenCalledWith(
+      expect.anything(),
+      expect.objectContaining({ lock: { mode: "pessimistic_write" } }),
+    );
+    expect(txManager.update).toHaveBeenCalled();
+    // The default manager only does the initial lock-free lookup, never a locked read.
+    for (const call of defaultManager.findOne.mock.calls) {
+      expect(call[1]).not.toHaveProperty("lock");
+    }
+  });
+
+  it("is a no-op (no transaction) when there is no open invoice", async () => {
+    const { service, transaction } = build(null);
+
+    const result = await service.expirePayable(
+      Freight.InvoiceSource.Booking,
+      "booking-1",
+      "prepaid",
+    );
+
+    expect(result).toBeNull();
+    expect(transaction).not.toHaveBeenCalled();
+  });
+});

apps/edr-freight-api/src/modules/billing/billing.service.ts

@@ -825,6 +825,13 @@ export class BillingService {
     type?: string,
     manager?: EntityManager,
   ): Promise<Invoice | null> {
+    // Lookup can use the default manager (no lock). But the pessimistic-lock write
+    // inside `transition` NEEDS an open transaction: pass the caller's `manager`
+    // through untouched (undefined when there is no caller txn) so `runTransition`
+    // opens its own. Passing `this.dataSource.manager` here made `runTransition`
+    // treat it as an already-open transaction and skip wrapping — the lock then
+    // threw `An open transaction is required for pessimistic lock`, aborting the
+    // whole settle pass (the "reservations settle/reserve one at a time" symptom).
     const mg = manager ?? this.dataSource.manager;
     const invoice = await mg.findOne(Invoice, {
       where: {
@@ -842,7 +849,7 @@ export class BillingService {
       Freight.InvoiceStatus.Expired,
       "expired",
       {},
-      mg,
+      manager,
     );
   }