"""rollup_sql.py — the READ-THROUGH rollup: one grouped SQL query answers every parent row.

⛔ WHY THIS EXISTS. A linked rollup folds rows that live in the store, and `MAX_ROWS` bounds them
for a measured reason: every `ut_*` table lives inside ONE `user_tables.json` that is parsed and
copied on essentially every request. ⚠ THE CAP MOVED TO 60,000 ON 2026-08-09 AND THAT DID NOT
RETIRE THIS MODULE — the arithmetic is why. (This line said 100,000 until wave 28: that number was
a candidate REJECTED during its own derivation, and three files repeated it for a day. A comment
naming a cap is a claim about a constant eight lines away — check it, do not copy it.)
Measured against tenant #0's own mirror, Royal's
**256,810 order lines are 63.9 MB and 2.57 s per copy**: 4.3x the new cap and 8x any tolerable
per-request cost. Four Odoo databases that DO fit (customers, products, invoices, orders) already
total 17.81 MB / ~320 ms. Order lines never live in that document, at any cap.

⭐ SO THIS KIND NEVER COPIES THE ROWS AT ALL. It names a governed semantic TOPIC and a METRIC KEY,
and `store_query(..., group_by=[dim])` answers EVERY parent in one pass against DuckDB.
MEASURED: 1,748 customers over 256,810 order lines in **232 ms**. The row count stops being the
product's problem and becomes the database's, which is the whole of D-87's "read-through binding".

⚠ A METRIC KEY, NEVER A FILTER FRAGMENT — the decision this module is built around.
`model/metrics/*.yml` already carries each metric's scope, its `store_filter_sql` AND the matching
`live_domain`, whose own comment reads *"BOTH or store_parity compares two different questions"*.
Binding a rollup to the KEY inherits the scope and the live-parity oracle for free. Letting a
rollup carry SQL would mint a second definition of a number the semantic layer exists to define
once, and the two would drift in silence.

⛔ TRUNCATION IS A WRONG NUMBER HERE, NOT A SHORT LIST. A row window is honest because counts and
totals are computed over the full scope beside it; a GROUP window has no such companion — the
groups ARE the answer. `store_query` now reports `truncated`, and this module REFUSES to write a
single cell when it is set. Half a rollup is worse than none: it looks finished.
"""
import sys
from pathlib import Path

_HERE = Path(__file__).resolve().parent
_PLATFORM = _HERE.parents[1] / "platform"
for _p in (str(_HERE), str(_PLATFORM)):
    if _p not in sys.path:
        sys.path.insert(0, _p)


class RollupSourceError(Exception):
    """The rollup could not be computed HONESTLY — no cells are written when this is raised."""


def _ut():
    import core.user_tables as user_tables
    return user_tables


def source_fields(table_def):
    """Every source-backed rollup field on a table definition, in declaration order."""
    out = []
    for f in ((table_def or {}).get("fields") or []):
        if not isinstance(f, dict) or f.get("type") != "rollup":
            continue
        bag = (f.get("rollup") or {}).get("source")
        if isinstance(bag, dict) and bag.get("topic") and bag.get("measure"):
            out.append(f)
    return out


def group_values(bag, today=None):
    """`{group_id: value}` for one source bag — ONE grouped query over the whole scope.

    ⚠ Keyed by the dim's `_id` column, never its label. Two customers can share a display name;
    `partner_id` is what the parent row actually joins on.
    """
    from harness import semantic as sem
    from harness import windows as W

    topic = str(bag.get("topic") or "")
    measure = str(bag.get("measure") or "")
    dim = str(bag.get("groupBy") or "")
    window = str(bag.get("window") or "").strip().lower()

    date_from = date_to = None
    if window and window != "all_time":
        # ⚠ RESOLVED HERE, AGAINST `today`, not stored as a date pair. `harness.windows` owns what
        # "ytd" means so there is one implementation; a literal year-start baked into the field
        # would be right until 1 January and wrong afterwards with nothing to notice.
        rng = W.resolve({"kind": window}, today or _today())
        if rng is None:
            raise RollupSourceError(
                f"window {window!r} could not be resolved — refusing rather than widening this "
                f"rollup to all time")
        date_from, date_to = rng

    try:
        res = sem.store_query(topic, [measure], group_by=[dim],
                              date_from=date_from, date_to=date_to,
                              limit=sem.MAX_GROUPS, today=today)
    except Exception as e:                                            # noqa: BLE001
        raise RollupSourceError(f"{type(e).__name__}: {e}") from e

    if res.get("truncated"):
        # ⛔ THE REFUSAL THAT MATTERS. See the module header: a truncated GROUP set is a wrong
        # answer per parent, not a shortened one, and it would look completely normal on screen.
        raise RollupSourceError(
            f"{topic}/{measure} grouped by {dim} exceeded {sem.MAX_GROUPS} groups — refusing to "
            f"write cells from a truncated result")

    # ⛔ THE KEY COLUMN IS NOT ALWAYS `<dim>_id`, and reading it unconditionally was a SILENT
    # BLANK. `store_query` emits `<dim>_id` only when the dim declares a `name_col`
    # (`f"{d['col']} AS {g}_id" if d.get("name_col") else f"{d['col']} AS {g}"`) — an id plus a
    # display name. A dim whose value IS its own label emits the bare `<dim>` instead, and two
    # shipped dims are that shape: `receivables.payment_state` and `gl_lines.account_type`.
    # Against either, this returned an EMPTY map and every cell was written blank — a configured
    # column that computes nothing, with no error anywhere. `payment_state` is not hypothetical:
    # it is a real column on `ut_odoo_invoices` and an obvious thing to roll up against.
    # ⚠ `_id` is tried FIRST so an id-keyed dim can never be keyed by its display name, which
    # would silently merge two customers who share one.
    id_col = f"{dim}_id"
    key_col = id_col if any(id_col in (r or {}) for r in (res.get("rows") or [])) else dim
    out = {}
    for row in res.get("rows") or []:
        gid = row.get(key_col)
        if gid is None:
            continue
        out[str(gid)] = row.get(measure)
    return out


def _today():
    import datetime as _dt
    return _dt.date.today().strftime("%Y-%m-%d")


def compute(rt, table_key, today=None, tables=None):
    """Write every source-backed rollup cell on `table_key`. Returns `{field_key: cells_written}`.

    `tables` (a live `user_tables` dict) is the gate's injection point — the same shape
    `automation_engine.compute_relation_cells` takes, so this can be proven against a fixture
    without a store.
    """
    ut = _ut()
    owned = tables is not None
    blob = tables if owned else (rt.get(ut.STORE_KEY) or {})
    tdef = (blob or {}).get(str(table_key)) or {}
    fields = source_fields(tdef)
    if not fields:
        return {}

    stamp = today or _today()
    plans = []
    for f in fields:
        bag = f["rollup"]["source"]
        # ⚠ Resolved BEFORE anything is written. A rollup that refuses must leave every cell as it
        # was — a partially applied pass would mix two vintages of the same column.
        plans.append((f["key"], str(bag.get("on") or ""), group_values(bag, today=stamp)))

    written = {}

    def _apply(cur):
        t = cur.get(str(table_key))
        if t is None:
            return cur
        rows = t.setdefault("rows", {})
        for fkey, on, values in plans:
            n = 0
            for row in rows.values():
                if not isinstance(row, dict):
                    continue
                join = str(row.get(on) or "").strip()
                if not join:
                    continue
                # ⚠ int-ish join keys arrive as "5280" or 5280 depending on who wrote them.
                hit = values.get(join)
                if hit is None and join.endswith(".0"):
                    hit = values.get(join[:-2])
                row[fkey] = "" if hit is None else str(hit)
                n += 1
            written[fkey] = n
        return cur

    if owned:
        _apply(blob)
    else:
        rt.update(ut.STORE_KEY, _apply, flush="async")
    return written