import { Freight, PaymentReferenceType } from "@edr/types";
import {
  BadRequestException,
  forwardRef,
  Inject,
  Injectable,
  Logger,
  NotFoundException,
} from "@nestjs/common";
import { EventEmitter2 } from "@nestjs/event-emitter";
import { DataSource, EntityManager, In } from "typeorm";

import { CompaniesService } from "../companies/companies.service";
import { PaymentService } from "../payment/payment.service";
import { InitiateResponseDto } from "../payment/payments.dto";
import {
  InvoiceDocumentModel,
  InvoiceDocumentService,
} from "./documents/invoice-document.service";
import { InvoiceLine } from "./entities/invoice-line.entity";
import { Invoice, InvoicePayment } from "./entities/invoice.entity";
import { InvoiceLineRepository } from "./invoice-line.repository";
import { nextDailyInvoiceNumber } from "./invoice-numbering.util";
import { applySettlement, round2 } from "./invoice-settlement.util";
import { InvoiceRepository } from "./invoice.repository";

/** Options forwarded to the payment gateway when settling an invoice. */
export interface PayInvoiceOptions {
  method?: string;
  platform?: "web" | "mobile";
  payerAccount?: string;
  returnUrl?: string;
  failureUrl?: string;
}

/** A single manual/offline settlement to record against an invoice. */
export interface RecordPaymentInput {
  /** Amount settled by this payment; must be > 0. */
  amount: number;
  method?: string | null;
  reference?: string | null;
  /** When the settlement occurred; defaults to now. */
  paidAt?: Date;
  metadata?: Record<string, unknown> | null;
}

/** Default invoice payment-term window, in days, used to compute `dueAt`. */
const DEFAULT_DUE_DAYS = 14;

/** Statuses an invoice can still be settled (paid/refunded/cancelled) from. */
const OPEN_STATUSES: Freight.InvoiceStatus[] = [
  Freight.InvoiceStatus.Issued,
  Freight.InvoiceStatus.Pending,
  Freight.InvoiceStatus.PartiallyPaid,
  Freight.InvoiceStatus.Overdue,
];

/** A single line to bill on a generated invoice. */
export interface InvoiceLineInput {
  chargeType: string;
  description?: string;
  /** Units this line bills for; defaults to 1. */
  quantity?: number;
  /** Price per unit; defaults to 0. */
  unitRate?: number;
  /** Line total; defaults to `quantity * unitRate`. */
  amount?: number;
  currency?: string;
  metadata?: Record<string, unknown> | null;
}

/** Everything needed to generate an invoice for any source. */
export interface GenerateInvoiceInput {
  /** Originating subsystem; namespaces events (`${source}.invoice.<event>`). */
  source: Freight.InvoiceSource;
  /** Identifier of the source record (e.g. booking id). */
  sourceId: string;
  /** What the invoice is for (e.g. "prepaid", "credit"). */
  type: string;
  companyId: string;
  companyProfileId: string;
  lines: InvoiceLineInput[];
  currency?: string;
  /** Explicit pre-tax subtotal; defaults to the sum of line amounts. */
  subtotalAmount?: number;
  /** Tax applied on top of the subtotal; defaults to 0. */
  taxAmount?: number;
  /** Explicit total; defaults to `subtotalAmount + taxAmount`. */
  totalAmount?: number;
  /** Issue date window; defaults to `DEFAULT_DUE_DAYS` from now. */
  dueAt?: Date;
  dueInDays?: number;
  /**
   * Initial status. DRAFT leaves `issuedAt` null; any issued status
   * (default PENDING) stamps `issuedAt`.
   */
  status?: Freight.InvoiceStatus;
}

/** Payload broadcast on `${source}.invoice.<event>`. */
export interface InvoiceEventPayload {
  invoiceId: string;
  invoiceNumber: string;
  source: Freight.InvoiceSource;
  sourceId: string;
  type: string;
  companyId: string;
  companyProfileId: string;
  totalAmount: number;
  currency: string;
  status: Freight.InvoiceStatus;
  paymentId?: string | null;
}

@Injectable()
export class BillingService {
  private readonly logger = new Logger(BillingService.name);

  constructor(
    private readonly dataSource: DataSource,
    private readonly invoices: InvoiceRepository,
    private readonly invoiceLines: InvoiceLineRepository,
    private readonly events: EventEmitter2,
    @Inject(forwardRef(() => PaymentService))
    private readonly payment: PaymentService,
    private readonly companies: CompaniesService,
    private readonly invoiceDocuments: InvoiceDocumentService,
  ) { }

  // ── Reads ──────────────────────────────────────────────────────────────────

  /** List every invoice (most recent first). */
  findAll(): Promise<Invoice[]> {
    return this.invoices.findAll({ order: { issuedAt: "DESC" } });
  }

  /**
   * Paginated invoice list for the backoffice — optionally narrowed to a
   * company (customer detail "Invoices" tab) and/or status/search (global
   * invoices page).
   */
  async findAllPaginated(
    filter: {
      companyId?: string;
      status?: Freight.InvoiceStatus;
      search?: string;
      page?: number;
      pageSize?: number;
    } = {},
  ): Promise<{ items: Invoice[]; total: number }> {
    const page = filter.page && filter.page > 0 ? filter.page : 1;
    const pageSize =
      filter.pageSize && filter.pageSize > 0 ? filter.pageSize : 20;

    const qb = this.dataSource
      .getRepository(Invoice)
      .createQueryBuilder("invoice")
      .leftJoinAndSelect("invoice.company", "company")
      .orderBy("invoice.issuedAt", "DESC")
      .skip((page - 1) * pageSize)
      .take(pageSize);

    if (filter.companyId) {
      qb.andWhere("invoice.companyId = :companyId", {
        companyId: filter.companyId,
      });
    }
    if (filter.status) {
      qb.andWhere("invoice.status = :status", { status: filter.status });
    }
    if (filter.search) {
      qb.andWhere(
        "(invoice.invoiceNumber ILIKE :search OR invoice.sourceId ILIKE :search)",
        { search: `%${filter.search}%` },
      );
    }

    const [items, total] = await qb.getManyAndCount();
    return { items, total };
  }

  /** Invoice header plus its line items. */
  async findById(id: string): Promise<Invoice & { lines: InvoiceLine[] }> {
    const invoice = await this.invoices.findById(id, {
      relations: { company: true, companyProfile: true },
    });
    if (!invoice) throw new NotFoundException(`Invoice ${id} not found`);
    const lines = await this.invoiceLines.findAll({
      where: { invoiceId: id },
      order: { createdAt: "ASC" },
    });
    return { ...invoice, lines } as Invoice & { lines: InvoiceLine[] };
  }

  // ── Documents (central PDF) ──────────────────────────────────────────────────

  /** Sealed PDF invoice for any source, rendered by the shared document service. */
  async document(id: string): Promise<{ filename: string; buffer: Buffer }> {
    const invoice = await this.findById(id);
    return this.invoiceDocuments.render(
      this.toDocumentModel(invoice, "INVOICE"),
    );
  }

  /** Sealed PDF receipt; available once any payment has been recorded. */
  async receipt(id: string): Promise<{ filename: string; buffer: Buffer }> {
    const invoice = await this.findById(id);
    if (Number(invoice.paidAmount) <= 0) {
      throw new BadRequestException(
        "A receipt is available only after payment is recorded.",
      );
    }
    return this.invoiceDocuments.render(
      this.toDocumentModel(invoice, "RECEIPT"),
    );
  }

  /** Map a global invoice (+ lines) onto the source-agnostic document model. */
  private toDocumentModel(
    invoice: Invoice & { lines: InvoiceLine[] },
    kind: "INVOICE" | "RECEIPT",
  ): InvoiceDocumentModel {
    const title = invoice.source
      ? invoice.source.charAt(0).toUpperCase() + invoice.source.slice(1)
      : "EDR";
    const totals: InvoiceDocumentModel["totals"] = [
      { label: "Subtotal", amount: Number(invoice.subtotalAmount) },
    ];
    if (Number(invoice.taxAmount) > 0) {
      totals.push({ label: "Tax", amount: Number(invoice.taxAmount) });
    }
    totals.push({
      label: "Total",
      amount: Number(invoice.totalAmount),
      grand: true,
    });
    totals.push({ label: "Paid", amount: Number(invoice.paidAmount) });
    totals.push({ label: "Balance", amount: Number(invoice.balanceAmount) });

    return {
      kind,
      title,
      documentNumber: invoice.invoiceNumber,
      issuedAt: invoice.issuedAt ?? invoice.createdAt,
      status: invoice.status,
      currency: invoice.currency,
      summary: [
        { label: "Status", value: invoice.status },
        { label: "Type", value: invoice.type },
        { label: "Reference", value: invoice.sourceId },
        { label: "Currency", value: invoice.currency },
        {
          label: "Issued",
          value: invoice.issuedAt
            ? new Date(invoice.issuedAt).toLocaleDateString("en-GB")
            : null,
        },
        {
          label: "Due",
          value: invoice.dueAt
            ? new Date(invoice.dueAt).toLocaleDateString("en-GB")
            : null,
        },
      ],
      categoryHeader: "Charge type",
      lines: invoice.lines.map((l) => ({
        description: l.description ?? l.chargeType,
        category: l.chargeType,
        quantity: l.quantity,
        unitRate: l.unitRate,
        amount: l.amount,
        currency: l.currency,
      })),
      totals,
    };
  }

  // ── Customer-scoped reads (portal) ───────────────────────────────────────────

  /** Resolve the customer's company id from their IAM user id (null if none). */
  async resolveCompanyId(userId: string): Promise<string | null> {
    try {
      const { company } = await this.companies.getCompanyInfoByUserId(userId);
      return company?.id ?? null;
    } catch {
      return null;
    }
  }

  /**
   * Every invoice billed to a company, newest first, with billing relations.
   * Optionally narrow to a single source record (e.g. a booking's invoices) via
   * `{ source, sourceId }`.
   */
  findByCompany(
    companyId: string,
    filter: { source?: string; sourceId?: string } = {},
  ): Promise<Invoice[]> {
    return this.invoices.findAll({
      where: {
        companyId,
        ...(filter.source ? { source: filter.source } : {}),
        ...(filter.sourceId ? { sourceId: filter.sourceId } : {}),
      },
      relations: { company: true, companyProfile: true },
      order: { createdAt: "DESC" },
    });
  }

  /** Invoices for a batch of source records (e.g. many last-mile legs), so a
   *  list can show which records already have an invoice without N+1 queries. */
  findBySourceIds(source: string, sourceIds: string[]): Promise<Invoice[]> {
    if (!sourceIds.length) return Promise.resolve([]);
    return this.invoices.findAll({
      where: { source, sourceId: In(sourceIds) },
      order: { createdAt: "DESC" },
    });
  }

  /** Invoices for the signed-in customer; empty when they have no company. */
  async findForUser(
    userId: string,
    filter: { source?: string; sourceId?: string } = {},
  ): Promise<Invoice[]> {
    const companyId = await this.resolveCompanyId(userId);
    return companyId ? this.findByCompany(companyId, filter) : [];
  }

  /** Company-scoped invoice detail (+ lines); 404 when not owned by the user. */
  async findByIdForUser(
    id: string,
    userId: string,
  ): Promise<Invoice & { lines: InvoiceLine[] }> {
    const companyId = await this.resolveCompanyId(userId);
    const invoice = await this.findById(id);
    if (!companyId || invoice.companyId !== companyId) {
      throw new NotFoundException(`Invoice ${id} not found`);
    }
    return invoice;
  }

  /**
   * Initiate gateway payment for one of the customer's own invoices. Verifies
   * ownership, then charges the invoice directly by ID (see {@link payInvoice}).
   */
  async payInvoiceForUser(
    id: string,
    userId: string,
    opts: PayInvoiceOptions = {},
  ): Promise<InitiateResponseDto> {
    await this.findByIdForUser(id, userId);
    return this.payInvoice(id, opts);
  }

  /** Sealed invoice PDF for one of the customer's own invoices (ownership-checked). */
  async documentForUser(
    id: string,
    userId: string,
  ): Promise<{ filename: string; buffer: Buffer }> {
    await this.findByIdForUser(id, userId);
    return this.document(id);
  }

  /** Sealed receipt PDF for one of the customer's own invoices (ownership-checked). */
  async receiptForUser(
    id: string,
    userId: string,
  ): Promise<{ filename: string; buffer: Buffer }> {
    await this.findByIdForUser(id, userId);
    return this.receipt(id);
  }

  // ── Generation ───────────────────────────────────────────────────────────────

  /** `<CODE>-YYYYMMDD-00001` — sequential per day & prefix, within the active transaction. */
  private nextInvoiceNumber(mg: EntityManager): Promise<string> {
    return nextDailyInvoiceNumber(mg, {
      table: "freight.invoices",
      code: "INV",
    });
  }

  /**
   * Generate an invoice for any source (booking, demurrage, manual, …).
   *
   * Persists the header plus its lines in one transaction and assigns the next
   * sequential `invoice_number`. The total defaults to the sum of line amounts
   * unless `totalAmount` is given. Issued invoices (default PENDING) stamp
   * `issuedAt`; pass `status: DRAFT` to leave it unissued.
   *
   * Pass `manager` to enlist in a caller's transaction (e.g. when generating an
   * invoice as part of a larger booking flow).
   */
  async generateInvoice(
    input: GenerateInvoiceInput,
    manager?: EntityManager,
  ): Promise<Invoice & { lines: InvoiceLine[] }> {
    console.log("oooooooooo", input);
    const run = (mg: EntityManager) => this.createInvoice(input, mg);
    return manager ? run(manager) : this.dataSource.transaction(run);
  }

  private async createInvoice(
    input: GenerateInvoiceInput,
    mg: EntityManager,
  ): Promise<Invoice & { lines: InvoiceLine[] }> {
    const currency = input.currency ?? "ETB";
    const status = input.status ?? Freight.InvoiceStatus.Pending;
    const issued = status !== Freight.InvoiceStatus.Draft;

    const lines = input.lines.map((l) => {
      const quantity = l.quantity ?? 1;
      const unitRate = l.unitRate ?? 0;
      return {
        chargeType: l.chargeType,
        description: l.description,
        quantity,
        unitRate,
        amount: l.amount ?? quantity * unitRate,
        currency: l.currency ?? currency,
        metadata: l.metadata ?? null,
      };
    });

    const subtotalAmount =
      input.subtotalAmount ??
      lines.reduce((sum, l) => sum + Number(l.amount), 0);
    const taxAmount = input.taxAmount ?? 0;
    const totalAmount = input.totalAmount ?? round2(subtotalAmount + taxAmount);

    const dueAt =
      input.dueAt ??
      new Date(
        Date.now() +
        (input.dueInDays ?? DEFAULT_DUE_DAYS) * 24 * 60 * 60 * 1000,
      );

    const invoiceNumber = await this.nextInvoiceNumber(mg);

    const invoice = await mg.save(
      mg.create(Invoice, {
        invoiceNumber,
        source: input.source,
        sourceId: input.sourceId,
        type: input.type,
        companyId: input.companyId,
        companyProfileId: input.companyProfileId,
        subtotalAmount: round2(subtotalAmount),
        taxAmount: round2(taxAmount),
        totalAmount: round2(totalAmount),
        paidAmount: 0,
        balanceAmount: round2(totalAmount),
        payments: [],
        currency,
        status,
        issuedAt: issued ? new Date() : null,
        dueAt,
      }),
    );

    const savedLines = await Promise.all(
      lines.map((l) =>
        mg.save(mg.create(InvoiceLine, { ...l, invoiceId: invoice.id })),
      ),
    );

    this.logger.log(
      `Generated invoice ${invoice.invoiceNumber} (${invoice.id}) for ${input.source}:${input.sourceId}`,
    );

    return { ...invoice, lines: savedLines };
  }

  // ── State transitions ────────────────────────────────────────────────────────

  /**
   * Run `fn` inside a transaction and only emit its returned domain event
   * after commit. When the caller passes their own `manager`, they own commit
   * timing — `fn`'s event fires inline as soon as it resolves (the outer
   * transaction may still roll back afterwards; this is the caller's
   * documented tradeoff). When no `manager` is given, this opens its own
   * transaction and defers the emit until after that transaction commits, so
   * listeners (e.g. booking advancement) can never observe an invoice change
   * that then rolls back.
   */
  private async runTransition<T>(
    manager: EntityManager | undefined,
    fn: (mg: EntityManager) => Promise<{ result: T; emit?: () => void }>,
  ): Promise<T> {
    if (manager) {
      const { result, emit } = await fn(manager);
      emit?.();
      return result;
    }
    let pending: (() => void) | undefined;
    const result = await this.dataSource.transaction(async (mg) => {
      const out = await fn(mg);
      pending = out.emit;
      return out.result;
    });
    pending?.();
    return result;
  }

  /**
   * Mark an invoice paid, stamp the paid timestamp, sync paid/balance amounts,
   * append the settlement to the `payments` ledger, link the gateway payment,
   * then emit `${source}.invoice.paid`. Full-payment only — no partial
   * settlement. No-op when the invoice is already paid. Pass `manager` to
   * enlist in a caller's transaction; otherwise locks the row for update and
   * emits only after commit (see {@link runTransition}).
   */
  async markInvoiceAsPaid(
    invoiceId: string,
    paymentId: string | null = null,
    manager?: EntityManager,
    settlement: { providerTxnId?: string; paidAt?: Date } = {},
  ): Promise<Invoice | null> {
    return this.runTransition(manager, async (mg) => {
      const invoice = await mg.findOne(Invoice, {
        where: { id: invoiceId },
        lock: { mode: "pessimistic_write" },
      });
      if (!invoice) {
        throw new NotFoundException(`Invoice ${invoiceId} not found`);
      }
      if (invoice.status === Freight.InvoiceStatus.Paid) {
        return { result: invoice };
      }

      const paidAt = invoice.paidAt ?? settlement.paidAt ?? new Date();
      const settledAmount = round2(
        Number(invoice.totalAmount) - Number(invoice.paidAmount ?? 0),
      );
      const entry: InvoicePayment = {
        amount: settledAmount,
        method: "GATEWAY",
        reference: settlement.providerTxnId ?? paymentId ?? null,
        paidAt: paidAt.toISOString(),
        metadata: null,
      };
      const payments = [...(invoice.payments ?? []), entry];

      const patch = {
        status: Freight.InvoiceStatus.Paid,
        paymentId,
        paidAt,
        paidAmount: invoice.totalAmount,
        balanceAmount: 0,
        payments,
      };
      await mg.update(Invoice, { id: invoiceId }, patch as never);

      const updated = { ...invoice, ...patch } as Invoice;
      return {
        result: updated,
        emit: () => this.emitInvoiceEvent("paid", updated),
      };
    });
  }

  /**
   * Record a (possibly partial) settlement against an invoice and sync its
   * status. Appends to the `payments` ledger, recomputes `paidAmount` /
   * `balanceAmount`, and moves the invoice to PARTIALLY_PAID or — once the
   * balance reaches zero — PAID, stamping `paidAt` and emitting
   * `${source}.invoice.paid`. Use this for manual/offline settlement (e.g. cash
   * at the warehouse counter); gateway settlement goes through
   * {@link markInvoiceAsPaid}.
   *
   * Throws when the invoice is missing, cancelled, refunded, already fully
   * paid, `amount` is not positive, or `amount` exceeds the outstanding
   * balance. Pass `manager` to enlist in a caller's transaction; otherwise
   * locks the row for update and emits only after commit (see
   * {@link runTransition}).
   */
  async recordPayment(
    invoiceId: string,
    input: RecordPaymentInput,
    manager?: EntityManager,
  ): Promise<Invoice> {
    if (!(input.amount > 0)) {
      throw new BadRequestException(
        "Payment amount must be greater than zero.",
      );
    }

    return this.runTransition(manager, async (mg) => {
      const invoice = await mg.findOne(Invoice, {
        where: { id: invoiceId },
        lock: { mode: "pessimistic_write" },
      });
      if (!invoice) {
        throw new NotFoundException(`Invoice ${invoiceId} not found`);
      }
      if (invoice.status === Freight.InvoiceStatus.Cancelled) {
        throw new BadRequestException("Cannot pay a cancelled invoice.");
      }
      if (invoice.status === Freight.InvoiceStatus.Refunded) {
        throw new BadRequestException("Cannot pay a refunded invoice.");
      }
      if (invoice.status === Freight.InvoiceStatus.Paid) {
        throw new BadRequestException("Invoice is already fully paid.");
      }
      // M27: a Draft invoice is not yet issued and an Expired invoice's pay
      // window has closed — neither is payable. Without these guards a payment
      // could settle an unissued draft or a lapsed invoice.
      if (invoice.status === Freight.InvoiceStatus.Draft) {
        throw new BadRequestException(
          "Cannot pay a draft invoice — it must be issued first.",
        );
      }
      if (invoice.status === Freight.InvoiceStatus.Expired) {
        throw new BadRequestException(
          "Cannot pay an expired invoice — its payment window has closed.",
        );
      }
      if (round2(input.amount) > Number(invoice.balanceAmount)) {
        throw new BadRequestException(
          `Payment of ${round2(input.amount)} exceeds the outstanding balance of ${Number(invoice.balanceAmount)}.`,
        );
      }

      const at = input.paidAt ?? new Date();
      const { paidAmount, balanceAmount, fullyPaid } = applySettlement(
        invoice.totalAmount,
        invoice.paidAmount,
        input.amount,
      );
      const status = fullyPaid
        ? Freight.InvoiceStatus.Paid
        : Freight.InvoiceStatus.PartiallyPaid;

      const entry: InvoicePayment = {
        amount: round2(input.amount),
        method: input.method ?? null,
        reference: input.reference ?? null,
        paidAt: at.toISOString(),
        metadata: input.metadata ?? null,
      };
      const payments = [...(invoice.payments ?? []), entry];

      const patch = {
        paidAmount,
        balanceAmount,
        status,
        payments,
        paidAt: fullyPaid ? at : (invoice.paidAt ?? null),
      };
      await mg.update(Invoice, { id: invoice.id }, patch as never);

      const updated = { ...invoice, ...patch } as Invoice;
      return {
        result: updated,
        emit: fullyPaid
          ? () => this.emitInvoiceEvent("paid", updated)
          : undefined,
      };
    });
  }

  /**
   * Mark an invoice refunded and emit `${source}.invoice.refunded`.
   * No-op when already refunded. Throws when the invoice has no recorded
   * payment (nothing to refund).
   */
  async markInvoiceAsRefunded(
    invoiceId: string,
    manager?: EntityManager,
  ): Promise<Invoice | null> {
    return this.transition(
      invoiceId,
      Freight.InvoiceStatus.Refunded,
      "refunded",
      {},
      manager,
      (invoice) => {
        if (!(Number(invoice.paidAmount) > 0)) {
          throw new BadRequestException(
            "Cannot refund an invoice with no recorded payment.",
          );
        }
      },
    );
  }

  /**
   * Mark an invoice cancelled and emit `${source}.invoice.cancelled`.
   * No-op when already cancelled. Throws when the invoice has payments
   * recorded against it (refund it instead).
   */
  async cancelInvoice(
    invoiceId: string,
    manager?: EntityManager,
  ): Promise<Invoice | null> {
    return this.transition(
      invoiceId,
      Freight.InvoiceStatus.Cancelled,
      "cancelled",
      {},
      manager,
      (invoice) => {
        if (Number(invoice.paidAmount) > 0) {
          throw new BadRequestException(
            "Cannot cancel an invoice that has payments recorded against it.",
          );
        }
      },
    );
  }

  /**
   * Load the invoice, apply the new status (+ extra columns), then emit
   * `${source}.invoice.<event>`. No-op (returns the invoice, skipping `guard`)
   * when it is already in the target status. Throws when the invoice does not
   * exist or `guard` rejects the current state. Pass `manager` to enlist in a
   * caller's transaction; otherwise locks the row for update and emits only
   * after commit (see {@link runTransition}).
   */
  private async transition(
    invoiceId: string,
    status: Freight.InvoiceStatus,
    event: string,
    extra: { paymentId?: string },
    manager?: EntityManager,
    guard?: (invoice: Invoice) => void,
  ): Promise<Invoice | null> {
    return this.runTransition(manager, async (mg) => {
      const invoice = await mg.findOne(Invoice, {
        where: { id: invoiceId },
        lock: { mode: "pessimistic_write" },
      });
      if (!invoice) {
        throw new NotFoundException(`Invoice ${invoiceId} not found`);
      }
      if (invoice.status === status) return { result: invoice };
      guard?.(invoice);

      await mg.update(Invoice, { id: invoice.id }, { status, ...extra });

      const updated = { ...invoice, ...extra, status } as Invoice;
      return {
        result: updated,
        emit: () => this.emitInvoiceEvent(event, updated),
      };
    });
  }

  /** Broadcast `${invoice.source}.invoice.<event>` to in-process listeners. */
  private emitInvoiceEvent(event: string, invoice: Invoice): void {
    const payload: InvoiceEventPayload = {
      invoiceId: invoice.id,
      invoiceNumber: invoice.invoiceNumber,
      source: invoice.source as Freight.InvoiceSource,
      sourceId: invoice.sourceId,
      type: invoice.type,
      companyId: invoice.companyId,
      companyProfileId: invoice.companyProfileId,
      totalAmount: invoice.totalAmount,
      currency: invoice.currency,
      status: invoice.status,
      paymentId: invoice.paymentId ?? null,
    };
    this.events
      .emitAsync(`${invoice.source}.invoice.${event}`, payload)
      .catch((err) =>
        this.logger.error(
          `Listener for ${invoice.source}.invoice.${event} (invoice ${invoice.id}) failed: ${err instanceof Error ? err.message : String(err)}`,
        ),
      );
  }

  // ── Payment reconciliation (by source) ───────────────────────────────────────

  /**
   * The invoice a source record already has open, or null if it needs a new
   * one. This is the idempotency check every `ensureInvoiceFor*` (booking,
   * first-mile, last-mile) runs before generating — it must see DRAFT
   * invoices too, not just issued ones, otherwise a source that already has
   * an unissued draft gets a second, duplicate invoice minted alongside it
   * instead of that draft being reused and then issued.
   *
   * Pass `type` to select a specific invoice when a source carries several (e.g.
   * a booking's up-front vs final charge); omit it to settle whichever single
   * invoice is currently open. Returns the most recent matching draft-or-open
   * (unpaid, non-cancelled) invoice.
   */
  findPayable(
    source: Freight.InvoiceSource,
    sourceId: string,
    type?: string,
  ): Promise<Invoice | null> {
    return this.dataSource.getRepository(Invoice).findOne({
      where: {
        source,
        sourceId,
        status: In([Freight.InvoiceStatus.Draft, ...OPEN_STATUSES]),
        ...(type ? { type } : {}),
      },
      order: { issuedAt: "DESC" },
    });
  }

  /**
   * Pass `type` to select a specific invoice when a source carries several (e.g.
   * a booking's up-front vs final charge); omit it to settle whichever single
   * invoice is currently open. Returns the most recent matching open (unpaid,
   * non-cancelled) invoice.
   */
  findInvoice(
    source: Freight.InvoiceSource,
    sourceId: string,
    type?: string,
  ): Promise<Invoice | null> {
    return this.dataSource.getRepository(Invoice).findOne({
      where: {
        source,
        sourceId,
        ...(type ? { type } : {}),
      },
      order: { issuedAt: "DESC" },
    });
  }

  /**
   * Expire a source's currently-open invoice (its pay window closed before
   * settlement), then emit `${source}.invoice.expired`. Resolves the open invoice
   * and transitions it to EXPIRED — a terminal, non-payable status (kept out of
   * `OPEN_STATUSES`). No-op (returns null) when the source has no open invoice
   * (already paid/cancelled/expired).
   *
   * Pass the caller's transaction `manager` (e.g. the booking pay-window expiry in
   * the batch engine) to enlist in its DB transaction.
   */
  async expirePayable(
    source: Freight.InvoiceSource,
    sourceId: string,
    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: {
        source,
        sourceId,
        status: In(OPEN_STATUSES),
        ...(type ? { type } : {}),
      },
      order: { issuedAt: "DESC" },
    });
    if (!invoice) return null;

    return this.transition(
      invoice.id,
      Freight.InvoiceStatus.Expired,
      "expired",
      {},
      manager,
    );
  }

  /**
   * Sync a source's open invoice `dueAt` to its real pay-window deadline. The
   * booking invoice is generated before the pay window opens (at booking
   * creation/approval), so its printed due date is refreshed when the batch engine
   * sets `paymentDeadline`. No-op when the source has no open invoice.
   */
  async syncPayableDueDate(
    source: Freight.InvoiceSource,
    sourceId: string,
    dueAt: Date,
    type?: string,
    manager?: EntityManager,
  ): Promise<void> {
    const mg = manager ?? this.dataSource.manager;
    const invoice = await mg.findOne(Invoice, {
      where: {
        source,
        sourceId,
        status: In(OPEN_STATUSES),
        ...(type ? { type } : {}),
      },
      order: { issuedAt: "DESC" },
    });
    if (!invoice) return;
    await mg.update(Invoice, { id: invoice.id }, { dueAt });
  }

  /**
   * Force an invoice to `status`, including issuing a still-DRAFT invoice
   * (stamping `issuedAt`) — unlike the other transitions here, this is a
   * blunt admin/workflow override, not a settlement. No-op when the invoice
   * is missing or already terminal (paid/cancelled/refunded/expired).
   */
  async updateStatus(
    invoiceId: string,
    status: Freight.InvoiceStatus,
    manager?: EntityManager,
  ): Promise<void> {
    // M27: this is the blunt "issue a draft" override — it stamps `issuedAt` but
    // does NOT touch paidAmount/balanceAmount. Its only legitimate use is the
    // Draft → Pending/Issued issue transition. It must NEVER mark an invoice
    // Paid/Refunded/Cancelled/Expired (or PartiallyPaid/Overdue): those carry
    // balance implications and must go through the dedicated settlement methods
    // (recordPayment / markInvoiceAsRefunded / cancelInvoice / expirePayable).
    if (
      status !== Freight.InvoiceStatus.Pending &&
      status !== Freight.InvoiceStatus.Issued
    ) {
      throw new BadRequestException(
        `updateStatus only issues an invoice (→ PENDING/ISSUED); use the dedicated settlement methods to set ${status}.`,
      );
    }
    const mg = manager ?? this.dataSource.manager;
    const invoice = await mg.findOne(Invoice, {
      where: {
        id: invoiceId,
        status: In([Freight.InvoiceStatus.Draft, ...OPEN_STATUSES]),
      },
    });
    if (!invoice) return;
    await mg.update(
      Invoice,
      { id: invoice.id },
      { status, issuedAt: invoice.issuedAt ?? new Date() },
    );
  }

  // ── Payment initiation & settlement (the gateway boundary) ───────────────────

  /**
   * Charge an invoice through the payment gateway. Billing is the single place
   * that turns "what is owed" (the invoice) into a payment intent — the domain
   * never talks to the payment service directly. Resolves the invoice by ID,
   * opens an intent for `invoice.balanceAmount` (so partial payments are honored),
   * records the intent id on the invoice (the settlement correlation key), and
   * returns the client action.
   *
   * When the provider settles synchronously, the invoice is settled inline here —
   * after the intent id is stored — so the `payment.succeeded` correlation can
   * never fire before the link exists. Throws when the invoice is not found or
   * not in an open/payable status.
   */
  async payInvoice(
    invoiceId: string,
    opts: {
      method?: string;
      platform?: "web" | "mobile";
      payerAccount?: string;
      returnUrl?: string;
      failureUrl?: string;
    } = {},
  ): Promise<InitiateResponseDto> {
    const invoice = await this.dataSource.getRepository(Invoice).findOne({
      where: { id: invoiceId, status: In(OPEN_STATUSES) },
    });
    if (!invoice) {
      throw new NotFoundException(
        `Invoice ${invoiceId} not found or not in a payable status`,
      );
    }

    const amountDue = Number(invoice.balanceAmount ?? invoice.totalAmount);
    if (!(amountDue > 0)) {
      throw new BadRequestException("Invoice has no outstanding balance.");
    }

    const result = await this.payment.initiate({
      referenceId: invoice.sourceId,
      source: invoice.source,
      // Freight payments settle under the generic SHIPMENT reference — how the
      // payment service attributes them to the freight API. The payment ↔ invoice
      // link is the intent id (`paymentId`); per-source post-payment reactions live
      // in the domain via `${source}.invoice.paid`. Neither billing nor the payment
      // service branches on a domain-specific reference type.
      referenceType: PaymentReferenceType.SHIPMENT,
      orderRef: invoice.invoiceNumber.replace("-", "_"),
      amountMinor: Math.round(Number(invoice.balanceAmount)),
      currency: invoice.currency,
      reason: `Payment for invoice ${invoice.invoiceNumber}`,
      method: opts.method ?? "TELEBIRR",
      platform: opts.platform,
      payerAccount: opts.payerAccount,
      returnUrl: opts.returnUrl,
      failureUrl: opts.failureUrl,
    });
    //
    // Link the intent to the invoice BEFORE any settlement can correlate against it.
    await this.dataSource
      .getRepository(Invoice)
      .update({ id: invoice.id }, { paymentId: result.intentId });

    // // DEMO: manually fire the gateway `payment.succeeded` callback here, without
    // // waiting for real gateway settlement. Runs AFTER the paymentId link above so
    // // `handlePaymentEvent → settleByPaymentId` can correlate the invoice. TODO:
    // // remove — real settlement flips this via the `${source}.invoice.paid` handler.
    // if (!result.immediateSuccess) {
    //   await this.payment.handlePaymentEvent({
    //     eventType: "payment.succeeded",
    //     eventId: `demo-${result.intentId}`,
    //     referenceId: invoice.sourceId,
    //     intentId: result.intentId,
    //     providerTxnId: result.providerTxnId,
    //     paidAt: (result.paidAt ?? new Date()).toISOString(),
    //   });
    // }

    if (result.immediateSuccess) {
      await this.settleByPaymentId(
        result.intentId,
        result.providerTxnId,
        result.paidAt,
      );
    }

    return result.response;
  }

  /**
   * Settle the open invoice linked to a gateway intent id, if any. Called by the
   * payment service when an intent succeeds: finds the invoice linked by
   * `paymentId`, marks it paid, and emits `${source}.invoice.paid` for the domain
   * to advance on. Idempotent — no-op when no open invoice is linked (already
   * settled, or settled inline by {@link payInvoice}).
   */
  async settleByPaymentId(
    paymentId: string,
    providerTxnId?: string,
    paidAt?: Date,
  ): Promise<Invoice | null> {
    const invoice = await this.dataSource.getRepository(Invoice).findOne({
      where: { paymentId, status: In(OPEN_STATUSES) },
      order: { issuedAt: "DESC" },
    });
    if (!invoice) return null;

    return this.markInvoiceAsPaid(invoice.id, paymentId, undefined, {
      providerTxnId,
      paidAt,
    });
  }
}