edr-platform/apps/edr-freight-api/src/modules/reports/revenue-classification.ts

import { ObjectLiteral, SelectQueryBuilder } from 'typeorm';

import { InvoiceLine } from '../billing/entities/invoice-line.entity';
import { Invoice } from '../billing/entities/invoice.entity';
import { Booking } from '../bookings/entities/booking.entity';
import { Company } from '../companies/entities/company.entity';
import { ShippingLineCompany } from '../shipping-lines/entities/shipping-line-company.entity';
import { CargoType } from '../rule-engine/entities/cargo-type.entity';
import { Yard } from '../rule-engine/entities/yard.entity';
import { applyBookingRefDirectionScope } from '../user-trade-access/trade-scope.util';
import { ReportContext, ReportFilterDef, ReportFilterOption } from './report.types';

/**
 * The shared vocabulary and SQL behind every revenue report.
 *
 * The fact table is `invoice_lines`, not `bookings`: `charge_type` is the only
 * column in the system that separates customs, first/last mile, demurrage,
 * storage and incidental revenue from base freight. A booking total is one
 * lump sum and cannot answer the revenue-classification requirement.
 *
 * Every consumer builds its FROM through {@link revenueLedgerQb}, so the table
 * aliases below (`il i b ct oy dy co slc`) are a fixed contract and the SQL
 * fragments here can reference them directly.
 */

// ---------------------------------------------------------------------------
// Revenue categories
// ---------------------------------------------------------------------------

export const REVENUE_CATEGORIES: ReportFilterOption[] = [
  {
    value: 'CONTAINER_IMPORT_MULTIMODAL',
    label: 'Full Container Import — Multimodal',
  },
  {
    value: 'CONTAINER_IMPORT_UNIMODAL',
    label: 'Full Container Import — Unimodal',
  },
  { value: 'CONTAINER_EXPORT', label: 'Full Container Export' },
  { value: 'EMPTY_CONTAINER_REEXPORT', label: 'Empty Container Re-export' },
  { value: 'FERTILIZER', label: 'Fertilizer Transportation' },
  { value: 'BREAK_BULK', label: 'Break Bulk (Steel, Machineries)' },
  { value: 'RORO', label: 'RoRo Transportation' },
  { value: 'OTHER_IMPORT_BULK', label: 'Other Import Bulk Cargo' },
  { value: 'OTHER_EXPORT_CARGO', label: 'Other Export Cargo' },
  { value: 'DOMESTIC', label: 'Domestic Cargo Transportation' },
  { value: 'INCIDENTAL', label: 'Incidental Charges' },
  { value: 'FIRST_LAST_MILE', label: 'First & Last Mile' },
  { value: 'CUSTOMS_CLEARANCE', label: 'Customs Clearance' },
  { value: 'UNCLASSIFIED', label: 'Unclassified' },
];

/**
 * `charge_type` is an unconstrained varchar written by eight different code
 * paths, so the same concept arrives under several spellings — three for
 * demurrage, four for first/last mile. Every set below absorbs all of them.
 */
export const MILE_CHARGES = ['FIRST_MILE', 'LAST_MILE', 'DELIVERY', 'LAST_MILE_ADVANCE'];

export const INCIDENTAL_CHARGES = [
  'FUEL_SURCHARGE',
  'LASHING',
  'OVERWEIGHT_PER_TON',
  'HAZARD_SURCHARGE',
  'REEFER_SURCHARGE',
  'PIL_EXTRA_FEE',
  'CANCELLATION_FEE',
  'ADJUSTMENT',
  'RATE_ADJUSTMENT',
  'DEMURRAGE',
  'CONTAINER_DEMURRAGE',
  'BULK_DEMURRAGE',
  'TRUCK_DETENTION',
  'STORAGE_FEE',
  'HANDLING_FEE',
  'DOUBLE_HANDLING',
  'SHIPPING_LINE_SERVICE',
];

export const DOMESTIC_CHARGES = ['INTERCITY_BULK', 'INTERCITY_CONTAINER'];

export const CONTAINER_FREIGHT_CHARGES = [
  'CONTAINER_IMPORT',
  'CONTAINER_EXPORT',
  'CONTAINER_20FT',
  'CONTAINER_40FT',
];

export const BULK_FREIGHT_CHARGES = ['BULK_IMPORT', 'BULK_EXPORT', 'FREIGHT'];

/** Cargo codes the business bills as break bulk, wherever the cargo tree puts them. */
export const BREAK_BULK_CODES = ['STEEL_BILLET', 'STEEL', 'MACHINERY', 'PIPES', 'TIMBER'];

export const RORO_CODES = ['TRUCK', 'AUTOMOBILE', 'CARS', 'RORO'];

export const FERTILIZER_CODES = ['FERTILIZER'];

/**
 * Multimodal means EDR carried the sea leg as well as the rail leg. Nothing in
 * the schema says so directly; a named sea carrier on the booking is the
 * agreed proxy. One constant, deliberately — flip it here if the business
 * defines multimodality differently.
 */
const MULTIMODAL_PREDICATE = 'b.shipping_line_id IS NOT NULL';

const list = (values: string[]): string => values.map((v) => `'${v}'`).join(', ');

/**
 * Assigns each invoice line exactly one revenue category. First match wins.
 *
 * Charge-derived rules run BEFORE cargo-derived ones on purpose: a customs or
 * demurrage line billed on a container-import booking is customs/incidental
 * revenue, not container-import revenue. Reversing the order would fold every
 * ancillary charge back into the freight categories.
 *
 * Nothing falls through silently — an unmatched line lands in UNCLASSIFIED and
 * every report surfaces that total as a KPI, because an audit report must
 * never quietly drop money.
 */
export const REVENUE_CATEGORY_EXPR = `CASE
  WHEN il.charge_type LIKE 'CUSTOMS_CLEARANCE%' THEN 'CUSTOMS_CLEARANCE'
  WHEN il.charge_type IN (${list(MILE_CHARGES)}) THEN 'FIRST_LAST_MILE'
  WHEN il.charge_type LIKE 'RETURN_SURCHARGE%'
    OR il.charge_type = 'CONTAINER_WITH_RETURN' THEN 'EMPTY_CONTAINER_REEXPORT'
  WHEN il.charge_type IN (${list(INCIDENTAL_CHARGES)}) THEN 'INCIDENTAL'
  WHEN il.charge_type IN (${list(DOMESTIC_CHARGES)})
    OR (oy.country IS NOT NULL AND oy.country = dy.country) THEN 'DOMESTIC'
  WHEN il.charge_type IN (${list(CONTAINER_FREIGHT_CHARGES)})
    OR b.freight_type = 'CONTAINER' THEN
    CASE
      WHEN b.trade_direction = 'EXPORT' THEN 'CONTAINER_EXPORT'
      WHEN ${MULTIMODAL_PREDICATE} THEN 'CONTAINER_IMPORT_MULTIMODAL'
      ELSE 'CONTAINER_IMPORT_UNIMODAL'
    END
  WHEN ct.code IN (${list(FERTILIZER_CODES)}) THEN 'FERTILIZER'
  WHEN ct.code IN (${list(BREAK_BULK_CODES)}) THEN 'BREAK_BULK'
  WHEN ct.code IN (${list(RORO_CODES)}) THEN 'RORO'
  WHEN b.trade_direction = 'EXPORT' THEN 'OTHER_EXPORT_CARGO'
  WHEN b.trade_direction = 'IMPORT' THEN 'OTHER_IMPORT_BULK'
  ELSE 'UNCLASSIFIED'
END`;

const labelCase = (expr: string, options: ReportFilterOption[]): string =>
  `CASE ${expr}\n  ${options
    .map((o) => `WHEN '${o.value}' THEN '${o.label.replace(/'/g, "''")}'`)
    .join('\n  ')}\nEND`;

/**
 * The same labelling applied to a key that is already a column — for reports
 * that classify in a subquery and label in the wrapper.
 */
export const CATEGORY_LABEL_OF = (keyExpr: string): string =>
  labelCase(keyExpr, REVENUE_CATEGORIES);

/** The category as a business label rather than its key, for display columns. */
export const CATEGORY_LABEL_EXPR = CATEGORY_LABEL_OF(REVENUE_CATEGORY_EXPR);

/**
 * Period-over-period change, as a percentage.
 *
 * The denominator is `ABS(prior)`, not `prior`. A category can post negative
 * revenue in a period — a credit note or rate adjustment outweighing its
 * charges — and dividing by a negative prior flips the sign, reporting a
 * recovery as a decline. Taking the magnitude keeps the sign of the change
 * itself.
 */
export const growthPctExpr = (revenue: string, prior: string): string =>
  `ROUND(100 * (${revenue} - ${prior}) / NULLIF(ABS(${prior}), 0), 1)::float8`;

// ---------------------------------------------------------------------------
// Payment classification
// ---------------------------------------------------------------------------

export const PAYMENT_CLASSES: ReportFilterOption[] = [
  { value: 'RAIL_TRANSPORT', label: 'Rail transport' },
  { value: 'CUSTOMS_CLEARANCE', label: 'Custom clearance' },
  { value: 'FIRST_LAST_MILE', label: 'First and last mile' },
  { value: 'OVERWEIGHT', label: 'Overweight' },
  { value: 'CANCELLATION', label: 'Cancellation' },
  { value: 'DEMURRAGE', label: 'Demurrage' },
  { value: 'STORAGE', label: 'Storage' },
  { value: 'LOADING_UNLOADING', label: 'Loading and unloading' },
  { value: 'ADDITIONAL', label: 'Additional payment' },
];

const RAIL_CHARGES = [...CONTAINER_FREIGHT_CHARGES, ...BULK_FREIGHT_CHARGES, ...DOMESTIC_CHARGES];

const DEMURRAGE_CHARGES = ['DEMURRAGE', 'CONTAINER_DEMURRAGE', 'BULK_DEMURRAGE', 'TRUCK_DETENTION'];

/**
 * Charges that legitimately belong in the spec's "additional payment" bucket.
 *
 * Listed explicitly rather than left to the ELSE arm: ELSE also catches charge
 * types nobody has mapped yet, and those two cases must not be
 * indistinguishable. Naming these is what lets the spec fail when a genuinely
 * new charge type appears.
 */
export const ADDITIONAL_CHARGES = [
  'FUEL_SURCHARGE',
  'HAZARD_SURCHARGE',
  'REEFER_SURCHARGE',
  'PIL_EXTRA_FEE',
  'RETURN_SURCHARGE',
  'RETURN_SURCHARGE_20FT',
  'RETURN_SURCHARGE_40FT',
  'CONTAINER_WITH_RETURN',
  'ADJUSTMENT',
  'RATE_ADJUSTMENT',
  'SHIPPING_LINE_SERVICE',
];

/**
 * The nine buckets the revenue spec asks payments to be classified into.
 *
 * Caveat worth repeating wherever this is shown: there is no dedicated
 * loading/unloading charge type in the system. HANDLING_FEE, DOUBLE_HANDLING
 * and LASHING are the nearest equivalent, so that bucket is an approximation,
 * not an exact match.
 */
export const PAYMENT_CLASS_EXPR = `CASE
  WHEN il.charge_type LIKE 'CUSTOMS_CLEARANCE%' THEN 'CUSTOMS_CLEARANCE'
  WHEN il.charge_type IN (${list(MILE_CHARGES)}) THEN 'FIRST_LAST_MILE'
  WHEN il.charge_type IN (${list(RAIL_CHARGES)}) THEN 'RAIL_TRANSPORT'
  WHEN il.charge_type = 'OVERWEIGHT_PER_TON' THEN 'OVERWEIGHT'
  WHEN il.charge_type = 'CANCELLATION_FEE' THEN 'CANCELLATION'
  WHEN il.charge_type IN (${list(DEMURRAGE_CHARGES)}) THEN 'DEMURRAGE'
  WHEN il.charge_type = 'STORAGE_FEE' THEN 'STORAGE'
  WHEN il.charge_type IN ('HANDLING_FEE', 'DOUBLE_HANDLING', 'LASHING')
    THEN 'LOADING_UNLOADING'
  WHEN il.charge_type IN (${list(ADDITIONAL_CHARGES)}) THEN 'ADDITIONAL'
  ELSE 'ADDITIONAL'
END`;

// ---------------------------------------------------------------------------
// Period granularity
// ---------------------------------------------------------------------------

/**
 * A granularity, as SQL builders rather than fragments to interpolate.
 *
 * Five of the eight are plain `date_trunc` units. The other three — half-year,
 * nine-month, ninety-day — have no `date_trunc` equivalent in Postgres, so they
 * are offset arithmetic from the start of the calendar year. Builders let both
 * kinds live behind one interface.
 */
interface PeriodUnit {
  label: string;
  /** Interval one whole block wide. Only exact for the six regular units. */
  step: string;
  /** Timestamp expression → the start of the block that timestamp falls in. */
  truncOn: (dateExpr: string) => string;
  /** Block-start expression → its display label. */
  labelOn: (truncExpr: string) => string;
  /**
   * Block-start expression → the start of the NEXT block. Not always
   * `+ step`: a ragged unit's final block of the year is shorter than its own
   * step, so stepping past it overshoots into the wrong block.
   */
  nextStartOn: (truncExpr: string) => string;
}

const regular = (trunc: string, fmt: string, label: string, step: string): PeriodUnit => ({
  label,
  step,
  truncOn: (dateExpr) => `date_trunc('${trunc}', ${dateExpr})`,
  labelOn: (truncExpr) => `to_char(${truncExpr}, '${fmt}')`,
  nextStartOn: (truncExpr) => `(${truncExpr} + INTERVAL '${step}')`,
});

/**
 * Blocks of `months` months counted from January, so they reset every calendar
 * year. Six divides twelve and nine does not: a nine-month year is Jan–Sep plus
 * a short Oct–Dec. That ragged tail is inherent to the unit — the alternative
 * is blocks that drift out of the calendar, which is not what "calendar
 * anchored" means.
 */
const monthBlocks = (months: number, marker: string, label: string): PeriodUnit => ({
  label,
  step: `${months} months`,
  truncOn: (dateExpr) =>
    `(date_trunc('year', ${dateExpr})` +
    ` + (((EXTRACT(MONTH FROM ${dateExpr})::int - 1) / ${months}) * INTERVAL '${months} months'))`,
  labelOn: (truncExpr) =>
    `(to_char(${truncExpr}, 'YYYY') || '-${marker}' ||` +
    ` ((EXTRACT(MONTH FROM ${truncExpr})::int - 1) / ${months} + 1)::text)`,
  nextStartOn: (truncExpr) =>
    `LEAST(${truncExpr} + INTERVAL '${months} months',` +
    ` date_trunc('year', ${truncExpr}) + INTERVAL '1 year')`,
});

/**
 * Frozen whitelist. The runner coerces a `select` filter to a trimmed string or
 * null; that string is used only as an object key here, so the user's value
 * never reaches SQL — one of eight compile-time constants does.
 *
 * Every label is zero-padded or single-digit-bounded, so lexicographic order
 * equals chronological order. The growth windows depend on that.
 */
const PERIOD_UNITS: Record<string, PeriodUnit> = {
  day: regular('day', 'YYYY-MM-DD', 'Daily', '1 day'),
  week: regular('week', 'IYYY-"W"IW', 'Weekly', '1 week'),
  month: regular('month', 'YYYY-MM', 'Monthly', '1 month'),
  // `quarter` is a valid date_trunc unit but NOT a valid interval unit —
  // INTERVAL '1 quarter' is a syntax error, so the step is spelled in months.
  quarter: regular('quarter', 'YYYY-"Q"Q', 'Quarterly', '3 months'),
  half_year: monthBlocks(6, 'H', 'Half-yearly'),
  nine_month: monthBlocks(9, 'N', 'Nine-monthly'),
  /**
   * Four 90-day blocks from January 1st: days 1, 91, 181, 271.
   *
   * The block index is capped at 3 on purpose. Uncapped, `(doy - 1) / 90` puts
   * December 27th onwards in a fifth block — a 5-day stub bucket at the end of
   * every year, which is noise rather than a period. Capping instead lets the
   * fourth block absorb the remainder and run 95 or 96 days.
   *
   * The label carries the zero-padded start day-of-year, which keeps it sorting
   * chronologically and — unlike an ordinal — says out loud that the blocks are
   * day-counted rather than month-aligned.
   */
  ninety_day: {
    label: '90-day',
    step: '90 days',
    truncOn: (dateExpr) =>
      `(date_trunc('year', ${dateExpr})` +
      ` + (LEAST((EXTRACT(DOY FROM ${dateExpr})::int - 1) / 90, 3) * INTERVAL '90 days'))`,
    labelOn: (truncExpr) =>
      `(to_char(${truncExpr}, 'YYYY') || '-D' || lpad(EXTRACT(DOY FROM ${truncExpr})::int::text, 3, '0'))`,
    // The fourth block ends with the year, not 90 days after it started.
    nextStartOn: (truncExpr) =>
      `(CASE WHEN EXTRACT(DOY FROM ${truncExpr})::int >= 271` +
      ` THEN date_trunc('year', ${truncExpr}) + INTERVAL '1 year'` +
      ` ELSE ${truncExpr} + INTERVAL '90 days' END)`,
  },
  year: regular('year', 'YYYY', 'Yearly', '1 year'),
};

export const PERIOD_FILTER: ReportFilterDef = {
  key: 'period',
  label: 'Granularity',
  type: 'select',
  options: Object.entries(PERIOD_UNITS).map(([value, u]) => ({
    value,
    label: u.label,
  })),
};

/** The timestamp every revenue report buckets and filters on. */
export const REVENUE_DATE = 'COALESCE(i.issued_at, i.created_at)';

/**
 * The period label expression, as a string.
 *
 * Callers must reuse the returned string VERBATIM in the select, the GROUP BY
 * and any window `ORDER BY`. Two traps make this non-negotiable:
 *
 * 1. A window `ORDER BY date_trunc(...)` when the group key is `to_char(date_trunc(...))`
 *    fails with "column i.issued_at must appear in the GROUP BY clause".
 * 2. Ordinal shorthand — `OVER (PARTITION BY 2 ORDER BY 1)` — is NOT a
 *    positional reference inside a window clause. Postgres reads the integers
 *    as constants, so it partitions by a constant and applies no ordering. It
 *    type-checks, it EXPLAINs clean, and it returns plausible garbage.
 */
export function periodExpr(params: Record<string, unknown>): string {
  return periodExprOn(REVENUE_DATE, params);
}

export function resolvePeriod(params: Record<string, unknown>): PeriodUnit {
  const key = String(params.period ?? '');
  return PERIOD_UNITS[key] ?? PERIOD_UNITS.month;
}

/**
 * The same bucketing over any timestamp column. Revenue buckets on the invoice
 * date; the operations reports bucket on a train's actual departure, and share
 * these units so a month means the same thing on both sides of the product.
 */
export const periodExprOn = (dateExpr: string, params: Record<string, unknown>): string =>
  resolvePeriod(params).labelOn(periodTruncExprOn(dateExpr, params));

export const periodTruncExprOn = (dateExpr: string, params: Record<string, unknown>): string =>
  resolvePeriod(params).truncOn(dateExpr);

/** The period's start timestamp — what to GROUP BY when a report needs it numerically. */
export const periodTruncExpr = (params: Record<string, unknown>): string =>
  periodTruncExprOn(REVENUE_DATE, params);

/**
 * The period as a number, for regression: seconds since epoch at the period's
 * start. Using the timestamp itself rather than `row_number()` keeps a trend
 * calculation to a single window level — Postgres rejects a window function
 * nested inside another window function's arguments.
 */
export const periodOrdinalExpr = (params: Record<string, unknown>): string =>
  `EXTRACT(EPOCH FROM ${periodTruncExpr(params)})`;

/**
 * Same scale, one period later — where a one-step-ahead projection lands.
 *
 * Asks the unit rather than adding its step, because the two differ for the
 * ragged units: a nine-month year's second block is three months long, and a
 * 90-day year's fourth is 95, so `+ step` would land past the next block start
 * and evaluate the regression at the wrong x.
 */
export const nextPeriodOrdinalExpr = (params: Record<string, unknown>): string =>
  `EXTRACT(EPOCH FROM ${resolvePeriod(params).nextStartOn(periodTruncExpr(params))})`;

// ---------------------------------------------------------------------------
// Volume — measured at line grain, never joined from the booking
// ---------------------------------------------------------------------------

/**
 * `invoice_lines.quantity` already carries the billed quantity per line, and
 * `metadata->>'unit'` says what it counts (PER_TON / PER_CONTAINER / PER_WAGON).
 * Joining booking-level tonnage instead would multiply it by the number of
 * lines on the booking.
 */
export const TONS_EXPR = `SUM(il.quantity) FILTER (WHERE il.metadata->>'unit' = 'PER_TON')`;

export const CONTAINERS_EXPR = `SUM(il.quantity) FILTER (WHERE il.metadata->>'unit' = 'PER_CONTAINER')`;

/**
 * TEU is never stored. It is derived from the charge code's size suffix; lines
 * whose code carries no size (CONTAINER_IMPORT / CONTAINER_EXPORT) count as one
 * TEU each, which under-counts any 40ft box billed under an unsized code.
 */
export const TEU_EXPR = `SUM(il.quantity * CASE WHEN il.charge_type LIKE '%40FT%' THEN 2 ELSE 1 END)
  FILTER (WHERE il.metadata->>'unit' = 'PER_CONTAINER')`;

/**
 * Revenue per unit, against whichever unit the category is actually billed in.
 * Exactly one of tons/TEU is non-null per category, so this is per-ton for bulk
 * and per-TEU for containers; the `unit` column says which.
 */
export const AVG_PER_UNIT_EXPR = `ROUND(
  SUM(il.amount) / NULLIF(COALESCE(${TONS_EXPR}, 0) + COALESCE(${TEU_EXPR}, 0), 0), 2
)::float8`;

export const UNIT_LABEL_EXPR = `CASE
  WHEN COALESCE(${TONS_EXPR}, 0) > 0 THEN 'per ton'
  WHEN COALESCE(${TEU_EXPR}, 0) > 0 THEN 'per TEU'
  ELSE ''
END`;

// ---------------------------------------------------------------------------
// The shared ledger query
// ---------------------------------------------------------------------------

/** Invoice states that never represent recognised revenue. */
const DEAD_INVOICE_STATUSES = ['DRAFT', 'CANCELLED'];

export const PAYMENT_METHOD_OPTIONS: ReportFilterOption[] = [
  'telebirr',
  'cbe-birr',
  'cbe-bill',
  'ebirr',
  'waafi',
  'dmoney',
  'cac-bank',
  'card',
].map((v) => ({ value: v, label: v }));

export const CURRENCY_FILTER: ReportFilterDef = {
  key: 'currency',
  label: 'Currency',
  type: 'select',
  options: [
    { value: 'ETB', label: 'ETB' },
    { value: 'USD', label: 'USD' },
    { value: 'DJF', label: 'DJF' },
  ],
};

/**
 * The filter set shared by every revenue report, so they drill into each other
 * without losing context.
 *
 * `currency` is not optional decoration: the ledger holds both ETB and USD
 * lines, and summing across them produces a number that means nothing. It
 * defaults to ETB in {@link revenueLedgerQb} rather than being left blank.
 */
export const REVENUE_FILTERS: ReportFilterDef[] = [
  { key: 'date', label: 'Issued', type: 'daterange' },
  CURRENCY_FILTER,
  {
    key: 'categories',
    label: 'Revenue category',
    type: 'multiselect',
    options: REVENUE_CATEGORIES,
  },
  { key: 'origin', label: 'Origin', type: 'select', optionsQuery: yardOptions },
  {
    key: 'destination',
    label: 'Destination',
    type: 'select',
    optionsQuery: yardOptions,
  },
  { key: 'customer', label: 'Customer / booking ref', type: 'text' },
  {
    key: 'methods',
    label: 'Payment method',
    type: 'multiselect',
    options: PAYMENT_METHOD_OPTIONS,
  },
];

/** Stations are reference data — 23 rows that change about yearly. */
export async function yardOptions(ds: ReportContext['ds']): Promise<ReportFilterOption[]> {
  return ds
    .createQueryBuilder()
    .from(Yard, 'y')
    .select('y.code', 'value')
    .addSelect('y.label', 'label')
    .where('y.deleted_at IS NULL AND y.is_active')
    .orderBy('y.display_order', 'ASC')
    .getRawMany<ReportFilterOption>();
}

/** Currency the ledger reports in when the caller does not choose one. */
export const DEFAULT_CURRENCY = 'ETB';

export const currencyOf = (params: Record<string, unknown>): string =>
  (params.currency as string) || DEFAULT_CURRENCY;

/**
 * Every revenue report starts here: one invoice line joined out to the booking
 * that explains it. Bookings are LEFT joined on purpose — warehouse, demurrage
 * and shipping-line-credit invoices carry no booking and must still be counted.
 *
 * The booking join is `i.source_id = b.id::text`, never `i.source_id::uuid`:
 * `source_id` is a varchar with no FK that holds non-UUID values for other
 * sources (`eims-self-test-…`), so casting it would throw at runtime.
 */
export function revenueLedgerQb(ctx: ReportContext): SelectQueryBuilder<ObjectLiteral> {
  const { params, directions } = ctx;

  const qb = ctx.ds
    .createQueryBuilder()
    .from(InvoiceLine, 'il')
    .innerJoin(Invoice, 'i', 'i.id = il.invoice_id AND i.deleted_at IS NULL')
    .leftJoin(
      Booking,
      'b',
      "i.source = 'booking' AND i.source_id = b.id::text AND b.deleted_at IS NULL",
    )
    .leftJoin(CargoType, 'ct', 'ct.id = b.cargo_type_id')
    .leftJoin(Yard, 'oy', 'oy.id = b.origin_yard_id')
    .leftJoin(Yard, 'dy', 'dy.id = b.destination_yard_id')
    .leftJoin(Company, 'co', 'co.id = i.company_id')
    .leftJoin(ShippingLineCompany, 'slc', 'slc.id = i.shipping_line_company_id')
    .where('il.deleted_at IS NULL')
    .andWhere('i.status NOT IN (:...deadInvoiceStatuses)', {
      deadInvoiceStatuses: DEAD_INVOICE_STATUSES,
    })
    .andWhere("i.source <> 'eims_self_test'")
    // Mixing ETB and USD into one SUM produces a meaningless number.
    .andWhere('il.currency = :currency', { currency: currencyOf(params) });

  if (params.dateFrom) {
    qb.andWhere(`${REVENUE_DATE} >= :dateFrom`, { dateFrom: params.dateFrom });
  }
  if (params.dateTo) {
    qb.andWhere(`${REVENUE_DATE} < :dateTo`, { dateTo: params.dateTo });
  }

  const categories = params.categories as string[] | null;
  if (categories?.length) {
    qb.andWhere(`${REVENUE_CATEGORY_EXPR} IN (:...categories)`, { categories });
  }

  if (params.origin) qb.andWhere('oy.code = :origin', { origin: params.origin });
  if (params.destination) {
    qb.andWhere('dy.code = :destination', { destination: params.destination });
  }

  if (params.customer) {
    qb.andWhere(
      '(co.name ILIKE :customer OR slc.name ILIKE :customer OR b.reference ILIKE :customer)',
      { customer: `%${params.customer as string}%` },
    );
  }

  const methods = params.methods as string[] | null;
  if (methods?.length) {
    qb.andWhere(
      `EXISTS (SELECT 1 FROM freight.payments p
               WHERE p.ref_id = i.source_id AND p.status = 'success'
                 AND p.method IN (:...methods))`,
      { methods },
    );
  }

  // Hides lines whose booking sits outside the caller's trade scope. Lines with
  // no booking carry no direction and stay visible.
  applyBookingRefDirectionScope(qb, 'i.source_id', directions);

  return qb;
}

/**
 * Invoice-grain sibling of {@link revenueLedgerQb}, for the reports that must
 * not multiply an invoice by its line count — outstanding balance,
 * reconciliation, receivable/payable. Same joins, same filters, minus the
 * line-only ones (charge category, currency lives on the invoice here).
 */
export function invoiceLedgerQb(ctx: ReportContext): SelectQueryBuilder<ObjectLiteral> {
  const { params, directions } = ctx;

  const qb = ctx.ds
    .createQueryBuilder()
    .from(Invoice, 'i')
    .leftJoin(
      Booking,
      'b',
      "i.source = 'booking' AND i.source_id = b.id::text AND b.deleted_at IS NULL",
    )
    .leftJoin(CargoType, 'ct', 'ct.id = b.cargo_type_id')
    .leftJoin(Yard, 'oy', 'oy.id = b.origin_yard_id')
    .leftJoin(Yard, 'dy', 'dy.id = b.destination_yard_id')
    .leftJoin(Company, 'co', 'co.id = i.company_id')
    .leftJoin(ShippingLineCompany, 'slc', 'slc.id = i.shipping_line_company_id')
    .where('i.deleted_at IS NULL')
    .andWhere('i.status NOT IN (:...deadInvoiceStatuses)', {
      deadInvoiceStatuses: DEAD_INVOICE_STATUSES,
    })
    .andWhere("i.source <> 'eims_self_test'")
    .andWhere('i.currency = :currency', { currency: currencyOf(params) });

  if (params.dateFrom) qb.andWhere(`${REVENUE_DATE} >= :dateFrom`, { dateFrom: params.dateFrom });
  if (params.dateTo) qb.andWhere(`${REVENUE_DATE} < :dateTo`, { dateTo: params.dateTo });
  if (params.origin) qb.andWhere('oy.code = :origin', { origin: params.origin });
  if (params.destination)
    qb.andWhere('dy.code = :destination', { destination: params.destination });
  if (params.customer) {
    qb.andWhere(
      '(co.name ILIKE :customer OR slc.name ILIKE :customer OR b.reference ILIKE :customer)',
      { customer: `%${params.customer as string}%` },
    );
  }

  applyBookingRefDirectionScope(qb, 'i.source_id', directions);
  return qb;
}

/**
 * What the payment gateway actually recorded against this invoice.
 *
 * `freight.payments` is keyed by the booking, not the invoice — `ref_id` holds
 * the booking id and there is no invoice column — while one booking routinely
 * carries several invoices (25 booking ids here back 58 of them). Reading the
 * booking's gateway total straight off each invoice therefore hands the same
 * money to every sibling: 113M of gateway receipts claimed against 44M of
 * recorded settlement, which surfaced as ~89M of variance that does not exist.
 *
 * So the booking's receipts are apportioned across its invoices by their share
 * of what was recorded as settled — the same device as {@link PAID_SHARE}, and
 * the only split that makes the report's gateway column sum to the payments
 * table. A booking whose invoices record no settlement at all cannot be split
 * that way; it falls back to the billed share, so gateway money nobody booked
 * still shows up as variance instead of vanishing.
 *
 * `invoices.payment_id` does resolve to a `freight.payments` row, but only 72
 * of 85 successful payments are pointed at by one, so keying on it drops real
 * receipts.
 */
export const GATEWAY_PAID = `(
  (SELECT COALESCE(SUM(p.amount), 0) FROM freight.payments p
   WHERE p.ref_id = i.source_id AND p.status = 'success')
  * COALESCE(
      i.paid_amount / NULLIF((SELECT SUM(i2.paid_amount) FROM freight.invoices i2
                              WHERE i2.source_id = i.source_id AND i2.deleted_at IS NULL
                                AND i2.status NOT IN ('DRAFT', 'CANCELLED')), 0),
      i.total_amount / NULLIF((SELECT SUM(i2.total_amount) FROM freight.invoices i2
                               WHERE i2.source_id = i.source_id AND i2.deleted_at IS NULL
                                 AND i2.status NOT IN ('DRAFT', 'CANCELLED')), 0),
      0)
)`;

/** The payer, whichever of the two mutually exclusive payer columns is set. */
export const PAYER_EXPR = "COALESCE(co.name, slc.name, 'Unknown')";

/** `SUM(amount)`, rounded to whole currency and typed as a JS number. */
export const REVENUE_SUM = 'ROUND(COALESCE(SUM(il.amount), 0))::float8';

/**
 * Settled share of a line, apportioned by how much of its invoice was paid.
 * Invoice-level `paid_amount` cannot be attributed to a single line any other
 * way.
 */
export const PAID_SHARE =
  'il.amount * CASE WHEN i.total_amount > 0 THEN i.paid_amount / i.total_amount ELSE 0 END';

/** The payment class as a business label rather than its key. */
export const PAYMENT_CLASS_LABEL_EXPR = labelCase(PAYMENT_CLASS_EXPR, PAYMENT_CLASSES);