"""routes_products.py — the PRODUCT table's read seam (wave 15 item 9/10, contract C-TOPIC).

The proof that C-TOPIC's bet is real: **a second table object is a pool module plus a route, not
a second component tree**. Everything structural here is borrowed rather than re-implemented —
the permission wall (`core.perm_scope`), the row builder (`aios_grid.rows_from_pool`), the
scope-keyed stale-while-refresh cache (`scope_cache`), the error shape, the module gate.

Landed nav-less in wave 15 ON PURPOSE (no registry row, so no user could navigate to a
half-built surface while the server half hardened); wave 16 shipped the `product_data`
registry row, the client `topic` prop and the write path, so the surface has been REACHABLE
from the nav since — this header stayed stale until 2026-08-04 (the wave-18 debt sweep) and
cost readers a false "you can't get there from here".

Wave 16 (C-TOPIC's second half): the WRITE PATH exists now, and it is exactly what the wave-15
header demanded before one could — its OWN table workspace bucket
(`modules.product_data.TABLE_OPS`, store key 'product_table_workspace') and its own event
validation (the events route builds the ctx over the PRODUCT field contract + product pids +
the product ops). ⛔ The separate bucket is load-bearing, not tidy: product pids are CRC32
hashes and customer pids are Odoo partner ids — one shared overlay bucket and a hash collision
silently writes a product note onto somebody's customer.
"""
import time

from fastapi import Body, Depends

from fastapi import APIRouter

import scope_cache
from deps import Session, err, module_gate

router = APIRouter(prefix="/api/v1")

#: The registry key this surface WILL carry. The gate is live now even though the nav is not, so
#: the day the registry row appears the wall is already the one that was tested.
MODULE = "product_data"

_CACHE_TTL = 900


def _pool_for(rt, team_id):
    """The cached SKU pool for one scope. Same stale-while-refresh discipline as the customer
    pool: only a scope's FIRST-ever build blocks, and the cache key is the SCOPE, never the user
    — the customer route learned that the hard way (a per-user payload cached on a scope key
    served one user's private columns to another)."""
    import modules.product_data as pd

    # ⚠ Signature is (cache, key, ttl, build) — the wave-15 version passed `ttl=` as a keyword
    # after the build lambda and DIED on every call ("multiple values for 'ttl'"). Latent until
    # wave 16 because nothing drove this route end-to-end: verify_perm_scope's section H stubs
    # the pool a layer below. verify_api's W16 section now exercises the real call.
    key = ("product_pool", team_id)

    # ⭐ D-10 (wave 24) — THE PRODUCT POOL HONOURS THE CONNECTOR PAUSE. It did not, and that was
    # a shipped defect on a live surface: the CUSTOMER pool got this guard at DEBT-2 and its
    # sibling — written from the same template, four files away — never did. So pausing Odoo
    # froze the Customer grid and the Product grid kept reading live, with the Settings toggle
    # reporting success. Sibling caches diverge exactly this way and nothing greps for
    # "the other one".
    #
    # ⚠ NO SNAPSHOT LEG, and that is the honest difference from the customer path rather than an
    # omission: `save_pool_snapshots` persists the `odoo_pool` scopes only — there is no product
    # snapshot bucket to serve. So this serves the in-process copy AT ANY AGE, and answers 503
    # naming the pause when there is none. Minting a product snapshot here would be a new
    # persistence format at a wave tail; refusing with a sentence is the answer the user can act
    # on, and it is the same 503 the customer path already gives for an unsnapshotted scope.
    # ⭐ WAVE 29 / R12 — THE POOL IS THE CATALOGUE NOW (5,875 active products, not the 2,717 that
    # happened to sell in a revenue window). Two consequences that land HERE rather than in
    # `product_data`:
    #   · the 503 below now also covers a TRUNCATED catalogue read. `modules.product_data.
    #     _catalogue_by_code` RAISES instead of degrading, precisely so a short pull arrives as a
    #     refusal a user can act on rather than as a plausible smaller grid nobody questions;
    #     `pool_unavailable` names the cause in its message.
    #   · the first build for a scope got slower (MEASURED 2026-08-11: ~44 s consolidated, of
    #     which the catalogue read is ~3 s; a BU-scoped build is ~9 s). Only a scope's FIRST-ever
    #     build blocks — the stale-while-refresh discipline above is what keeps that off the
    #     request path afterwards.
    import routes_keychain
    if routes_keychain.odoo_paused(rt):
        hit = rt.pool_cache.get(key)
        if hit:
            return hit[1]
        raise err(503, "connector_paused",
                  "this data source is paused and nothing has been read since — "
                  "an admin can resume it under Settings → Connectors")

    return scope_cache.get(rt.pool_cache, key, _CACHE_TTL,
                           lambda: pd.pool(team_id=team_id))


def scoped_pool(session: Session):
    """`(pids, team_id, rows_src, fields_base)` — THE PRODUCT WALL, on its own.

    Extracted from `product_assembly` (wave 19, item 12) so a caller that needs only "which
    products may this session touch" — record comments, say — asks the SAME question in the same
    order rather than re-deriving it: pool scope from the permanent filter, rows, THEN the row
    wall, THEN the pids. Re-deriving it is how a second wall drifts from the first, and the
    walls are the whole point of this route.

    ⚠ The caller is responsible for the GRANT (`session.require(MODULE)`); this is the row half.
    """
    import core.perm_scope as perm_scope

    team_id, _agent = perm_scope.derive_pool_scope(session.user, MODULE)
    try:
        rows_src = _pool_for(session.runtime, team_id)
    except Exception as e:
        raise err(503, "pool_unavailable",
                  f"the product catalogue could not be built — {str(e)[:160]}")
    fields_base = pd_fields(consolidated=team_id is None)
    # The SAME wall the customer assembly applies, in the same order: rows first (before pids
    # are taken, so an out-of-filter row never enters allowed_pids), then the field closure,
    # then the values stripped from the rows as well as the field list.
    rows_src = perm_scope.apply_row_scope(rows_src, session.user, MODULE, fields_base)
    pids = frozenset(r["pid"] for r in rows_src if r.get("pid") is not None)
    return pids, team_id, rows_src, fields_base


def product_assembly(session: Session, scope: str = "product", storage_key: str = "",
                     consume_corrections: bool = True):
    """The product topic's mirror of `routes_customers.grid_assembly` — SAME g-dict keys, so
    the /workspace and events routes consume either interchangeably.

    One deliberate absence, a topic fact rather than a gap:
      * `measures`/`measure_sets` are EMPTY — `core.measure_resolve` is CUSTOMER-grain (the
        C-TOPIC v1 descope, booked in the wave doc); the events ctx therefore refuses measure
        creates on this surface, which is the correct fail-closed shape.

    ⭐ WAVE 19 / R9 — `lists` IS NO LONGER EMPTY. Wave 16 passed `with_cohorts=False` because
    there was one customer-keyed cohort bucket and product pids are CRC32 hashes of SKU codes;
    intersecting the two id spaces would have printed a plausible, meaningless member count. The
    owner's ruling is that a cohort belongs to its database, so `modules.cohort` grew a bucket
    per topic and the product surface reads `product_cohorts` — ids from this pool, resolved
    against this pool. `derived` carries their membership cells (the Cohorts column) for the same
    reason it does on the customer surface; the measure half of that channel stays empty.
    """
    import aios_grid
    import core.perm_scope as perm_scope
    import modules.product_data as pd
    from core import grid_events

    pids, team_id, rows_src, fields_base = scoped_pool(session)

    ctx = grid_events.EventCtx(
        uname=session.uname, allowed_pids=pids, fields=[],
        hidden_keys=perm_scope.hidden_keys(session.user, MODULE, fields_base),
        admin=session.admin, fallback_ws=None, seen_ids={},
        scope_key="product", table=pd.TABLE_OPS)
    ws = grid_events.table_workspace(ctx, allowed_pids=pids,
                                     consume_corrections=consume_corrections)
    # ⭐⭐ W30-T36 — THE TENANT-WIDE STRATUM, MERGED OVER THE PER-USER ONE, IN THE ASSEMBLY SO
    # EVERY CONSUMER SEES ONE TRUTH. `routes_grid`'s /workspace route serves
    # `workspace["overlays"] = g["ws"].get("overlays")` verbatim, so doing it here means that door
    # carries shared values with no change in a file this lane does not own.
    #
    # ⚠ SAFE TO MUTATE, and that was checked rather than assumed: `table_workspace` reads through
    # `store.get`, which deep-copies, so `ws` is a detached copy and nothing writes it back. A
    # shared value can therefore never leak INTO the per-user bucket by way of this merge.
    # ⚠ SHARED WINS per key — these are contract columns, so the tenant-wide value is the answer
    # and any per-user leftover under the same key is stale by construction.
    _shared = pd.shared_cells(pids)
    if _shared:
        _ov = dict(ws.get("overlays") or {})
        for _pid, _cells in _shared.items():
            _ov[_pid] = {**(_ov.get(_pid) or {}), **_cells}
        ws["overlays"] = _ov
    workspace, fields, views, lists = aios_grid.workspace_wire(
        ws, session.uname, set(pids), defs={}, scope_key=scope, storage_key=storage_key,
        fields_base=fields_base)

    hidden = perm_scope.hidden_keys(session.user, MODULE, fields)
    if hidden:
        fields = [f for f in fields if f.get("key") not in hidden]
        rows_src = [perm_scope.strip_row(r, hidden) for r in rows_src]

    return {"rows_src": rows_src, "pids": pids, "ws": ws, "workspace": workspace,
            "fields": fields, "views": views, "lists": lists,
            # R9: the Cohorts column's cells, built from THIS topic's lists. Same read-only
            # `derived` channel the customer assembly uses — the measure half stays empty.
            "derived": aios_grid.cohort_cells(lists),
            "measures": [], "measure_sets": {}, "today": time.strftime("%Y-%m-%d"),
            "team_id": team_id}


@router.get("/products")
def products(session: Session = Depends(module_gate(MODULE))):
    """The product table for this session's scope — the /customers envelope byte-for-byte
    (`{fields, rows, today, pulled_at}`) plus two additive keys (`identity`, `scope`).

    The BU scope is DERIVED FROM THE PERMANENT FILTER (`perm_scope.derive_pool_scope`), exactly
    as the customer route derives it — and for the same reason, which is worth restating because
    it is the wave's central lesson: `team_id` shapes the revenue VALUES on each row, so a BU
    enforced as a post-filter yields a correct row list carrying both units' numbers.
    """
    import aios_grid

    g = product_assembly(session)
    rows = aios_grid.rows_from_pool(
        g["rows_src"], g["fields"], g["ws"].get("overlays"), derived=g["derived"])
    rows = _seed_image(_seed_shared(rows, g["rows_src"], g["fields"]), g["fields"])
    return {"fields": g["fields"], "rows": rows,
            "today": g["today"], "pulled_at": time.strftime("%Y-%m-%d %H:%M"),
            "identity": {"pid": "pid", "businessKey": "code"},
            "scope": {"team_id": g["team_id"], "consolidated": g["team_id"] is None}}


@router.patch("/products/{pid}")
def patch_product(pid: int, body: dict = Body(default=None),
                  session: Session = Depends(module_gate(MODULE))):
    """Write the product table's EDITABLE overlay stratum — the customers PATCH, on the product
    topic's ctx. Routed through `core.grid_events.handle_one` so the per-key `permissions.edit`
    wall, the pid wall and the truncation rules stay ONE implementation; the ctx's `table` ops
    aim the write at the PRODUCT bucket."""
    import modules.product_data as pd
    from core import grid_events

    updates = dict(body or {})
    if not updates:
        raise err(400, "empty_patch", "no fields to update")
    g = product_assembly(session, consume_corrections=False)
    if pid not in g["pids"]:
        # 403, not 404 — the code may exist; it is simply not in this session's catalogue.
        raise err(403, "out_of_scope", "that product is not in your catalogue")
    import core.perm_scope as perm_scope

    ctx = grid_events.EventCtx(
        uname=session.uname, allowed_pids=g["pids"], fields=g["fields"],
        admin=session.admin, fallback_ws=None, seen_ids={},
        hidden_keys=perm_scope.hidden_keys(
            session.user, MODULE, pd_fields(consolidated=g["team_id"] is None)),
        scope_key="product", table=pd.TABLE_OPS)
    try:
        grid_events.handle_one(
            {"id": f"patch:product:{pid}:{time.time_ns()}", "type": "overlay_patch",
             "pid": pid, "updates": updates}, ctx)
    except grid_events.StoreUnavailable:
        raise err(503, "store_unavailable",
                  "the tenant store is unavailable — your change was not saved")

    # What actually landed, read back from the PRODUCT bucket rather than echoed from the
    # request: a refused key or a truncated value must not be reported as accepted.
    # ⚠ W30-T36 — AND FROM THE SHARED BUCKET TOO. The four shared columns are written to
    # `<TABLE_KEY>__shared` by `_ProductTableStore.patch_overlay`, so a read-back that consulted
    # only the per-user stratum would find nothing and report every accepted shared write as
    # `refused` — a correct write reported as a failure, which is the read-path-cannot-witness-
    # the-write-path shape ([[read-path-cannot-witness-write-path]]).
    stored = (grid_events.table_workspace(ctx, allowed_pids=None)
              .get("overlays") or {}).get(str(pid)) or {}
    stored = {**stored, **((pd.shared_cells([pid]).get(str(pid))) or {})}
    accepted = {k: stored.get(k) for k in updates if k in stored}
    refused = sorted(k for k in updates
                     if k not in accepted or stored.get(k) != str(updates[k]))
    out = {"pid": pid, "updates": accepted}
    if refused:
        out["refused"] = refused
    return out


#: ⭐ WAVE 19 R7 — the RI Product table's built-in picture column. `source: 'overlay'` (editable,
#: so a user can upload a different shot for one SKU) with its DEFAULT supplied per render from
#: the row's `code` — see `_seed_image`. A stamped overlay value would have been the other way to
#: do it and it is the wrong one twice over: it is a 1,142-row migration that has to be right
#: once, and it FREEZES the code into the cell, so a re-coded SKU would keep pointing at the old
#: master with nothing saying why.
PRODUCT_IMAGE_KEY = "image"


def _seed_shared(rows, rows_src, fields):
    """W30-T36 — the supplier master as a per-render DEFAULT for the four SHARED columns.

    Exactly `_seed_image`'s shape and for the same reasons: no migration, and nothing frozen. An
    untouched grid renders byte-identically to the read-only version it replaces — including
    `supplier: "(none)"`, which the pool has always stamped — and **empty means "no override"**,
    so clearing a cell restores the mastersheet value rather than blanking the column.

    ⛔⛔ GATED PER KEY ON THE SERVED FIELD LIST, AND HERE THAT IS A PERMISSION WALL RATHER THAN
    TIDINESS. `_seed_image`'s docstring predicted this failure in writing — *"An unconditional seed
    would put the key straight back onto every row AFTER that strip… the value here is the visible
    `code`, so nothing new escapes today; the contract is the point, and a future non-code default
    would escape."* **`first_cost` IS that future non-code default**: a money number, and
    `product_assembly` strips a hidden field from BOTH wires precisely so a restricted reader
    cannot read it off the row payload. An ungated seed would re-attach it after the strip.
    (Belt and braces, deliberately: `perm_scope.strip_row` has already removed the key from
    `rows_src` too, so the pool row cannot supply it either.)

    ⚠ JOINED ON `pid`, never on position: `rows_from_pool` iterates the pool, but nothing in the
    contract promises the two lists stay index-aligned, and an off-by-one here would put one SKU's
    supplier on another SKU's row — a wrong answer that looks entirely plausible.
    """
    import modules.product_data as pd

    served = {f.get("key") for f in (fields or ())}
    keys = [k for k in pd.SHARED_KEYS() if k in served]
    if not keys:
        return rows
    by_pid = {r.get("pid"): r for r in (rows_src or ())}
    for row in rows:
        src = by_pid.get(row.get("pid")) or {}
        for key in keys:
            # A non-empty stored value WINS; only an empty cell is seeded. Blank is therefore
            # "no override" rather than "blank", which is the documented meaning `_seed_image`
            # already gives this column family and the only one that keeps the master reachable
            # after a mistaken edit.
            if row.get(key) in (None, ""):
                row[key] = src.get(key)
    return rows


def _seed_image(rows, fields):
    """R7's "auto-seeded from SKU `code`", as a per-render DEFAULT rather than stored data.

    Royal's 1,142 masters are named for `default_code`, so an untouched product row already names
    its own picture; this is what makes them appear with nothing uploaded. A user's own value is
    NON-EMPTY and therefore wins — the fallback only fills a cell nobody has set.

    ⚠ Clearing the cell restores the SKU's own picture rather than blanking it, and that is the
    documented meaning of empty on this column ("no override"). A product with no master on file
    still shows an empty frame, because the reference resolves to a 404 — the honest outcome, and
    the record modal names the failing reference in words.

    ⛔ GATED ON THE SERVED FIELD LIST, and that is C-PERM, not tidiness. `product_assembly` strips
    a hidden field from BOTH wires — the field list and the row payload — because narrowing only
    the first leaves the value sitting in the second where anything can read it. An unconditional
    seed would put the key straight back onto every row AFTER that strip, re-creating exactly the
    shape the rule forbids. (The value here is the visible `code`, so nothing new escapes today;
    the contract is the point, and a future non-code default would escape.)

    ⚠ STATED CONSEQUENCE: hiding `code` blanks this column, because the reference IS the code.
    That coupling is inherent to seeding from a business key, not a bug — and it fails in the safe
    direction (an empty frame, never another row's picture).
    """
    if not any(f.get("key") == PRODUCT_IMAGE_KEY for f in fields or ()):
        return rows
    for row in rows:
        if not row.get(PRODUCT_IMAGE_KEY):
            row[PRODUCT_IMAGE_KEY] = row.get("code") or ""
    return rows


def pd_fields(consolidated=True):
    """The product field contract, minus `product_data.CONSOLIDATED_ONLY` when the caller is scoped.

    ⭐ OWNER RULING 2026-08-11 EMPTIED THAT TUPLE, so today both callers receive the SAME columns
    and this narrowing is a no-op that is kept, not deleted. Previously a BU-scoped caller lost the
    whole inventory block, on the grounds that company-wide stock must not sit beside BU-shaped
    revenue. The owner's answer — *"just scope any inventory with Sales from Fisch, leave the
    rest"* — splits the block instead: the physical columns stay unscoped (there is one warehouse),
    the sales-derived ones are recomputed from the unit's own LTM units. The scope difference now
    lives entirely in the VALUES, which is where a reader can see it.

    The filter stays because the mechanism is still the right one for the next column that
    genuinely cannot be BU-shaped; `verify_perm_scope` asserts the tuple is empty so nothing here
    is silently filtering on a list that grew back.
    """
    import json
    from pathlib import Path

    import aios_grid

    doc = json.loads((Path(aios_grid.__file__).resolve().parent /
                      "aios_grid_fields.json").read_text(encoding="utf-8"))
    fields = list((doc.get("product_data") or {}).get("fields") or [])
    # R7: the Image column is BUILT IN on this table, injected here rather than added to
    # `aios_grid_fields.json`. That file is the ODOO-SOURCED contract — every key in it is a
    # column `modules.product_data.pool()` reads off a SKU — and this one is neither read from
    # Odoo nor written to it. Injected after the JSON so the canonical file stays the answer to
    # "what does Odoo give us", which is the question `verify_fields_contract` referees.
    fields = fields + [{
        "key": PRODUCT_IMAGE_KEY, "label": "Image", "type": "image", "source": "overlay",
        "note": "The product's picture. Empty shows the SKU's own master image; upload one "
                "from the record panel to override it.",
    }]
    if consolidated:
        return fields
    import modules.product_data as pd
    return [f for f in fields if f.get("key") not in pd.CONSOLIDATED_ONLY]