"""routes_changes.py — THE CHANGE TOKEN (wave 29, item 20 / ruling R11, contract C6).

    GET /api/v1/changes?scope=<scope>
      -> {"scope": "ut_leads",
          "tokens": {"rows": "b3f1c9:14", "overlay": "b3f1c9:3"},
          "backend": "hf"}

The owner's report: *a record created in another tab, by an automation, or by a connector sync
does not appear in a filtered view until you reload.* The client had no way to ask whether
anything had changed except to re-download everything, so it never asked.

⛔ WHY THIS ROUTE EXISTS INSTEAD OF THE OBVIOUS "JUST POLL THE ROWS ENDPOINT", and the arithmetic
is the whole argument. `GET /tables/{key}/rows` costs THREE full-tenant deep copies under ONE
lock, on ONE uvicorn process (the container runs no `--workers`), against a documented ceiling of
35.8 MB / ~1.4 s per bucket ⇒ roughly 4.2 s of lock-held CPU per poll, so TWO TABS SATURATE THE
SERVER and every unrelated write queues behind them. This handler performs ZERO `store.get()`
calls: `revision_of` below reads an in-memory counter that `Store.put`/`Store.update` bump at the
moment a write is accepted. `verify_live_workspace.py`'s server half counts
`get`/`_read_strict`/`_download` on this path and goes RED at one — because a token endpoint that
deep-copies pays the identical 1.4 s and wins nothing.

⭐ IT REACHES ALL THREE WRITERS THE OWNER NAMED, and that is a property of the QUESTION, not of
the plumbing: it asks *"did this bucket change"*, never *"did somebody emit an event"*. The event
seam (`user_tables.ROW_HOOKS`) fires only from HUMAN write doors by construction — the automation
engine's own writer deliberately stays silent, and `verify_automation.py` has a negative control
enforcing that silence. A push channel could therefore only ever carry 1 of the 3 writers. A
counter on the store sees the automation and the connector sync for free, and the loop-prevention
law is untouched.

WHAT WALLS THIS:

  * The TENANT is the wall, and it is structural. Every bucket name is namespaced through the
    session's OWN runtime (`TenantRuntime.store_key`), so a scope naming another tenant's table
    resolves to a bucket this tenant does not have and reads 0. There is no cross-tenant answer
    to give. **Measured, wave-29 QA:** a real `nurilab-admin` session against a `royal-imports`
    table gets no data here.
  * ⛔ …AND, SINCE W29-T82, A PER-TABLE WALL INSIDE THE TENANT. The paragraph that used to sit
    here argued this route needed none, on the ground that `user_tables.STORE_KEY` is ONE bucket
    for every `ut_*` table so "there is no per-table fact to withhold". **Half of that is false
    and the QA pass measured it.** The `rows` token is indeed shared — but `overlay` and `shared`
    are PER-TABLE buckets (`<scope>_table_workspace`), so a non-zero token against `:0` told a
    session whether a table it may not open EXISTS and is being edited. That is an
    existence/activity oracle, and every sibling door (`GET rows`, `POST rows`, `PATCH rows/{pid}`,
    `PATCH fields/{k}`, `POST rows/import`) answers **404 `unknown_table`** rather than leak it.
    One door out of six holding a different line is the door somebody probes.
  * ⚠ AND THE COST ARGUMENT WAS RIGHT, so the wall is CACHED rather than paid per poll.
    `user_tables.may_open` reads the whole `user_tables` bucket through `store.get`, which is a
    JSON round-trip deep copy under the store lock — the exact 1.4 s this route exists not to
    spend, and it would be spent by every tab on every interval. `_may_watch` below memoises the
    verdict per (tenant, user, scope) for `_WALL_TTL` seconds. The common case costs nothing: a
    tab only polls a scope it already has OPEN, which already passed the same wall on the
    workspace fetch. A prober pays one read per scope per TTL and gets 404s.

So: a valid session, a scope sanitised to a bucket-safe shape, and — for a `ut_*` table — a
cached verdict that this session may open it.
"""
import time
from fastapi import APIRouter, Depends

import core.shared_overlay as shared_overlay
import core.store as store
import core.user_tables as user_tables
import modules.customer_data as customer_data
import modules.product_data as product_data

from deps import Session, err, require_session

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

#: A scope reaches a STORE KEY, so it is sanitised as strictly as one: lowercase, digits and
#: underscores only. Not paranoia about SQL — `Store` addresses files and `PgStore` binds
#: parameters — but a scope is user-supplied text that becomes part of a path in the HF backend,
#: and "it cannot escape today" is a property of the current backend rather than of this route.
_SCOPE_MAX = 64


def revision_of(handle, name):
    """`{'rev','updated_at','token'}` for bucket `name` as THIS caller addresses it — or None.

    ⭐ ONE resolver, and the gate calls the HANDLER rather than reproducing this dance, so the
    route and its control cannot disagree about which physical key was read. A control that
    re-derives the address is how a gate ends up certifying a key nobody serves.

    `handle` is whatever the caller holds — a tenant runtime, a `Store`, or `core.store` itself.
    The runtime is recognised DUCK-TYPED (`store_key` + `store_handle`) rather than imported:
    `store_key()` is the ONE place a tenant's namespace prefix is applied, and applying it here a
    second time is how one tenant reads another's counter.

    ⛔ None — never a fabricated token — when the active backend publishes no revision. `PgStore`
    maintains `rev` in `store_kv` and does not yet expose it, so under `STORE_BACKEND=pg` this
    answers None, the client stops polling, and the app behaves exactly as it does today. A token
    frozen at 0 would instead promise liveness the backend is not delivering.
    """
    target, key = handle, str(name)
    namespaced = getattr(handle, "store_key", None)
    if callable(namespaced):                     # a tenant runtime owns the namespace prefix
        key = namespaced(name)
        bound = getattr(handle, "store_handle", None)
        target = bound if bound is not None else store._d()
    fn = getattr(target, "revision", None)
    if not callable(fn):
        return None
    try:
        return fn(key)
    except Exception:
        # A backend that raises here must not take the poll down with it: the caller degrades to
        # "no token for this bucket", which is the same honest null the pg path returns.
        return None


def _clean_scope(raw):
    s = str(raw or "").strip().lower()
    if not s or len(s) > _SCOPE_MAX or any(c not in "abcdefghijklmnopqrstuvwxyz0123456789_"
                                           for c in s):
        return ""
    return s


def buckets_for(scope):
    """`{label: store bucket}` for one topic scope — the map the CLIENT never gets to write.

    ⛔ Every value is imported from the module that WRITES it, never re-spelled here.
    `user_tables.STORE_KEY`, `customer_data.TABLE_KEY` and `product_data.TABLE_KEY` are the
    literal keys those modules read and update, so a rename moves this map with it. A hand-copied
    `'customer_table_workspace'` would keep answering 200 with a token for a bucket nobody writes
    — a poll that costs nothing, changes never, and looks exactly like "there were no changes".

    An absent label means "no store residency": `customer` and `product` rows come from Odoo
    through a 15-minute server cache, which R11 deliberately leaves alone. The client reads a
    missing/null token as "do not poll this", never as "unchanged".
    """
    s = _clean_scope(scope)
    if not s:
        return None
    #: The shared stratum's bucket is asked FOR rather than spelled, so C5's `BUCKET_SUFFIX` has
    #: exactly one definition and a rename cannot leave the poller watching a dead key.
    if s.startswith(user_tables.KEY_PREFIX):
        workspace = f"{s}_table_workspace"
    elif s == "customer":
        workspace = customer_data.TABLE_KEY
    elif s == "product":
        workspace = product_data.TABLE_KEY
    else:
        return {}
    names = {"overlay": workspace, "shared": shared_overlay.bucket(workspace)}
    if s.startswith(user_tables.KEY_PREFIX):
        names["rows"] = user_tables.STORE_KEY
    return names


#: How long a may-open verdict is trusted. Short enough that a REVOKED grant stops answering
#: within a poll interval or two; long enough that a tab polling every few seconds pays the
#: bucket read at most once per TTL. ⚠ A REVOCATION is therefore visible here up to `_WALL_TTL`
#: late — stated rather than glossed, because the alternative (no cache) is the 1.4 s-per-poll
#: cost this route was built to avoid, and the leak this bounds is an existence oracle, not data.
_WALL_TTL = 30.0
#: (tenant, user, scope) -> (expires_at, verdict). Bounded: a prober walking thousands of scopes
#: must not be able to grow this without limit, so it is cleared wholesale when it gets large —
#: cheaper than an LRU and the only cost of being wrong is one extra bucket read.
_WALL_CACHE = {}
_WALL_MAX = 512


def _may_watch(session, scope):
    """May this session watch this scope? Cached, because the honest check is expensive.

    ⛔ FAIL-CLOSED THROUGH THE ONE RESOLVER. `user_tables.may_open` is where "who may open this
    database" is decided (creator · admin · an explicit share grant); asking a second way here is
    how two doors end up with two answers ([[one-evaluator-per-question]]). Non-`ut_*` scopes
    (`customer`, `product`) are module-gated surfaces whose own doors decide access and whose
    buckets are tenant-wide, so there is no per-table fact to withhold for them.
    """
    if not scope.startswith(user_tables.KEY_PREFIX):
        return True
    # ⚠ THE NAMESPACE STRING, not `runtime.tenant` — that attribute is a `Tenant` OBJECT and is
    # unhashable, so keying on it raised a TypeError INSIDE the handler, which `TestClient`
    # re-raises: the gate CRASHED rather than reddening ([[gate-must-go-red-not-crash]], caught
    # by the very leg added for this ticket). `store_namespace` is the string that already makes
    # one tenant's buckets distinct from another's, so it is the right axis anyway.
    key = (str(getattr(session.runtime, "store_namespace", "")), session.uname, scope)
    now = time.monotonic()
    hit = _WALL_CACHE.get(key)
    if hit and hit[0] > now:
        return hit[1]
    verdict = bool(user_tables.may_open(scope, session.uname, session.admin, st=session.runtime))
    # ⛔ ONLY A `True` IS CACHED, and the asymmetry is deliberate. `user_tables.all_tables()`
    # SWALLOWS a store error and answers `{}`, which makes `may_open` False — so caching a
    # negative would turn one transient store hiccup into "no database <name> here" for the whole
    # TTL, on a poller, for a table the user has open. A false negative that expires in 30 s is a
    # worse failure than one extra bucket read: the read costs latency, the cache costs trust
    # ([[lost-write-looks-like-failed-read]] — a refusal derived from a swallowed error reads to
    # the user as their data being gone).
    if verdict:
        if len(_WALL_CACHE) >= _WALL_MAX:
            _WALL_CACHE.clear()
        _WALL_CACHE[key] = (now + _WALL_TTL, True)
    return verdict


@router.get("/changes")
def changes(scope: str = "", session: Session = Depends(require_session)):
    """One tiny read per tab per interval. The response is small, uncacheable and boring."""
    names = buckets_for(scope)
    if names is None:
        raise err(400, "bad_scope", "name the table you are watching, e.g. ?scope=customer")
    # ⛔ W29-T82 — THE SAME REFUSAL EVERY SIBLING DOOR GIVES, word for word. A session that may
    # not open this table must not be able to tell whether it exists, and "404 unknown_table" is
    # the sentence the other five row doors answer with. A different code here would itself be
    # the tell.
    if names and not _may_watch(session, _clean_scope(scope)):
        raise err(404, "unknown_table", f"no database {_clean_scope(scope)!r} here")
    if not names:
        # ⛔ A well-formed scope this map does not know is a 404, NOT an empty `tokens` object.
        # An empty map is indistinguishable from "nothing here ever changes", so a topic added
        # later would poll forever against silence and read as a working live view that never
        # updates. Loud beats plausible.
        raise err(404, "unknown_scope", f"nothing here watches {_clean_scope(scope)!r}")
    tokens = {}
    for label, bucket in names.items():
        rev = revision_of(session.runtime, bucket)
        # ⛔ A NULL IS AN ANSWER, not an omission. The pg backend maintains `rev` in `store_kv`
        # and does not yet expose it, so after D-4 flips this reads None until `PgStore` grows a
        # `revision()`. Sending null and letting the client stop polling is the honest failure;
        # a token frozen at 0 would promise liveness the backend is not delivering.
        tokens[label] = rev.get("token") if isinstance(rev, dict) else None
    return {"scope": _clean_scope(scope), "tokens": tokens, "backend": store.backend()}