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

import { ObjectLiteral, SelectQueryBuilder } from 'typeorm';

import { Booking } from '../bookings/entities/booking.entity';
import { OperationsStandard } from '../operations-reporting/entities/operations-standard.entity';
import { CargoType } from '../rule-engine/entities/cargo-type.entity';
import { TrainCheckpointEvent } from '../train-scheduling/entities/train-checkpoint-event.entity';
import { TrainSchedule } from '../train-schedules/entities/train-schedule.entity';
import { TrainSet } from '../train-sets/entities/train-set.entity';
import { TrainSetWagon } from '../train-sets/entities/train-set-wagon.entity';
import { WagonBookingAllocation } from '../train-schedules/entities/wagon-booking-allocation.entity';
import { Yard } from '../rule-engine/entities/yard.entity';
import { applyDirectionScope } from '../user-trade-access/trade-scope.util';
import { ReportContext, ReportFilterDef, ReportFilterOption } from './report.types';
import { REVENUE_CATEGORIES, resolvePeriod, yardOptions } from './revenue-classification';

/**
 * The shared vocabulary and SQL behind every operations report — turnaround,
 * delay, trainset, TEU and cargo volume.
 *
 * The fact table is `wagon_booking_allocations`: one row is one booking's cargo
 * on one wagon of one departure. That is the marshalling record — what was
 * actually put on the train — and it is the only grain that can answer both
 * "how many TEU moved" and "how many wagons did it take", which the volume and
 * trainset reports need together.
 *
 * Every consumer builds its FROM through {@link allocationLedgerQb}, so the
 * table aliases below (`wba tsw ts b ct oy dy std`) are a fixed contract and
 * the fragments here reference them directly.
 *
 * This is deliberately a SECOND classification module rather than an extension
 * of `revenue-classification.ts`. That one classifies invoice lines by charge
 * code; this one classifies physical cargo by booking and cargo type. The two
 * answer different questions and a row that is one revenue category can be a
 * different operational category — an incidental charge on a container booking,
 * for instance, is INCIDENTAL revenue but container tonnage.
 */

// ---------------------------------------------------------------------------
// Cargo categories
// ---------------------------------------------------------------------------

export const CARGO_CATEGORIES: ReportFilterOption[] = [
  { value: 'CONTAINER_IMPORT_MULTIMODAL', label: 'Multimodal container import' },
  { value: 'CONTAINER_IMPORT_UNIMODAL', label: 'Unimodal container import' },
  { value: 'CONTAINER_EXPORT', label: 'Export container' },
  { value: 'EMPTY_CONTAINER', label: 'Empty container' },
  { value: 'FERTILIZER', label: 'Fertilizer' },
  { value: 'RORO', label: 'RoRo' },
  { value: 'BREAK_BULK', label: 'Break bulk' },
  { value: 'SAND', label: 'Sand' },
  { value: 'BULK', label: 'Bulk' },
  { value: 'OTHER_IMPORT', label: 'Other imports' },
  { value: 'OTHER_EXPORT', label: 'Other export cargo' },
  { value: 'UNCLASSIFIED', label: 'Unclassified' },
];

/**
 * Container classes for the TEU report — the four the spec names.
 * `EMPTY_CONTAINER_RETURN` is the empty re-export leg.
 */
export const CONTAINER_CLASSES: ReportFilterOption[] = [
  { value: 'CONTAINER_IMPORT_MULTIMODAL', label: 'Multimodal container import' },
  { value: 'CONTAINER_IMPORT_UNIMODAL', label: 'Unimodal container import' },
  { value: 'CONTAINER_EXPORT', label: 'Full export container' },
  { value: 'EMPTY_CONTAINER_RETURN', label: 'Empty container return' },
];

/** `cargo_types.code` is admin-managed, so each set absorbs every spelling seeded so far. */
export const RORO_CODES = ['TRUCK', 'AUTOMOBILE', 'CARS', 'RORO'];
export const BREAK_BULK_CODES = [
  'BREAK_BULK',
  'STEEL_BILLET',
  'STEEL',
  'MACHINERY',
  'PIPES',
  'TIMBER',
];
export const FERTILIZER_CODES = ['FERTILIZER'];
export const SAND_CODES = ['SAND'];

/** Cargo charged at the lighter per-wagon rate — vegetables, milk, meat, livestock. */
export const PERISHABLE_CODES = ['PERISHABLE', 'LIVESTOCK'];

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

/**
 * A booking whose equipment_return is RETURN is the empty-container movement
 * itself; WITH_RETURN / WITHOUT_RETURN describe a laden booking's obligation.
 * This is the only booking-level marker of an empty box — no table records
 * laden-vs-empty on the container row.
 */
const IS_EMPTY_CONTAINER = "b.equipment_return = 'RETURN'";

/**
 * Multimodal means a named sea carrier is on the booking — the same proxy the
 * revenue reports use. There is no explicit multimodal flag; confirm with the
 * business before treating this as definitive.
 */
const IS_MULTIMODAL = 'b.shipping_line_id IS NOT NULL';

const IS_CONTAINER = "COALESCE(b.freight_type, wba.load_type) = 'CONTAINER'";

export const CARGO_CATEGORY_EXPR = `CASE
  WHEN ${IS_EMPTY_CONTAINER} THEN 'EMPTY_CONTAINER'
  WHEN ${IS_CONTAINER} AND b.trade_direction = 'EXPORT' THEN 'CONTAINER_EXPORT'
  WHEN ${IS_CONTAINER} AND ${IS_MULTIMODAL} THEN 'CONTAINER_IMPORT_MULTIMODAL'
  WHEN ${IS_CONTAINER} THEN 'CONTAINER_IMPORT_UNIMODAL'
  WHEN ct.code IN (${quote(FERTILIZER_CODES)}) THEN 'FERTILIZER'
  WHEN ct.code IN (${quote(RORO_CODES)}) THEN 'RORO'
  WHEN ct.code IN (${quote(BREAK_BULK_CODES)}) THEN 'BREAK_BULK'
  WHEN ct.code IN (${quote(SAND_CODES)}) THEN 'SAND'
  WHEN b.trade_direction = 'EXPORT' THEN 'OTHER_EXPORT'
  WHEN b.trade_direction = 'IMPORT' THEN 'OTHER_IMPORT'
  WHEN b.id IS NOT NULL THEN 'BULK'
  ELSE 'UNCLASSIFIED'
END`;

/**
 * The same cargo, classified into the REVENUE vocabulary — the categories
 * `revenue-classification.ts` bills against, minus its charge-only buckets
 * (incidental, first/last mile, customs), which no physical wagon can be.
 *
 * Mirrors the cargo arms of `REVENUE_CATEGORY_EXPR` in that expression's own
 * order, so a ton and the birr charged for it land in the same bucket: empty
 * re-export before domestic, domestic before anything about what is in the box.
 * Reports that must reconcile tonnage against revenue group by this one; the
 * operational vocabulary above keeps sand and bulk apart, which no invoice does.
 */
export const REVENUE_CARGO_CATEGORY_EXPR = `CASE
  WHEN ${IS_EMPTY_CONTAINER} THEN 'EMPTY_CONTAINER_REEXPORT'
  WHEN oy.country IS NOT NULL AND oy.country = dy.country THEN 'DOMESTIC'
  WHEN ${IS_CONTAINER} AND b.trade_direction = 'EXPORT' THEN 'CONTAINER_EXPORT'
  WHEN ${IS_CONTAINER} AND ${IS_MULTIMODAL} THEN 'CONTAINER_IMPORT_MULTIMODAL'
  WHEN ${IS_CONTAINER} THEN 'CONTAINER_IMPORT_UNIMODAL'
  WHEN ct.code IN (${quote(FERTILIZER_CODES)}) THEN 'FERTILIZER'
  WHEN ct.code IN (${quote(BREAK_BULK_CODES)}) THEN 'BREAK_BULK'
  WHEN ct.code IN (${quote(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`;

/** The revenue vocabulary as a filter, plus the wagon that carries no cargo. */
export const REVENUE_CARGO_FILTER: ReportFilterDef = {
  key: 'categories',
  label: 'Cargo type',
  type: 'multiselect',
  options: [...REVENUE_CATEGORIES, { value: 'EMPTY_WAGON', label: 'Empty wagon' }],
};

export const CONTAINER_CLASS_EXPR = `CASE
  WHEN ${IS_EMPTY_CONTAINER} THEN 'EMPTY_CONTAINER_RETURN'
  WHEN b.trade_direction = 'EXPORT' THEN 'CONTAINER_EXPORT'
  WHEN ${IS_MULTIMODAL} THEN 'CONTAINER_IMPORT_MULTIMODAL'
  ELSE 'CONTAINER_IMPORT_UNIMODAL'
END`;

/**
 * Every fixed key a planner may enter on the targets screen — the category and
 * container-class vocabularies. Station targets are keyed on a yard code, which
 * is reference data rather than a fixed list, so they are not enumerated here.
 */
export const TARGET_DIMENSION_KEYS: string[] = [
  ...CARGO_CATEGORIES.map((o) => o.value),
  ...CONTAINER_CLASSES.map((o) => o.value),
];

/** Turns a key-emitting CASE into a label-emitting one, so a report shows business names. */
const labelCase = (keyExpr: string, options: ReportFilterOption[]): string =>
  `CASE ${options
    .map((o) => `WHEN (${keyExpr}) = '${o.value}' THEN '${o.label.replace(/'/g, "''")}'`)
    .join(' ')} ELSE (${keyExpr}) END`;

export const CARGO_CATEGORY_LABEL_EXPR = labelCase(CARGO_CATEGORY_EXPR, CARGO_CATEGORIES);
export const CONTAINER_CLASS_LABEL_EXPR = labelCase(CONTAINER_CLASS_EXPR, CONTAINER_CLASSES);

/**
 * 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, CARGO_CATEGORIES);
export const CONTAINER_CLASS_LABEL_OF = (keyExpr: string): string =>
  labelCase(keyExpr, CONTAINER_CLASSES);

// ---------------------------------------------------------------------------
// Standards
// ---------------------------------------------------------------------------

/**
 * A standard, read off the joined `operations_standards` row.
 *
 * The fallback is not decoration: the row is seeded by migration, but a report
 * must not return zeros — or divide by zero — on an environment where the seed
 * has not run. The fallbacks are the spec's own figures.
 */
const stdRow = (column: string, fallback: number): string =>
  `COALESCE(std.${column}, ${fallback})`;

/**
 * The same value in an aggregate select. `std` is a single joined row, so the
 * column is constant across the group — but Postgres still demands it be
 * grouped or aggregated, and wrapping it in MAX() is cheaper than dragging it
 * through every report's GROUP BY.
 */
const stdAgg = (column: string, fallback: number): string =>
  `MAX(COALESCE(std.${column}, ${fallback}))`;

/** Standard hours a train may stand at a station, by the station's country. */
export const STATION_STANDARD_HOURS_EXPR = `CASE
  WHEN y.country = 'Djibouti' THEN ${stdRow('station_standard_hours_djibouti', 13)}
  ELSE ${stdRow('station_standard_hours_ethiopia', 10)}
END`;

/** Whichever end of the corridor is on the Djibouti side, if either is. */
export const DJIBOUTI_YARD_CODE_EXPR = `CASE
  WHEN oy.country = 'Djibouti' THEN oy.code
  WHEN dy.country = 'Djibouti' THEN dy.code
END`;

/** True when the departure carried any container allocation. */
export const SCHEDULE_IS_CONTAINER = `EXISTS (
  SELECT 1 FROM freight.wagon_booking_allocations a
    JOIN freight.train_set_wagons w ON w.id = a.train_set_wagon_id AND w.deleted_at IS NULL
   WHERE w.train_set_id = ts.train_set_id
     AND a.deleted_at IS NULL AND a.load_type = 'CONTAINER'
)`;

/**
 * Standard turn-around cycle for a departure, in hours. Container trains run
 * the 65-hour cycle; a bulk cycle depends on which Djibouti terminal it works.
 * Schedule grain — it reads `ts`, `oy` and `dy`, not the allocation aliases.
 */
export const CYCLE_STANDARD_HOURS_EXPR = `CASE
  WHEN ${SCHEDULE_IS_CONTAINER} THEN ${stdRow('cycle_standard_hours_container', 65)}
  WHEN ${DJIBOUTI_YARD_CODE_EXPR} = 'DORALEH_MULTIPURPOSE_PORT_DMP' THEN ${stdRow('cycle_standard_hours_bulk_dmp', 88)}
  WHEN ${DJIBOUTI_YARD_CODE_EXPR} = 'BCC' THEN ${stdRow('cycle_standard_hours_bulk_bcc', 96)}
  ELSE ${stdRow('cycle_standard_hours_bulk_nagad', 96)}
END`;

export const DELAY_TOLERANCE_HOURS_EXPR = `(${stdRow('delay_tolerance_minutes', 30)} / 60.0)`;

/**
 * Standard loading-and-unloading time for a stop, by what the train carries.
 *
 * Deliberately NOT wrapped in a fallback like every other standard here: the
 * reporting spec publishes no figure for handling, so there is nothing honest
 * to fall back to. Until a planner enters one in Operating standards this is
 * NULL, and the rate and verdict that read it stay empty rather than judging a
 * train against a number nobody agreed to.
 */
export const HANDLING_STANDARD_HOURS_EXPR = `CASE
  WHEN ${SCHEDULE_IS_CONTAINER} THEN std.handling_standard_hours_container
  ELSE std.handling_standard_hours_bulk
END`;

/**
 * Joins the single standards row. Restricted by id to the earliest live row so
 * a stray second row could never fan a report's result out.
 */
export const STANDARDS_JOIN = `std.id = (
  SELECT s.id FROM freight.operations_standards s
   WHERE s.deleted_at IS NULL ORDER BY s.created_at ASC LIMIT 1
)`;

// ---------------------------------------------------------------------------
// Distance
// ---------------------------------------------------------------------------

/**
 * Configured rail distance for a yard pair, in km. Symmetric: `yard_distances`
 * stores one row per pair and an A→B row governs B→A.
 *
 * Returns NULL when the pair is not configured, and every caller must let that
 * null through rather than coalescing to zero — a missing distance is not a
 * zero distance, and Ton/Km computed from one would understate silently.
 */
export const distanceKmBetween = (fromCol: string, toCol: string): string => `(
  SELECT yd.distance_km FROM freight.yard_distances yd
   WHERE yd.deleted_at IS NULL
     AND ((yd.from_yard_id = ${fromCol} AND yd.to_yard_id = ${toCol})
       OR (yd.from_yard_id = ${toCol} AND yd.to_yard_id = ${fromCol}))
   LIMIT 1
)`;

/** Standard running time for a leg, falling back to the default leg standard. */
export const legStandardHours = (fromCol: string, toCol: string): string => `COALESCE((
  SELECT yd.standard_hours FROM freight.yard_distances yd
   WHERE yd.deleted_at IS NULL
     AND ((yd.from_yard_id = ${fromCol} AND yd.to_yard_id = ${toCol})
       OR (yd.from_yard_id = ${toCol} AND yd.to_yard_id = ${fromCol}))
   LIMIT 1
), ${stdRow('default_leg_standard_hours', 21)})`;

/** The schedule's own corridor, origin to destination. */
export const SCHEDULE_KM_EXPR = distanceKmBetween('ts.origin_station_id', 'ts.destination_station_id');

// ---------------------------------------------------------------------------
// Volume — TEU, charged and actual
// ---------------------------------------------------------------------------

/** Per-allocation aggregate over its container items. */
const containerItems = (selection: string): string => `(
  SELECT ${selection}
    FROM freight.wagon_allocation_container_items ci
    LEFT JOIN freight.container_types cty ON cty.id = ci.container_type_id
   WHERE ci.wagon_booking_allocation_id = wba.id AND ci.deleted_at IS NULL
)`;

/**
 * TEU for one allocation: a 40ft box is two twenty-foot equivalents, anything
 * else one.
 *
 * Note this is the third TEU derivation in the codebase and the only one taken
 * from the marshalling record. `revenue-classification.ts` derives TEU from the
 * charge code's size suffix (billing truth, blind to unsized codes) and
 * `wagon-teu-utilization.report.ts` from a wagon's currently pinned containers
 * (live state). This one answers "what did we actually move", which is what the
 * reporting spec asks for.
 */
export const ALLOC_TEU = containerItems(
  'COALESCE(SUM(CASE WHEN cty.size_ft >= 40 THEN 2 ELSE 1 END), 0)',
);
export const ALLOC_CONTAINERS_20 = containerItems('COUNT(*) FILTER (WHERE cty.size_ft = 20)');
export const ALLOC_CONTAINERS_40 = containerItems('COUNT(*) FILTER (WHERE cty.size_ft >= 40)');
export const ALLOC_CONTAINERS = containerItems('COUNT(*)');

export const TEU_EXPR = `COALESCE(SUM(${ALLOC_TEU}), 0)::int`;
export const CONTAINERS_EXPR = `COALESCE(SUM(${ALLOC_CONTAINERS}), 0)::int`;

/**
 * Actual volume — "loading capacity from marshalling" in the spec.
 * `allocated_weight_tons` is what the allocation flow recorded onto the wagon,
 * and is populated for every allocation in the system.
 */
export const ACTUAL_TONS_EXPR = 'COALESCE(SUM(wba.allocated_weight_tons), 0)::float8';

const IS_PERISHABLE = `COALESCE(ct.code, '') IN (${quote(PERISHABLE_CODES)})`;
const IS_BULK_LOAD = "wba.load_type <> 'CONTAINER'";

/**
 * Charged volume — the standard weight capacity the spec bills against, not
 * what was weighed.
 *
 * Containers are charged per box (20/40 tons laden, 2.24/3.88 empty). Bulk is
 * charged per WAGON (70 tons, or 38 for perishables), so it counts distinct
 * wagons rather than allocations: two bookings sharing one wagon are one
 * wagon's charge, not two.
 */
export const CHARGED_TONS_EXPR = `(
  COALESCE(SUM(
    CASE WHEN ${IS_BULK_LOAD} THEN 0 ELSE
      ${ALLOC_CONTAINERS_20} * CASE WHEN ${IS_EMPTY_CONTAINER}
        THEN ${stdRow('charged_tons_empty_20ft', 2.24)}
        ELSE ${stdRow('charged_tons_full_20ft', 20)} END
      + ${ALLOC_CONTAINERS_40} * CASE WHEN ${IS_EMPTY_CONTAINER}
        THEN ${stdRow('charged_tons_empty_40ft', 3.88)}
        ELSE ${stdRow('charged_tons_full_40ft', 40)} END
    END), 0)
  + COUNT(DISTINCT tsw.id) FILTER (WHERE ${IS_BULK_LOAD} AND ${IS_PERISHABLE})
    * ${stdAgg('charged_tons_per_wagon_perishable', 38)}
  + COUNT(DISTINCT tsw.id) FILTER (WHERE ${IS_BULK_LOAD} AND NOT ${IS_PERISHABLE})
    * ${stdAgg('charged_tons_per_wagon_general', 70)}
)::float8`;

/**
 * Wagons actually carrying cargo in the grouped set. The FILTER only bites on a
 * query built with `includeEmptyWagons` — every row of an allocation-grain
 * query has an allocation, so it is a no-op there.
 */
export const LOADED_WAGONS_EXPR =
  'COUNT(DISTINCT tsw.id) FILTER (WHERE wba.id IS NOT NULL)::int';

/**
 * Trainsets operated: wagons loaded divided by a full trainset for this cargo.
 * Seven full multimodal trains plus 30 of a 50-wagon set reads 7.6 — the
 * fraction the spec's worked example asks for.
 */
export const TRAINSETS_EXPR = `ROUND(
  COUNT(DISTINCT tsw.id)::numeric
  / NULLIF(MAX(COALESCE(ct.full_trainset_wagons, ${stdRow('default_full_trainset_wagons', 50)})), 0)
, 2)::float8`;

// ---------------------------------------------------------------------------
// Rates
// ---------------------------------------------------------------------------

/**
 * Implement rate — operated against plan, as a percentage.
 *
 * NULL when there is no plan, never 100 and never 0: an unplanned period has no
 * achievement to report, and coercing a missing plan to zero would read as
 * infinite achievement.
 */
export const implementRateExpr = (operated: string, planned: string): string =>
  `ROUND(100 * (${operated})::numeric / NULLIF(${planned}, 0), 1)::float8`;

/**
 * Turn-around implement rate, the spec's own formula:
 * `[((SC − AD) / SC) + 1] × 100`. Finishing exactly on standard scores 100;
 * a cycle an hour quicker than a 65-hour standard scores ~101.5.
 */
export const cycleRateExpr = (actual: string, standard: string): string =>
  `ROUND((((${standard}) - (${actual})) / NULLIF(${standard}, 0) + 1) * 100, 1)::float8`;

/** Hours between two timestamps, one decimal place. */
export const hoursBetween = (from: string, to: string): string =>
  `ROUND(EXTRACT(EPOCH FROM ((${to}) - (${from})))::numeric / 3600, 1)::float8`;

// ---------------------------------------------------------------------------
// Filters and the shared ledger
// ---------------------------------------------------------------------------

/**
 * The date every operations report buckets and filters on: when the train
 * actually left, falling back to the plan for a departure not yet dispatched.
 */
export const OPS_DATE = 'COALESCE(ts.actual_departure_at, ts.scheduled_departure_date)';

export const DIRECTION_FILTER: ReportFilterDef = {
  key: 'direction',
  label: 'Direction',
  type: 'select',
  options: [
    { value: 'IMPORT', label: 'Import' },
    { value: 'EXPORT', label: 'Export' },
    { value: 'DOMESTIC', label: 'Domestic' },
  ],
};

export const COUNTRY_FILTER: ReportFilterDef = {
  key: 'country',
  label: 'Country',
  type: 'select',
  options: [
    { value: 'Ethiopia', label: 'Ethiopia' },
    { value: 'Djibouti', label: 'Djibouti' },
  ],
};

/** Shared by every operations report, so they read the same way side by side. */
export const OPERATIONS_FILTERS: ReportFilterDef[] = [
  { key: 'date', label: 'Departure', type: 'daterange' },
  DIRECTION_FILTER,
  { key: 'trainNumber', label: 'Train No.', type: 'text' },
  { key: 'origin', label: 'Origin', type: 'select', optionsQuery: yardOptions },
  { key: 'destination', label: 'Destination', type: 'select', optionsQuery: yardOptions },
];

export const CARGO_CATEGORY_FILTER: ReportFilterDef = {
  key: 'categories',
  label: 'Cargo category',
  type: 'multiselect',
  options: CARGO_CATEGORIES,
};

/** Schedule states that never represent an operated train. */
const DEAD_SCHEDULE_STATUSES = ['DRAFT', 'CANCELLED'];

/**
 * Every operations report starts here: one wagon allocation, joined out to the
 * departure that carried it and the booking that explains it.
 *
 * The booking is LEFT joined — a wagon can be allocated before its booking data
 * is complete, and dropping those rows would understate wagon usage.
 *
 * `includeEmptyWagons` turns the ledger around to start from the wagon instead:
 * every wagon of the departure is a row, and one that carried nothing has a
 * NULL `wba`. Only the volume report wants that — it reports the empty wagons
 * as their own line — and it costs the other reports a row grain they would
 * have to filter back out.
 */
export function allocationLedgerQb(
  ctx: ReportContext,
  opts: { includeEmptyWagons?: boolean } = {},
): SelectQueryBuilder<ObjectLiteral> {
  const { params, directions } = ctx;

  const qb = ctx.ds.createQueryBuilder();

  if (opts.includeEmptyWagons) {
    qb.from(TrainSetWagon, 'tsw')
      .leftJoin(
        WagonBookingAllocation,
        'wba',
        'wba.train_set_wagon_id = tsw.id AND wba.deleted_at IS NULL',
      )
      .where('tsw.deleted_at IS NULL');
  } else {
    qb.from(WagonBookingAllocation, 'wba')
      .innerJoin(TrainSetWagon, 'tsw', 'tsw.id = wba.train_set_wagon_id AND tsw.deleted_at IS NULL')
      .where('wba.deleted_at IS NULL');
  }

  qb.innerJoin(TrainSchedule, 'ts', 'ts.train_set_id = tsw.train_set_id AND ts.deleted_at IS NULL')
    .leftJoin(Booking, 'b', 'b.id = wba.booking_id AND b.deleted_at IS NULL')
    .leftJoin(CargoType, 'ct', 'ct.id = b.cargo_type_id')
    .leftJoin(Yard, 'oy', 'oy.id = ts.origin_station_id')
    .leftJoin(Yard, 'dy', 'dy.id = ts.destination_station_id')
    .leftJoin(OperationsStandard, 'std', STANDARDS_JOIN)
    .andWhere('ts.status NOT IN (:...deadScheduleStatuses)', {
      deadScheduleStatuses: DEAD_SCHEDULE_STATUSES,
    });

  applyOperationsFilters(qb, params);
  applyDirectionScope(qb, 'COALESCE(b.trade_direction, ts.direction)', directions);
  return qb;
}

/**
 * The schedule-grain query, for reports that measure trains rather than cargo —
 * turnaround, delay, station stay. Same aliases, minus the allocation.
 */
export function scheduleLedgerQb(ctx: ReportContext): SelectQueryBuilder<ObjectLiteral> {
  const { params, directions } = ctx;

  const qb = ctx.ds
    .createQueryBuilder()
    .from(TrainSchedule, 'ts')
    .leftJoin(Yard, 'oy', 'oy.id = ts.origin_station_id')
    .leftJoin(Yard, 'dy', 'dy.id = ts.destination_station_id')
    .leftJoin(OperationsStandard, 'std', STANDARDS_JOIN)
    .where('ts.deleted_at IS NULL')
    .andWhere('ts.status NOT IN (:...deadScheduleStatuses)', {
      deadScheduleStatuses: DEAD_SCHEDULE_STATUSES,
    });

  applyOperationsFilters(qb, params);
  applyDirectionScope(qb, 'ts.direction', directions);
  return qb;
}

// ---------------------------------------------------------------------------
// Station stops — the stay, and the work done inside it
// ---------------------------------------------------------------------------

/**
 * A stay is an ARRIVED followed by the next DEPARTED at the same station by the
 * same physical train — NOT by the same schedule.
 *
 * When a train turns around at a station the two halves belong to different
 * departures: the arrival closes the inbound schedule and the departure opens
 * the outbound one. Pairing within a schedule finds only pass-through stops and
 * silently drops every turnaround, which is the longest stay a train makes.
 */
const TRAIN_KEY = 'COALESCE(tset.train_id::text, ts.train_set_id::text)';
const STAY_WINDOW = `PARTITION BY ${TRAIN_KEY}, ev.yard_id ORDER BY ev.occurred_at`;

/**
 * The loading window this stop's cargo actually took, read off the bookings
 * that boarded here.
 *
 * `bookings.loaded_at` is stamped per booking the moment a yard operator
 * presses Load, so the first and last of them bound the loading work without
 * anyone entering a second set of times. The bookings belong to the DEPARTING
 * schedule — a train arrives on one leg and loads for the next — which is why
 * this reads `departed_schedule_id` rather than the arrival's own schedule.
 *
 * There is no equivalent for unloading: `autoUnloadAtYard` stamps every
 * booking's `arrived_at` at the moment the checkpoint is logged, so a window
 * derived from it would collapse onto the arrival and report ~0 hours of
 * unloading. Unloading is only ever what staff recorded by hand.
 */
const derivedLoading = (agg: 'MIN' | 'MAX', scheduleExpr: string, yardExpr: string): string => `(
  SELECT ${agg}(b.loaded_at)
    FROM freight.bookings b
    JOIN freight.train_schedule_bookings tsb
      ON tsb.booking_id = b.id AND tsb.deleted_at IS NULL
     AND tsb.train_schedule_id = (${scheduleExpr})
   WHERE b.deleted_at IS NULL
     AND b.origin_yard_id = (${yardExpr})
     AND b.loaded_at IS NOT NULL
)`;

/**
 * The departing schedule and the yard are passed in rather than read off a
 * fixed alias: the stop-shaped query carries them as columns, while the
 * turnaround report reads raw `train_checkpoint_events` rows and works out the
 * departing leg from the cycle it already knows.
 */
export const loadingStartOn = (
  alias: string,
  scheduleExpr: string,
  yardExpr: string,
): string =>
  `COALESCE(${alias}.loading_started_at, ${derivedLoading('MIN', scheduleExpr, yardExpr)})`;
export const loadingEndOn = (alias: string, scheduleExpr: string, yardExpr: string): string =>
  `COALESCE(${alias}.loading_completed_at, ${derivedLoading('MAX', scheduleExpr, yardExpr)})`;

/** The stop-shaped query's own columns — what every stay-based report uses. */
const STOP_SCHEDULE = (alias: string): string => `${alias}.departed_schedule_id`;
const STOP_YARD = (alias: string): string => `${alias}.yard_id`;

/** Hand-recorded times win; the booking-derived window is the fallback. */
export const loadingStart = (alias: string): string =>
  loadingStartOn(alias, STOP_SCHEDULE(alias), STOP_YARD(alias));
export const loadingEnd = (alias: string): string =>
  loadingEndOn(alias, STOP_SCHEDULE(alias), STOP_YARD(alias));

/** Which of the two the loading columns came from, so nobody mistakes one for the other. */
export const loadingSource = (alias: string): string => `CASE
  WHEN ${alias}.loading_started_at IS NOT NULL
    OR ${alias}.loading_completed_at IS NOT NULL THEN 'Logged'
  WHEN ${derivedLoading('MIN', STOP_SCHEDULE(alias), STOP_YARD(alias))} IS NOT NULL THEN 'Derived'
  ELSE '—'
END`;

/**
 * When station work began and ended at a stop.
 *
 * LEAST and GREATEST ignore nulls, so a stop that only loaded (an export
 * origin) or only unloaded reports that half's window on its own, and where
 * both halves are known the pair spans exactly what the spec measures —
 * unloading start to loading end. NULL when nothing was logged or derived: an
 * unlogged stop has an unknown handling time, not a zero one.
 */
export const handlingStartOn = (
  alias: string,
  scheduleExpr: string,
  yardExpr: string,
): string =>
  `LEAST(${alias}.unloading_started_at, ${loadingStartOn(alias, scheduleExpr, yardExpr)})`;
export const handlingEndOn = (alias: string, scheduleExpr: string, yardExpr: string): string =>
  `GREATEST(${loadingEndOn(alias, scheduleExpr, yardExpr)}, ${alias}.unloading_completed_at)`;

export const handlingStart = (alias: string): string =>
  handlingStartOn(alias, STOP_SCHEDULE(alias), STOP_YARD(alias));
export const handlingEnd = (alias: string): string =>
  handlingEndOn(alias, STOP_SCHEDULE(alias), STOP_YARD(alias));

/** Total loading and unloading time at a stop, in hours. */
export const handlingHours = (alias: string): string =>
  hoursBetween(handlingStart(alias), handlingEnd(alias));

export const unloadingHours = (alias: string): string =>
  hoursBetween(`${alias}.unloading_started_at`, `${alias}.unloading_completed_at`);
export const loadingHours = (alias: string): string =>
  hoursBetween(loadingStart(alias), loadingEnd(alias));

/**
 * What the stay was spent on other than handling — the spec's "other activity".
 *
 * Stays NULL when handling was never logged rather than collapsing to the whole
 * stay, and floors at zero: handling logged slightly outside the arrival and
 * departure pair is a sloppy entry, not negative activity.
 */
export const otherActivityHours = (stayExpr: string, handlingExpr: string): string =>
  `CASE WHEN (${handlingExpr}) IS NULL THEN NULL
        ELSE ROUND(GREATEST((${stayExpr})::numeric - (${handlingExpr})::numeric, 0), 1)::float8 END`;

/**
 * Every logged stop, with the event that followed it at the same station and
 * whatever loading and unloading was recorded against it.
 *
 * Shared by the staying-time and loading-and-unloading reports so "a stop"
 * means one thing across the suite.
 */
export function stationStopsQb(ctx: ReportContext): SelectQueryBuilder<ObjectLiteral> {
  const { params, directions } = ctx;

  const qb = ctx.ds
    .createQueryBuilder()
    .from(TrainCheckpointEvent, 'ev')
    .innerJoin(TrainSchedule, 'ts', 'ts.id = ev.train_schedule_id AND ts.deleted_at IS NULL')
    .leftJoin(TrainSet, 'tset', 'tset.id = ts.train_set_id AND tset.deleted_at IS NULL')
    .innerJoin(Yard, 'y', 'y.id = ev.yard_id')
    .leftJoin(Yard, 'oy', 'oy.id = ts.origin_station_id')
    .leftJoin(Yard, 'dy', 'dy.id = ts.destination_station_id')
    .leftJoin(OperationsStandard, 'std', STANDARDS_JOIN)
    .where('ev.deleted_at IS NULL')
    .andWhere("ev.kind IN ('ARRIVED', 'DEPARTED')")
    .select('ts.train_number', 'train_number')
    .addSelect("COALESCE(y.label, y.code, '—')", 'station')
    .addSelect("COALESCE(y.code, '—')", 'station_code')
    .addSelect("COALESCE(y.country, '—')", 'country')
    .addSelect('ev.kind', 'kind')
    .addSelect('ev.yard_id', 'yard_id')
    .addSelect('ev.occurred_at', 'arrived_at')
    .addSelect(`lead(ev.occurred_at) OVER (${STAY_WINDOW})`, 'departed_at')
    .addSelect(`lead(ev.kind) OVER (${STAY_WINDOW})`, 'next_kind')
    // The leg the train LEAVES on, which is the one it loads for. A turnaround
    // departs on a different schedule than it arrived on, so the booking-derived
    // loading window has to follow this rather than `ev.train_schedule_id`.
    .addSelect(`lead(ev.train_schedule_id) OVER (${STAY_WINDOW})`, 'departed_schedule_id')
    // Handling rides the arrival row, which is the row a stay is built from.
    .addSelect('ev.unloading_started_at', 'unloading_started_at')
    .addSelect('ev.unloading_completed_at', 'unloading_completed_at')
    .addSelect('ev.loading_started_at', 'loading_started_at')
    .addSelect('ev.loading_completed_at', 'loading_completed_at')
    // Classed by the leg that ARRIVED. A stop whose inbound and outbound legs
    // differ in type is rare and reads as what pulled in.
    .addSelect(`CASE WHEN ${SCHEDULE_IS_CONTAINER} THEN 'Container' ELSE 'Bulk' END`, 'train_type')
    .addSelect(`ROUND(${STATION_STANDARD_HOURS_EXPR}, 1)`, 'standard_hours')
    .addSelect(`ROUND(${HANDLING_STANDARD_HOURS_EXPR}, 1)`, 'handling_standard_hours')
    .addSelect("COALESCE(ev.note, '')", 'note');

  if (params.dateFrom) qb.andWhere(`${OPS_DATE} >= :dateFrom`, { dateFrom: params.dateFrom });
  if (params.dateTo) qb.andWhere(`${OPS_DATE} < :dateTo`, { dateTo: params.dateTo });
  if (params.direction) qb.andWhere('ts.direction = :direction', { direction: params.direction });
  if (params.trainNumber) {
    qb.andWhere('ts.train_number ILIKE :trainNumber', {
      trainNumber: `%${params.trainNumber as string}%`,
    });
  }
  if (params.station) qb.andWhere('y.code = :station', { station: params.station });
  if (params.country) qb.andWhere('y.country = :country', { country: params.country });
  // Whitelisted, not bound: the same EXISTS has to read identically in the
  // SELECT above, and a bound parameter cannot be reused across both.
  if (params.trainType === 'CONTAINER') qb.andWhere(SCHEDULE_IS_CONTAINER);
  if (params.trainType === 'BULK') qb.andWhere(`NOT ${SCHEDULE_IS_CONTAINER}`);

  applyDirectionScope(qb, 'ts.direction', directions);
  return qb;
}

export const TRAIN_TYPE_FILTER: ReportFilterDef = {
  key: 'trainType',
  label: 'Train type',
  type: 'select',
  options: [
    { value: 'CONTAINER', label: 'Container' },
    { value: 'BULK', label: 'Bulk' },
  ],
};

/** Only completed stops — an arrival whose departure was also logged, alias `s`. */
export function stationStaysQb(ctx: ReportContext): SelectQueryBuilder<ObjectLiteral> {
  const inner = stationStopsQb(ctx);
  return ctx.ds
    .createQueryBuilder()
    .from(`(${inner.getQuery()})`, 's')
    .setParameters(inner.getParameters())
    .where("s.kind = 'ARRIVED'")
    .andWhere("s.next_kind = 'DEPARTED'");
}

export function applyOperationsFilters(
  qb: SelectQueryBuilder<ObjectLiteral>,
  params: Record<string, unknown>,
): void {
  if (params.dateFrom) qb.andWhere(`${OPS_DATE} >= :dateFrom`, { dateFrom: params.dateFrom });
  if (params.dateTo) qb.andWhere(`${OPS_DATE} < :dateTo`, { dateTo: params.dateTo });
  if (params.direction) qb.andWhere('ts.direction = :direction', { direction: params.direction });
  if (params.trainNumber) {
    qb.andWhere('ts.train_number ILIKE :trainNumber', {
      trainNumber: `%${params.trainNumber as string}%`,
    });
  }
  if (params.origin) qb.andWhere('oy.code = :origin', { origin: params.origin });
  if (params.destination) qb.andWhere('dy.code = :destination', { destination: params.destination });
}

/**
 * Restricts an allocation-grain query to a set of cargo categories. Kept
 * separate from {@link applyOperationsFilters} because the schedule-grain
 * query has no cargo to filter by.
 */
export function applyCategoryFilter(
  qb: SelectQueryBuilder<ObjectLiteral>,
  params: Record<string, unknown>,
  categoryExpr: string = CARGO_CATEGORY_EXPR,
): void {
  const categories = params.categories as string[] | null;
  if (categories?.length) {
    qb.andWhere(`${categoryExpr} IN (:...categories)`, { categories });
  }
}

/**
 * Appended to every plan-versus-actual report's description, because neither
 * the re-bucketing nor the catch-up rule is guessable from the table.
 */
export const PLAN_GRANULARITY_NOTE =
  ' A plan is spread evenly across its own period and re-gathered into whichever bucket ' +
  'the report shows, so a monthly target fills a quarter or a year exactly, and a daily ' +
  'or weekly view gets its share of it. A week that straddles two months draws on both. ' +
  'Plan is the committed figure and never moves. Required is the same target treated as a ' +
  'quota: whatever is still outstanding, spread across the time still left, so a period ' +
  'that fell behind raises what the periods after it must carry. A target already met in ' +
  'full requires nothing further. No target carries a route, a train or a direction, so ' +
  'filtering by one leaves the plan columns empty rather than comparing a corridor’s whole ' +
  'target against one slice of its work.';

/**
 * The user's date filter as open-ended bounds, so the clipping arithmetic below
 * never has to branch on null.
 */
const PLAN_FROM = "COALESCE(CAST(:planFrom AS timestamptz), '-infinity'::timestamptz)";
const PLAN_TO = "COALESCE(CAST(:planTo AS timestamptz), 'infinity'::timestamptz)";

/**
 * How long one target's period runs. A target's span is exact — 90 days is 90
 * days — and need not line up with the ragged year-end display blocks the
 * `nine_month` and `ninety_day` granularities produce. The spread below is
 * proportional, so partial overlap resolves correctly either way.
 */
const TARGET_SPAN = `CASE ot.period_type
  WHEN 'day' THEN INTERVAL '1 day'
  WHEN 'week' THEN INTERVAL '7 days'
  WHEN 'month' THEN INTERVAL '1 month'
  WHEN 'quarter' THEN INTERVAL '3 months'
  WHEN 'half_year' THEN INTERVAL '6 months'
  WHEN 'nine_month' THEN INTERVAL '9 months'
  WHEN 'ninety_day' THEN INTERVAL '90 days'
  WHEN 'year' THEN INTERVAL '1 year'
  ELSE INTERVAL '1 day'
END`;

/**
 * The planned rows for a metric, as a derived table: one row per bucket per
 * planned key, carrying both a committed and a required figure.
 *
 * **Plan** — a target is a rate over its own period, not a lump at its start.
 * The committed value is spread evenly across the days it covers and
 * re-gathered into the report's buckets, so three monthly targets add up to a
 * quarter exactly, a daily view gets a thirty-first of the month, and a week
 * straddling a month boundary draws proportionally on both. The even spread is
 * an assumption, and the only one available: a monthly figure carries no
 * information about which days inside it were busier. This number never moves —
 * Implement Rate is measured against it, so a month that missed keeps reading
 * as a month that missed.
 *
 * **Required** — the same target read as a quota. At each bucket, whatever is
 * still outstanding (committed minus everything delivered in earlier buckets)
 * is spread across the time still left in the period. A year 20% met at the
 * halfway mark asks the remaining months for the other 80%. Over-delivery
 * clamps to zero rather than going negative: a met quota requires nothing more.
 *
 * `actualsSql` must produce `(bucket, act_key, act_category, actual)` and must
 * be built **without the user's date bounds** — see {@link attainmentCtx}.
 * Attainment is a fact about the target's whole period; measuring it through
 * the report's date filter would read a mid-year view as "nothing delivered
 * yet" and demand the entire year's work from one month.
 *
 * The reports FULL OUTER JOIN this to their operated aggregate so a category
 * that was planned but never ran still appears, at zero. The OCC monthly report
 * does exactly that — Nagad–Dire Dawa is planned 2,106 t and operated none, and
 * publishes as 0%. Dropping the row would hide a total miss, which is the one
 * thing a plan-versus-actual table exists to show.
 *
 * Period bounds ride on `:planFrom` / `:planTo`, which the caller must bind
 * with {@link plannedRowsParams} — they come from the user's date filter.
 */
/**
 * The plan side of a plan-versus-actual report has to obey the same cargo
 * filter the operated side does. Without it the FULL OUTER JOIN re-introduces
 * every planned key the user filtered out, as a row of zeros.
 *
 * Values are whitelisted against the vocabulary and inlined rather than bound:
 * this fragment is assembled into raw CTE text, and the filter params are not
 * validated upstream. An unknown value matches nothing — same as it does on the
 * operated side, where the CASE can never emit it.
 */
const planKeyFilter = (dimension: string, params: Record<string, unknown>): string => {
  const isClass = dimension === 'container_class';
  const selected = (isClass ? params.classes : params.categories) as string[] | null;
  if (!selected?.length) return '';
  const vocab = isClass ? CONTAINER_CLASSES : CARGO_CATEGORIES;
  const valid = selected.filter((v) => vocab.some((o) => o.value === v));
  // A station target is keyed on the yard and carries its cargo type alongside.
  const column = dimension === 'station' ? 'ot.cargo_category' : 'ot.dimension_key';
  return valid.length ? `AND ${column} IN (${quote(valid)})` : 'AND FALSE';
};

/**
 * Filters that narrow the operated population below the grain any target is
 * kept at. No target carries a route, a train or a direction, so a plan read
 * beside a route-filtered actual is the whole corridor's plan sitting next to
 * one slice of its work — the implement rate then reads as a miss that never
 * happened.
 *
 * There is no honest number to show, so the plan side reports nothing at all
 * and Implement Rate goes NULL, the same way it does for a period with no
 * target. `country` is absent on purpose: on a station plan it is a property of
 * the planned key itself, and {@link planCountryFilter} narrows rather than
 * suppresses.
 */
const PLAN_GRAIN_BREAKERS = ['origin', 'destination', 'trainNumber', 'direction'];

const planGrainFilter = (params: Record<string, unknown>): string =>
  PLAN_GRAIN_BREAKERS.some((key) => params[key]) ? 'AND FALSE' : '';

/**
 * A station target is keyed on a yard code, so the country filter — which
 * decides which end of the corridor the report calls "the station" — is a real
 * predicate on the plan, not a grain break. Without it the Djibouti view lists
 * every Ethiopian station's target as a row that moved nothing.
 */
const planCountryFilter = (dimension: string, params: Record<string, unknown>): string => {
  if (dimension !== 'station') return '';
  const country = COUNTRY_FILTER.options?.find((o) => o.value === params.country)?.value;
  if (!country) return '';
  return `AND EXISTS (SELECT 1 FROM freight.yards y
                       WHERE y.code = ot.dimension_key AND y.deleted_at IS NULL
                         AND y.country = '${country}')`;
};

export const plannedRowsSql = (
  metric: string,
  dimension: string,
  params: Record<string, unknown>,
  actualsSql: string,
): string => {
  const unit = resolvePeriod(params);
  // Reused verbatim in the GROUP BY, per the trap documented on `periodExpr`.
  const bucketOf = unit.truncOn('d.day');
  return `
  WITH tgt AS (
    SELECT ot.id,
           ot.dimension_key,
           ot.cargo_category,
           ot.planned_value,
           ot.period_start::timestamptz AS starts,
           ot.period_start::timestamptz + ${TARGET_SPAN} AS ends
      FROM freight.operations_targets ot
     WHERE ot.deleted_at IS NULL
       AND ot.metric = '${metric}'
       AND ot.dimension = '${dimension}'
       AND ot.planned_value > 0
       ${planKeyFilter(dimension, params)}
       ${planCountryFilter(dimension, params)}
       ${planGrainFilter(params)}
  ),
  -- One row per target per bucket. Generated a day at a time rather than a
  -- bucket at a time: the ragged units restart their blocks each January, so
  -- stepping by the unit's own width walks off the anchor in the second year.
  -- Day grain also makes a bucket that only partly overlaps the target fall out
  -- for free, at the same sub-day precision the clipping used before.
  spread AS (
    SELECT t.id,
           t.dimension_key,
           t.cargo_category,
           t.planned_value,
           EXTRACT(EPOCH FROM (t.ends - t.starts)) AS secs_total,
           ${bucketOf} AS bucket,
           SUM(GREATEST(0, EXTRACT(EPOCH FROM (
             LEAST(d.day + INTERVAL '1 day', t.ends)
             - GREATEST(d.day, t.starts))))) AS secs_full,
           SUM(GREATEST(0, EXTRACT(EPOCH FROM (
             LEAST(d.day + INTERVAL '1 day', t.ends, ${PLAN_TO})
             - GREATEST(d.day, t.starts, ${PLAN_FROM}))))) AS secs_in
      FROM tgt t
      CROSS JOIN LATERAL generate_series(
        date_trunc('day', t.starts),
        t.ends - INTERVAL '1 microsecond',
        INTERVAL '1 day'
      ) AS d(day)
     GROUP BY t.id, t.dimension_key, t.cargo_category, t.planned_value,
              t.starts, t.ends, ${bucketOf}
  ),
  -- secs_before and actual_before are strictly-preceding running sums, so a
  -- bucket's requirement is decided by what happened before it, never by its
  -- own result. The frame is spelled out rather than defaulted: the default
  -- RANGE frame would fold peer rows into the current one.
  cascaded AS (
    SELECT s.*,
           COALESCE(SUM(s.secs_full) OVER prior, 0) AS secs_before,
           COALESCE(SUM(a.actual) OVER prior, 0) AS actual_before
      FROM spread s
      LEFT JOIN (${actualsSql}) a
        ON a.bucket = s.bucket
       AND a.act_key = s.dimension_key
       AND a.act_category IS NOT DISTINCT FROM s.cargo_category
    WINDOW prior AS (
      PARTITION BY s.id ORDER BY s.bucket
      ROWS BETWEEN UNBOUNDED PRECEDING AND 1 PRECEDING
    )
  )
  SELECT ${unit.labelOn('c.bucket')} AS period,
         c.dimension_key AS plan_key,
         c.cargo_category AS plan_category,
         SUM(c.planned_value * c.secs_in / NULLIF(c.secs_total, 0)) AS plan_value,
         SUM(GREATEST(0, c.planned_value - c.actual_before)
             * c.secs_in / NULLIF(c.secs_total - c.secs_before, 0)) AS plan_required
    FROM cascaded c
   WHERE c.secs_in > 0
   GROUP BY 1, 2, 3`;
};

/**
 * The report's own ledger with the user's date bounds removed, for the
 * attainment series {@link plannedRowsSql} cascades from. Every other filter
 * stays applied, so the catch-up figure is measured on the same population as
 * the `operated` column it sits beside.
 */
export const attainmentCtx = (ctx: ReportContext): ReportContext => ({
  ...ctx,
  params: { ...ctx.params, dateFrom: null, dateTo: null },
});

/** The bindings {@link plannedRowsSql} expects. */
export const plannedRowsParams = (
  params: Record<string, unknown>,
): Record<string, unknown> => ({
  planFrom: params.dateFrom ?? null,
  planTo: params.dateTo ?? null,
});