// ---------------------------------------------------------------------------
// customer-grid / rankOps.ts
// Wave 2026-08-02 item 10 (contract C-OPS) — the RANK operators, pure.
//
// Every other filter operator answers a question about ONE cell: `revenue > 5000`
// is true or false for a row on its own. A rank operator cannot be answered that
// way at all — "is this customer in the top 10" depends on which OTHER rows are
// in the question. That is the whole design problem, and it is why this is a
// separate resolution PASS rather than eight more cases in `matchFilter`:
//
//   1. strip every rank leaf from the tree                    (stripRankLeaves)
//   2. evaluate the residue -> that match-set is the DOMAIN   (the caller does)
//   3. rank the domain, once per column, and hand each leaf its slice (this file)
//   4. evaluate the FULL tree, rank leaves as membership tests (useVisibleRows)
//
// So "top 10 of agent Naomi" narrows by agent FIRST and then takes ten, which is
// what the sentence says and not what a naive per-row implementation would do.
//
// THE FAIL-CLOSED LAW, restated because a rank leaf is where it bites hardest: a
// leaf this file cannot answer gets an EMPTY set, never a missing one. A missing
// set would land in `evalNode`'s "no answer" branch, and the tempting reading of
// "no answer" is "no narrowing" — which shows every record under a count nobody
// would doubt. The unanswerable ones are COUNTED instead, so the empty table has
// a marker beside it saying why.
// ---------------------------------------------------------------------------

import type { Field, FieldType, FilterNode, FilterRule, Row } from "./types";
import { isFilterGroup, isMeasureRule, isRankOp, isRuleActive } from "./types";

/** `leaf -> the pids that satisfy it`. Keyed by the leaf OBJECT: a rank leaf carries no id of
 *  its own (only measure rules do), and keying by position would silently re-associate every
 *  answer the moment a condition was deleted — the trap `FilterRule.id` exists to avoid. The
 *  tree is stable for the life of one pipeline run, which is exactly this map's lifetime. */
export type RankSets = Map<FilterRule, Set<number>>;

export interface RankResolution {
  sets: RankSets;
  /**
   * ACTIVE rank leaves that could not be answered AT ALL — the same distinction
   * `unresolvedConditions` draws for dates and cohorts, and for the same reason: they match
   * nothing, and "0 records" with no marker is indistinguishable from a filter that genuinely
   * matches nobody. Three causes, all of them counted:
   *   - the column is gone, or is not of a type that can be ranked
   *   - the column has no numeric value ANYWHERE in a non-empty domain (a measure column
   *     whose values have not arrived — cold store, or a window still resolving)
   *   - the operator's value is junk (a saved view holding `inQuartile 9`)
   */
  unanswerable: number;
}

/** The resolution of a tree with no rank leaves in it. Shared, never mutated. */
export const NO_RANKING: RankResolution = { sets: new Map(), unanswerable: 0 };

/**
 * Types a rank operator may be applied to.
 *
 * The engine's own copy, deliberately — `useVisibleRows.isNumericType` carries the same list
 * with the same comment, because that file is the self-contained reference the Python port
 * mirrors. Two copies pinned by a gate beat one import that quietly widens both.
 */
function isRankableType(t: FieldType): boolean {
  return t === "currency" || t === "int" || t === "pct" || t === "rating" || t === "formula";
}

/**
 * True when this leaf asks a RANK question.
 *
 * A MEASURE leaf is excluded by construction even if it somehow carried a rank op: a measure
 * condition is answered by a server-side pid set, so ranking it client-side would be a second,
 * disagreeing answer to the same question. (The builder cannot produce one — measure leaves
 * offer `MEASURE_OPS` — so this is the backstop, not the behaviour.)
 */
export function isRankRule(node: FilterNode): boolean {
  return (
    !isFilterGroup(node) && !isMeasureRule(node) && isRankOp((node as FilterRule).op)
  );
}

/** Does this tree contain ANY rank leaf, active or not? The pipeline's cheap early-out: with
 *  no rank leaf, not one line of this file runs and the engine is byte-for-byte as shipped. */
export function hasRankLeaf(nodes: FilterNode[]): boolean {
  for (const node of nodes ?? []) {
    if (isFilterGroup(node)) {
      if (hasRankLeaf(node.children)) return true;
      continue;
    }
    if (isRankRule(node)) return true;
  }
  return false;
}

/**
 * The tree with every rank leaf removed — the question whose answer is the ranking DOMAIN.
 *
 * A group left with no children is DROPPED rather than kept empty. Both behave identically
 * today (`evalNode` returns null for a group with no active children, and null is ignored),
 * but an empty group is a shape the rest of the engine never otherwise sees, and shapes that
 * only appear in one code path are how a future edit acquires a special case.
 */
export function stripRankLeaves(nodes: FilterNode[]): FilterNode[] {
  const out: FilterNode[] = [];
  for (const node of nodes ?? []) {
    if (isFilterGroup(node)) {
      const children = stripRankLeaves(node.children);
      if (children.length) out.push({ conj: node.conj, children });
      continue;
    }
    if (isRankRule(node)) continue;
    out.push(node);
  }
  return out;
}

/** Every ACTIVE rank leaf, in tree order. Inactive ones (a `topN` with no N typed yet) are
 *  left out for the same reason every other half-typed condition is: they are not asking. */
export function activeRankLeaves(nodes: FilterNode[]): FilterRule[] {
  const out: FilterRule[] = [];
  for (const node of nodes ?? []) {
    if (isFilterGroup(node)) {
      out.push(...activeRankLeaves(node.children));
      continue;
    }
    if (!isRankRule(node)) continue;
    const rule = node as FilterRule;
    if (isRuleActive(rule)) out.push(rule);
  }
  return out;
}

/** One domain row reduced to what ranking needs. */
interface Ranked {
  pid: number;
  v: number;
}

/**
 * The domain's rows that HAVE a value for `colId`, ordered.
 *
 * ⚠ `Number(raw)` is used and `toNum` is NOT: the engine's coercion helper turns anything
 * unparseable into 0, which is right for a comparison (`x > 5` against a blank is false either
 * way) and catastrophic here — every blank customer would enter the ranking as a real zero and
 * fill the bottom of every "bottom 10". A blank is not a zero; it is an absence, and an absence
 * has no rank. Note that a genuine 0 stays in: `raw === ""` is the test, not falsiness.
 *
 * The tie-break is `pid` ASCENDING in BOTH directions, never a reversal of the whole order:
 * "keep exactly N, tie-break pid ascending" has to mean the same N whichever end you ask from,
 * or "top 10" and "bottom 10" of a ten-row table would not be the same ten rows.
 */
function order(domain: Row[], colId: string, dir: "asc" | "desc"): Ranked[] {
  const out: Ranked[] = [];
  for (const r of domain) {
    const raw = r[colId];
    if (raw == null || raw === "") continue;
    const v = typeof raw === "number" ? raw : Number(raw);
    if (!Number.isFinite(v)) continue;
    out.push({ pid: r.pid, v });
  }
  out.sort((a, b) => (a.v === b.v ? a.pid - b.pid : dir === "asc" ? a.v - b.v : b.v - a.v));
  return out;
}

/** The leaf's rhs as a whole number inside [lo, hi], or null when it is junk. Null is an
 *  UNANSWERABLE leaf, never a defaulted one — silently reading `inQuartile 9` as 4 would answer
 *  a question the view does not ask. */
function bound(value: string, lo: number, hi: number): number | null {
  const n = Number(value);
  if (!Number.isFinite(n)) return null;
  const i = Math.trunc(n);
  if (i !== n || i < lo || i > hi) return null;
  return i;
}

/**
 * Which of `k` equal slices of the RANK AXIS row `index` falls in. Bucket 1 is the lowest
 * slice and bucket `k` the highest, so `inQuartile 4` is the top quarter — which is how the
 * operator reads.
 *
 * `floor(i * k / n) + 1` — the slice of [0, 1) that i/n lands in — rather than the literal
 * `floor(i / ceil(n / k)) + 1` the contract's "ceil split" first suggests. The difference is
 * not cosmetic: fixed ceil-sized buckets leave the TOP one EMPTY whenever n sits a little
 * above a multiple of k (six rows into quarters gives 2/2/2/0), so "the top quarter" would
 * match nobody while three lower quarters were full. This form gives 2/1/2/1 — every bucket
 * non-empty, sizes differing by at most one. Amendment booked in the wave doc.
 *
 * Buckets are by RANK INDEX, not by value: two rows sharing a value can land either side of a
 * boundary, broken by pid ascending. That is what "equal-size buckets" means and it is the
 * contract's explicit choice over value interpolation — said here because the builder's helper
 * line has to say it to the user in one sentence.
 */
function slice(index: number, n: number, k: number): number {
  return Math.min(k, Math.floor((index * k) / n) + 1);
}

/** How many rows `pct` percent of `n` is. Ceil: "the top 10%" of 43 must not be 4.3, and
 *  rounding DOWN would make the top 1% of anything under 100 rows match nobody. */
function pctCount(n: number, pct: number): number {
  return Math.min(n, Math.max(1, Math.ceil((n * pct) / 100)));
}

/**
 * Resolve every ACTIVE rank leaf against the domain.
 *
 * `domain` is the rows matching everything EXCEPT the rank leaves — the caller computes it, so
 * that this file never needs to know how a cohort or a measure condition is answered.
 */
export function resolveRankLeaves(
  filters: FilterNode[],
  domain: Row[],
  fieldByKey: Map<string, Field>
): RankResolution {
  const leaves = activeRankLeaves(filters);
  if (!leaves.length) return NO_RANKING;

  const sets: RankSets = new Map();
  let unanswerable = 0;
  // Several leaves can name the same column ("top 10" AND "above average" by revenue), and the
  // sort is the expensive half. Cached per column PER DIRECTION — see `order`.
  const cache = new Map<string, Ranked[]>();
  const ordered = (colId: string, dir: "asc" | "desc"): Ranked[] => {
    const ck = `${dir}:${colId}`;
    let list = cache.get(ck);
    if (!list) {
      list = order(domain, colId, dir);
      cache.set(ck, list);
    }
    return list;
  };
  const refuse = (leaf: FilterRule) => {
    sets.set(leaf, new Set());
    unanswerable += 1;
  };

  for (const leaf of leaves) {
    const field = fieldByKey.get(leaf.colId);
    // A rank op on a column that is gone, or on text/date/select. NOT the engine's usual
    // "an operator that does not apply to this type is always-true" case, and the divergence
    // is deliberate: always-true is safe for `contains` on a number (it narrows nothing and
    // asks nothing), but a rank operator is a NARROWING question, and answering "everybody"
    // to "who is in the top ten" is the widening sin wearing a plausible face.
    if (!field || !isRankableType(field.type)) {
      refuse(leaf);
      continue;
    }

    const asc = ordered(leaf.colId, "asc");
    if (!asc.length) {
      // No value anywhere. With an EMPTY domain that is just "your other conditions matched
      // nobody" — already explained by the rows — so it is not counted twice. With a populated
      // domain it means the column itself has no values: a measure column whose derived cells
      // have not arrived, which is precisely the state that must not read as "top 10 of
      // nothing = everything".
      sets.set(leaf, new Set());
      if (domain.length) unanswerable += 1;
      continue;
    }

    const n = asc.length;
    let keep: Ranked[] | null = null;
    switch (leaf.op) {
      case "topN": {
        const k = bound(leaf.value, 1, 10000);
        keep = k == null ? null : ordered(leaf.colId, "desc").slice(0, k);
        break;
      }
      case "bottomN": {
        const k = bound(leaf.value, 1, 10000);
        keep = k == null ? null : asc.slice(0, k);
        break;
      }
      case "inTopPct": {
        const p = bound(leaf.value, 1, 100);
        keep = p == null ? null : ordered(leaf.colId, "desc").slice(0, pctCount(n, p));
        break;
      }
      case "inBottomPct": {
        const p = bound(leaf.value, 1, 100);
        keep = p == null ? null : asc.slice(0, pctCount(n, p));
        break;
      }
      case "aboveAvg":
      case "belowAvg": {
        // The mean of the DOMAIN's values, not of the whole table — the same residue rule
        // every other rank op follows, so "above average, among agent Naomi's customers"
        // means what it says.
        const mean = asc.reduce((sum, e) => sum + e.v, 0) / n;
        keep = asc.filter((e) => (leaf.op === "aboveAvg" ? e.v > mean : e.v < mean));
        break;
      }
      case "inQuartile":
      case "inDecile": {
        const k = leaf.op === "inQuartile" ? 4 : 10;
        const want = bound(leaf.value, 1, k);
        // FEWER ROWS THAN BUCKETS is a question the data cannot answer, and it is the one
        // place where inventing a definition would be indefensible: with six ranked rows and
        // ten deciles, SOME decile is empty whatever the rule, so any formula chooses which
        // asks go unanswered. Refusing says so out loud (and gets counted) instead of quietly
        // returning nobody for "the top decile" and everybody's guess for the rest.
        keep = want == null || n < k ? null : asc.filter((_, i) => slice(i, n, k) === want);
        break;
      }
      default:
        // A rank op this file does not implement. Unreachable through `RANK_OPS`, and it
        // REFUSES rather than falling through — the one behaviour that keeps adding an
        // operator to the vocabulary from silently widening every view that uses it.
        keep = null;
    }

    if (keep == null) {
      refuse(leaf);
      continue;
    }
    sets.set(leaf, new Set(keep.map((e) => e.pid)));
  }

  return { sets, unanswerable };
}