"""The TENANT-WIDE overlay stratum — one value per (row, column) for the whole workspace.

Wave 29, item 20 / owner ruling R11, contract C5. The sibling of `core/table_store.py`, which
holds the PER-USER strata (`store.get(key)[username]`) and always has.

⛔ THE DEFECT THIS EXISTS FOR IS NOT "SHARING WOULD BE NICE" — IT IS THAT A SHARED VIEW SILENTLY
WIDENS. `modules/product_data.py` already states it, in the comment above the supplier master:

    "A user-created field and its values live in the PER-USER strata; only VIEWS are shared. So a
     shared 'Buy list' view that filtered on a user-created column would, for every OTHER account,
     name a column that does not exist — and an unknown column is an INACTIVE condition in the
     tri-state engine, which IGNORES it and therefore WIDENS. The buy list would silently show the
     whole catalogue to everyone but its author."

That is why the four supplier columns were frozen as read-only CONTRACT columns rather than made
editable, and why the owner's ask ("turn this Excel sheet into a User created Field that we can
edit") has been parked for two waves. A column whose value is the same for every reader makes the
shared view mean ONE thing, which is the precondition for editing it at all.

────────────────────────────────────────────────────────────────────────────────────────────────
⛔⛔ THIS MODULE IS NOT A PERMISSION WALL, AND MUST NEVER BECOME ONE BY ACCIDENT.

It refuses no reader and no writer. The caller has already answered "may this session open this
surface" (`user_tables.may_open`, `Session.require`, the BU pool) and this module answers only
"what is stored". Two different questions; one of them belongs upstream, where the session is.

What it DOES enforce is the one thing a caller can get wrong silently:

    ⭐ `cells(table_key, pids)` — `pids` IS REQUIRED, POSITIONALLY, AND THERE IS NO "EVERYTHING"
      CALL. A default of "all rows" is the widening hazard above wearing a friendly face: the
      caller passes the row set it has ALREADY scoped, so a cell for a row this reader may not see
      cannot come back to be merged. There is deliberately no `all_cells()` to reach for; a caller
      that genuinely needs the lot passes the lot, in writing, where a reviewer can see it.

The row wall then holds twice over, because `aios_grid.rows_from_pool` iterates the POOL and looks
each pid UP in the overlay — never the reverse. A cell for a row outside the pool has nothing to
attach to. This module is built to keep that true rather than to re-implement it.
────────────────────────────────────────────────────────────────────────────────────────────────

RESIDENCY. Its own bucket, `<table_key>__shared`, beside the per-user one — never a `__shared__`
member inside it. Two reasons, and the first is `product_data.py`'s own argument for separate
topic keys ("separate store keys make that structurally impossible, which is the whole reason the
table-page factory exists"): a per-user reader iterating usernames cannot encounter shared data
when there is no shared data in that bucket to encounter. The second is cost — the per-user bucket
carries every user's views, and a cell edit should not read-modify-write all of it.

SHAPE, chosen to be drop-in:

    {"fields": {field_key: Field},              # the tenant-wide column DEFINITIONS
     "cells":  {"<pid>": {field_key: value}}}   # exactly `rows_from_pool(overlays=...)`'s shape

`cells()` returns STRING pid keys for that reason — `rows_from_pool` does `overlays.get(str(pid))`,
so a caller merges the shared stratum over the per-user one with one `dict.update` and cannot get
the key type wrong. (`table_store` stores `{str(pid): {...}}` too; one shape, three readers.)
"""
import core.store as store

#: The suffix that turns a topic's per-user workspace key into its shared one. Callers pass the
#: key they ALREADY hold (`product_data.TABLE_KEY`, `f'{ut_key}_table_workspace'`) — one
#: identifier for both strata, so there is no second naming convention to get wrong.
BUCKET_SUFFIX = '__shared'

#: One cell value's ceiling, matching the `json` field kind's. A shared cell is read by every user
#: in the tenant, so an unbounded one is an unbounded cost for all of them.
MAX_VALUE_CHARS = 32_768


def bucket(table_key):
    """The store key this topic's shared stratum lives under."""
    key = str(table_key or '').strip()
    if not key:
        raise ValueError('shared_overlay: a table_key is required — it names the bucket')
    return f'{key}{BUCKET_SUFFIX}'


def _st(st):
    return st if st is not None else store


def _read(table_key, st=None):
    """The whole stratum, always in its full two-key shape. Lenient like every other display
    read: an unreachable store degrades to "nothing shared yet", never to an exception on a page
    that would otherwise render."""
    try:
        data = _st(st).get(bucket(table_key)) or {}
    except Exception:
        data = {}
    return {'fields': dict(data.get('fields') or {}),
            'cells': dict(data.get('cells') or {})}


def _write(table_key, change, st=None, flush='async'):
    """Read-modify-write one shared stratum.

    `flush='async'` by default for the same reason `table_store._update` uses it: a typed cell
    lands here inside the request round-trip, and the historical synchronous hub commit cost
    seconds per keystroke. Structural writes (a field definition) pass `flush='sync'` — they are
    rare, and a lost column definition is a worse failure than a lost keystroke.
    """
    def _up(data):
        data = data if isinstance(data, dict) else {}
        data.setdefault('fields', {})
        data.setdefault('cells', {})
        change(data)
        return data
    return _st(st).update(bucket(table_key), _up, flush=flush)


def _pid(pid):
    """Pids are ints everywhere in the grid and strings in JSON. ONE coercion, here, so a caller
    passing either cannot write a row that a reader keyed the other way never finds."""
    return str(int(pid))


def _value(value):
    """A cell holds a SCALAR. The Row contract is scalar on every surface — the grid, the filter
    engine, formulas, export — and the `json` kind is a validated STRING rather than an object.

    ⛔ A dict or a list RAISES rather than being dropped. This is called by our own code, so a
    non-scalar is a programming error, and silently storing nothing would surface later as "the
    shared column is blank for everyone" with no failure anywhere near the cause.
    """
    if isinstance(value, (dict, list, tuple, set)):
        raise ValueError(f'shared_overlay: a cell holds a scalar, not {type(value).__name__}')
    if value is None or isinstance(value, bool) or isinstance(value, (int, float)):
        return value
    text = str(value)
    return text[:MAX_VALUE_CHARS]


# ----------------------------------------------------------------- the column DEFINITIONS
def fields(table_key, st=None):
    """`{field_key: Field}` — the columns this topic shares tenant-wide.

    Unscoped on purpose, and it is the one thing here that is: a shared column's EXISTENCE is
    tenant-wide by definition — that is the whole feature, and it is what stops a shared view
    naming a column half the workspace lacks. Its VALUES are scoped by `cells()`.
    """
    return _read(table_key, st)['fields']


def is_shared(table_key, field_key, st=None):
    """Is this column's value tenant-wide? The predicate a caller uses to decide WHICH stratum to
    read and write — one evaluator, so the read path and the write path cannot disagree about
    where a given column lives ([[one-evaluator-per-question]])."""
    return str(field_key or '') in _read(table_key, st)['fields']


def put_field(table_key, field_key, defn, st=None):
    """Declare (or redefine) a shared column. Returns what was stored."""
    key = str(field_key or '').strip()
    if not key:
        raise ValueError('shared_overlay: a field needs a key')
    entry = dict(defn or {})
    entry['key'] = key

    def _add(data):
        data['fields'][key] = entry

    _write(table_key, _add, st, flush='sync')
    return entry


def drop_field(table_key, field_key, st=None):
    """Remove a shared column AND every stored value for it.

    ⛔ The cells go with the definition, exactly as `table_store.delete_field` scrubs its own:
    orphaned values would silently resurface if the key were ever reused — and a resurrected
    value in a TENANT-WIDE stratum reappears for everybody at once.
    """
    key = str(field_key or '').strip()
    if not key:
        return False
    hit = [False]

    def _drop(data):
        hit[0] = data['fields'].pop(key, None) is not None
        for row in data['cells'].values():
            if isinstance(row, dict):
                row.pop(key, None)
        # ⚠ AND DROP THE ROWS THAT ARE NOW EMPTY. `cells()` already skips a blank row, so this
        # changes nothing a reader sees — but this bucket is read by every user in the tenant,
        # and a `{}` per pid that once held a since-deleted column is dead weight that only ever
        # grows. Caught by `prune()` reporting rows nobody could see as dropped.
        for pid in [p for p, row in data['cells'].items() if not row]:
            data['cells'].pop(pid, None)

    _write(table_key, _drop, st, flush='sync')
    return hit[0]


# ----------------------------------------------------------------------------- the VALUES
def cells(table_key, pids, st=None):
    """`{"<pid>": {field_key: value}}` for the rows named by `pids` — and ONLY those.

    ⛔ `pids` IS REQUIRED. See the module header: this is the row wall expressed as a signature,
    so a caller cannot get "every shared cell in the tenant" by forgetting an argument. Pass the
    pool you have already scoped for this session.

    An empty `pids` legitimately returns `{}` — a reader with no rows sees no cells, which is the
    correct answer rather than a special case.
    """
    if pids is None:
        raise TypeError('shared_overlay.cells: pids is required — pass the row set you have '
                        'already scoped for this reader (there is deliberately no "all" call)')
    wanted = {_pid(p) for p in pids}
    if not wanted:
        return {}
    stored = _read(table_key, st)['cells']
    return {pid: dict(row) for pid, row in stored.items()
            if pid in wanted and isinstance(row, dict) and row}


def put_cell(table_key, pid, field_key, value, st=None):
    """Write ONE shared cell. Returns the value as stored (which may be truncated)."""
    return put_cells(table_key, pid, {field_key: value}, st=st).get(str(field_key or ''))


def put_cells(table_key, pid, values, st=None):
    """Write several shared cells on one row, in one store update. Returns what was stored."""
    row_id = _pid(pid)
    clean = {str(k): _value(v) for k, v in dict(values or {}).items() if str(k)}
    if not clean:
        return {}

    def _patch(data):
        data['cells'].setdefault(row_id, {}).update(clean)

    _write(table_key, _patch, st)
    return clean


def clear_row(table_key, pid, st=None):
    """Forget every shared cell on one row. True when there was something to forget."""
    row_id = _pid(pid)
    hit = [False]

    def _clear(data):
        hit[0] = data['cells'].pop(row_id, None) is not None

    _write(table_key, _clear, st)
    return hit[0]


def prune(table_key, live_pids, st=None):
    """Drop shared cells for rows that no longer exist. Returns how many rows were dropped.

    ⛔ NOT a scoped read wearing another name, and the difference is the whole reason `cells()`
    refuses to serve everything: `live_pids` here means *every row this TABLE has*, which is a
    fact about the data, whereas `pids` in `cells()` means *every row this READER may see*, which
    is a fact about the session. Calling this with one session's pool would delete the shared
    values of every row that session cannot see. The name says which is which; so does this note.
    """
    keep = {_pid(p) for p in live_pids}
    dropped = [0]

    def _prune(data):
        gone = [pid for pid in data['cells'] if pid not in keep]
        for pid in gone:
            data['cells'].pop(pid, None)
        dropped[0] = len(gone)

    _write(table_key, _prune, st, flush='sync')
    return dropped[0]