platform/core/table_store.py

"""The generic per-user TABLE WORKSPACE store — the persistence half of the table-page factory.

One durable store key holds one table OBJECT's per-user Airtable-style state:

    {username: {'views':    {view_id: SavedView},
                'fields':   {field_key: Field},        # notes + custom_ + measure_ strata
                'overlays': {str(pid): {field_key: value}}}}

`make(table_key)` returns the six operations a table page's host loop needs, closed over that
key. The Customer table's ops (`modules/customer_data.py`, key 'customer_table_workspace') are
these exact functions — the logic MOVED here 2026-07-27 so that duplicating the Customer table
pattern to a new object is a registry row + a config, not a copy of the store plumbing
(owner directive: the table-page factory).

A LIST (membership/formula semantics) is deliberately a different store from a VIEW
(presentation/query state) — see modules/customer_data.py's customer_lists key.
"""
import core.store as store

#: Wave-9 I17 — the SHARED bucket. Views whose permissions make them visible to anyone but
#: their creator live here instead of in a personal workspace, under a key that cannot collide
#: with a username (usernames come from core/users.py and are never dunder-wrapped; `is_shared`
#: guards it anyway). ONE HOME PER VIEW, never both: a view moved to personal is REMOVED from
#: here, and a view shared is removed from its creator's workspace. Two homes would mean two
#: divergent copies the moment either was edited.
SHARED_KEY = '__shared__'


def _may_see(view, viewer, is_admin=False):
    """Visibility for ONE shared view, fail-closed.

    'collaborative' = everyone who can already open the module (the caller has gated that).
    'users'         = the named users, plus the creator, plus admins — an admin who could not
                      see a view could not administer it either.
    Anything unrecognised returns False rather than defaulting open: an unreadable permission
    must never widen access ([[aios-permissioning]] — no fail-open defaults).
    """
    if not isinstance(view, dict):
        return False
    if view.get('createdBy') == viewer or is_admin:
        return True
    perms = view.get('permissions') or {}
    edit = perms.get('edit')
    if edit == 'collaborative':
        return True
    if edit == 'users':
        return viewer in set(perms.get('users') or ())
    return False                      # 'personal', absent, or junk


def _may_edit(view, viewer, is_admin=False):
    """Who may WRITE a shared view. Same set as visibility today — the owner's item asks 'who
    can edit' and lists who 'can have access', i.e. seeing and editing are one grant. Kept as a
    separate function so they can diverge (a future read-only share) without hunting callers."""
    return _may_see(view, viewer, is_admin)


def _may_administer(view, viewer, is_admin=False):
    """Who may change a view's PERMISSIONS, or delete it: the creator or an admin ONLY.

    Deliberately narrower than _may_edit. If a collaborator could rewrite `permissions` they
    could grant themselves sole ownership of somebody else's view, or quietly widen a
    users-scoped view to everyone — the classic privilege-escalation-by-edit hole.
    """
    if not isinstance(view, dict):
        return False
    return bool(is_admin) or view.get('createdBy') == viewer


def is_shared(view):
    """A view belongs in the shared bucket when its permissions reach beyond its creator."""
    return ((view or {}).get('permissions') or {}).get('edit') in ('collaborative', 'users')


def source_override_is_empty(payload):
    """Does this stored definition of a SOURCE (non-custom) column carry any user state?

    A cleared note on an immutable source field returns to the canonical schema instead of
    leaving a meaningless override row. Custom fields remain even with an empty note — and so
    does a PRESET field carrying a measure-window override (wave-2 item 8), a DISPLAY-format
    override (wave-5 item 10), and, since W29-T83, a COLUMN SUMMARY.

    ⛔ EVERY CLAUSE IS A SETTING A USER MADE, and each one omitted is a setting that silently
    stops surviving a session. `agg` was missing: choosing Average on Customer's `Overdue days`
    built an override whose only content was that summary, so this rule threw the whole row away
    on write while the menu went on reading "Summary: Average" from the client's own optimistic
    copy until the next login — a discarded WRITE wearing the face of a failed read
    ([[lost-write-looks-like-failed-read]]). Measured on `bac40c2`; a `ut_*` table, which stores
    its definitions through another door entirely, kept it.

    ⚠ ONE RULE, TWO CALLERS — here and `grid_events`' store-less fallback. Two copies of a
    discard rule is how one of them keeps a write the other bins ([[one-evaluator-per-question]]).
    ⚠ A CLEARED summary still drops the row, which is the intent: with nothing else set, the
    column goes back to whatever the contract declares for it.
    """
    payload = payload or {}
    if payload.get('custom'):
        return False
    return (not str(payload.get('note') or '').strip()
            and not isinstance(payload.get('measure'), dict)
            and not isinstance(payload.get('format'), dict)
            and not str(payload.get('agg') or '').strip())


def _unique_name(wanted, existing, *, fallback='Untitled', max_len=120):
    """Allocate one human-facing name inside a store.update transaction.

    Keys/ids remain structural identity.  Names compare case-insensitively after collapsing
    whitespace, because those variants are indistinguishable in the UI.  This helper belongs
    in the store layer: allocating from a pre-write snapshot lets two concurrent requests both
    choose the same free name before either write lands.
    """
    limit = max(1, int(max_len))

    def _clean(value):
        return ' '.join(str(value or '').split())

    base = (_clean(wanted) or _clean(fallback) or 'Untitled')[:limit].rstrip()
    taken = {_clean(value).casefold() for value in existing if _clean(value)}
    if base.casefold() not in taken:
        return base
    index = 2
    while True:
        suffix = f' {index}'
        stem = base[:max(0, limit - len(suffix))].rstrip()
        candidate = f'{stem}{suffix}' if stem else str(index)[-limit:]
        if candidate.casefold() not in taken:
            return candidate
        index += 1


class TableStore:
    """The six store operations for one table object's workspace, closed over its store key.

    `st` (wave 18, C3-UT) is the STORE HANDLE — anything exposing `get(name)` /
    `update(name, fn, flush=)`. Default = `core.store` (tenant #0, every existing caller,
    zero behaviour change). The API passes the session's `TenantRuntime`, whose accessors
    apply the tenant prefix / repo binding — which is what makes a user table created by a
    Nurilab admin land in Nurilab's store instead of Royal's.
    """

    def __init__(self, table_key, st=None):
        self.table_key = table_key
        self._st = st if st is not None else store

    @property
    def st(self):
        """The bound store handle — for SIBLING registries (core/shares) that must read the
        same tenant's buckets this workspace lives in (wave 21, C1)."""
        return self._st

    def find_view(self, view_id):
        """`(owner_username, view)` for a view living in ANY personal stratum, else None.

        ⭐ Wave 21 (item 9, C1): the R10 grant registry names bare ids, so projecting a granted
        view means locating the OWNER's record inside this topic's bucket. Personal strata
        only — the `__shared__` bucket has its own read path (`shared_views`), and serving one
        view from two finders is how two copies drift."""
        vid = str(view_id or '').strip()
        if not vid:
            return None
        try:
            data = self._st.get(self.table_key) or {}
        except Exception:
            return None
        for username, ws in data.items():
            if username == SHARED_KEY or not isinstance(ws, dict):
                continue
            v = (ws.get('views') or {}).get(vid)
            if isinstance(v, dict):
                return str(username), dict(v)
        return None

    def find_folder(self, folder_id):
        """`(owner_username, folder_row, {view_id: view})` for a VIEWS folder living in any
        personal stratum, else None. `find_view`'s sibling, and here for the same reason.

        ⭐ D-37 (wave 20's R10 remainder, closed 2026-08-05): the grant registry accepts kind
        `folder` and has since wave 20, but only the VIEW kind was ever projected — so "share
        this folder with Karen" recorded a row, listed under Shared with me, and put nothing on
        Karen's screen. Projecting a folder means two lookups the view path does not need: WHO
        owns it, and WHICH views are filed in it. Folder membership lives in the owner's
        `itemFolders` map (item id -> folder id), never on the view record, so the views are
        found by asking that map rather than by reading a list off the folder.

        Views only (`folders['views']`): the cohort surface has its own store and its own
        sharing question, and answering both here would make one function mean two things.
        """
        fid = str(folder_id or '').strip()
        if not fid:
            return None
        try:
            data = self._st.get(self.table_key) or {}
        except Exception:
            return None
        for username, ws in data.items():
            if username == SHARED_KEY or not isinstance(ws, dict):
                continue
            rows = (ws.get('folders') or {}).get('views') or []
            hit = next((f for f in rows
                        if isinstance(f, dict) and str(f.get('id') or '') == fid), None)
            if not hit:
                continue
            # ⚠ `itemFolders` IS KEYED BY SURFACE FIRST (`{'views': {itemId: folderId}, …}`) —
            # reading item ids off the top level finds the surface names instead and matches
            # nothing, so the projection silently returns an EMPTY folder and the feature looks
            # exactly as broken as it was before the fix. Caught by this change's own gate,
            # which is the entire argument for writing one.
            placed = (ws.get('itemFolders') or {}).get('views') or {}
            views = ws.get('views') or {}
            inside = {str(vid): dict(v) for vid, v in views.items()
                      if isinstance(v, dict) and str(placed.get(str(vid)) or '') == fid}
            return str(username), dict(hit), inside
        return None

    # ---------------------------------------------------------------- read
    def workspace(self, username, consume_corrections=True):
        """One user's durable workspace: always the full three-strata shape."""
        try:
            data = self._st.get(self.table_key) or {}
            ws = data.get(username, {}) or {}
            # A collision acknowledgement is protocol state, not part of a field definition.
            # Consume it with the first fresh workspace payload after the correcting write,
            # then splice a bounded copy into that payload only.  Keeping it out of `fields`
            # prevents an old request id surviving forever and overriding a later rename.
            corrections = {}
            if consume_corrections and ws.get('fieldCorrections'):
                def _take(current):
                    current_ws = current.get(username) or {}
                    pending = current_ws.get('fieldCorrections') or {}
                    corrections.update({
                        str(key)[:80]: dict(value)
                        for key, value in pending.items()
                        if isinstance(value, dict)
                    })
                    current_ws.pop('fieldCorrections', None)
                    return current

                data = self._st.update(self.table_key, _take, flush='async')
                ws = (data or {}).get(username, {}) or {}
        except Exception:
            ws = {}
            corrections = {}
        fields = {
            key: dict(value) if isinstance(value, dict) else value
            for key, value in (ws.get('fields') or {}).items()
        }
        for key, ack in corrections.items():
            field = fields.get(key)
            accepted_label = str(ack.get('label') or '')[:120]
            requested_label = str(ack.get('labelCorrectedFrom') or '')[:120]
            correction_id = str(ack.get('labelCorrectionId') or '')[:180]
            # A newer field write clears/replaces the pending ack in the SAME transaction.
            # The label check is an extra belt against ever attaching a stale ack to a newer
            # definition if a future store implementation weakens that ordering.
            if (isinstance(field, dict) and accepted_label
                    and str(field.get('label') or '') == accepted_label
                    and requested_label and correction_id):
                field['labelCorrectedFrom'] = requested_label
                field['labelCorrectionId'] = correction_id
        out = {
            'views': dict(ws.get('views') or {}),
            'fields': fields,
            'overlays': dict(ws.get('overlays') or {}),
            # wave-8 I11 (C4): folders over the saved views / cohorts sidebars. A FOURTH
            # stratum rather than a key on each item — see aios_grid.clean_folders for why
            # (a cohort lives in another store, and filing is an organising act, not part of
            # what a view is). Absent for every workspace saved before this wave, which is
            # exactly "no folders yet".
            'folders': dict(ws.get('folders') or {}),
            'itemFolders': dict(ws.get('itemFolders') or {}),
        }
        # 2026-07-31 (owner item 3): WHERE THE USER LEFT OFF survives a new browser. The
        # client's localStorage copy wins when present; this is the server's answer for a
        # fresh profile, which used to fall all the way to the system default view.
        if ws.get('activeViewId'):
            out['activeViewId'] = str(ws['activeViewId'])
        # Wave 2026-08-02 (C-LAYOUT): the per-user record-detail field order. A fifth
        # stratum, absent until the user first reorders — exactly "default order".
        if isinstance(ws.get('recordLayout'), dict):
            out['recordLayout'] = dict(ws['recordLayout'])
        return out

    # ---------------------------------------------------------------- write
    def _update(self, username, change):
        def _up(data):
            ws = data.setdefault(username, {})
            ws.setdefault('views', {})
            ws.setdefault('fields', {})
            ws.setdefault('overlays', {})
            ws.setdefault('folders', {})
            ws.setdefault('itemFolders', {})
            change(ws)
            return data
        # flush='async' (wave-7 W3): this is THE hot path — every autosaved filter tweak,
        # column note and typed overlay cell lands here inside the component round-trip, and
        # the historical synchronous hub commit cost seconds per edit. The mutation applies to
        # the in-process cache (read-your-writes for every subsequent render); the hub write
        # coalesces in the background. Registry/auth writes elsewhere stay flush='sync'.
        return self._st.update(self.table_key, _up, flush='async')

    def rename_choice_values(self, username, change):
        """Apply an arbitrary workspace rewrite (wave 20, item 15 / C-RENAME).

        ⚠ NAMED FOR ITS ONE CALLER RATHER THAN EXPOSED AS A GENERIC `mutate`, deliberately. A
        public "do anything to the workspace" method is an invitation to put write logic in
        callers instead of here, and every OTHER method on this class exists precisely because
        that logic belongs in one place. Renaming a choice is the one operation that must touch
        three strata AT ONCE — the field's `choices`, the cells in `overlays`, and the views that
        filter or colour by the old value — inside a SINGLE transaction, because a rename that
        updated the cells and not the filters would leave a saved view matching nothing.

        `change(ws)` receives the whole workspace with every stratum pre-created (see `_update`).
        """
        return self._update(username, change)

    def save_active_view(self, username, view_id):
        """Remember which view this user last opened (owner item 3, 2026-07-31).

        Presentation state, not authorisation: the READ side re-validates the id against what
        the caller may actually see, so a stale or foreign id degrades to the default view
        rather than granting anything. Stored per user like every other stratum.
        """
        vid = str(view_id or '').strip()[:120]
        if not vid or username == SHARED_KEY:
            return

        def _set(ws):
            ws['activeViewId'] = vid
        self._update(username, _set)

    def save_record_layout(self, username, order):
        """The per-user RECORD-DETAIL field order (wave 2026-08-02, C-LAYOUT).

        Presentation state for ONE surface — the record modal. Deliberately not view config:
        the owner's ask is per-user, not per-view, and it must never reorder grid columns.
        The event handler validated keys against the live field set; the wire re-validates at
        serve time (aios_grid.workspace_wire), so a deleted field cannot outlive itself here.
        An empty order clears the stratum back to "default order".
        """
        if username == SHARED_KEY:
            return
        clean, seen = [], set()
        for key in (order or [])[:200]:
            key = str(key or '').strip()[:80]
            if key and key not in seen:
                seen.add(key)
                clean.append(key)

        def _set(ws):
            if clean:
                ws['recordLayout'] = {'order': clean}
            else:
                ws.pop('recordLayout', None)

        self._update(username, _set)

    def save_folders(self, username, folders, item_folders):
        """Replace the folder stratum wholesale (wave-8 I11).

        Wholesale rather than per-folder because the caller has ALREADY validated the complete
        picture through aios_grid.clean_folders / clean_item_folders, and those two are
        interdependent: a placement is only legal while its folder exists, so committing them
        separately would leave a window where a reader sees an item filed into a folder that is
        not there yet. One write, one consistent state.
        """
        def _set(ws):
            ws['folders'] = dict(folders or {})
            ws['itemFolders'] = dict(item_folders or {})
        self._update(username, _set)

    def save_view_order(self, username, order):
        """⭐ WAVE-27 item 5 (contract C7) — this user's own ORDER for the views rail.

        Wholesale, like `save_folders` above and for the same reason: the client sends the full
        list it is looking at, not a delta, because a partial order cannot say where an UNNAMED
        view went.

        ⛔ PER USER, and it belongs in this stratum rather than on the view records themselves.
        `aios_grid`'s own folder note argues it out for placements and every word applies: an
        arrangement is a per-user ORGANISING act, not part of what a view IS — so keeping it out
        of the view config means duplicating, sharing or exporting a view does not drag one
        person's rail position along with it. It also means a SHARED view can sit in a different
        place for each person who can see it, which is the only coherent answer once two people
        share one view.

        An empty list CLEARS the arrangement (back to server order) rather than storing `[]`.
        """
        def _set(ws):
            clean = []
            seen = set()
            for vid in (order or []):
                vid = str(vid).strip()[:120]
                if vid and vid not in seen:
                    seen.add(vid)
                    clean.append(vid)
            if clean:
                ws['viewOrder'] = clean
            else:
                ws.pop('viewOrder', None)
        self._update(username, _set)

    def shared_views(self, viewer, is_admin=False):
        """Every SHARED view this viewer may see, by id (wave-9 I17).

        Read-only and independent of the viewer's own workspace: the caller merges. Returns
        only what `_may_see` allows, so a caller cannot accidentally render somebody else's
        personal view by forgetting to filter.
        """
        try:
            bucket = ((self._st.get(self.table_key) or {}).get(SHARED_KEY) or {}).get('views') or {}
        except Exception:
            return {}
        return {vid: dict(v) for vid, v in bucket.items()
                if _may_see(v, viewer, is_admin)}

    def shared_view(self, view_id):
        """One shared view RAW — no visibility filter. For authorisation decisions only: a
        caller must know a view exists and who owns it before it can decide whether the actor
        may touch it. Never hand the result to a renderer without checking `_may_see`."""
        try:
            return ((self._st.get(self.table_key) or {}).get(SHARED_KEY) or {}
                    ).get('views', {}).get(str(view_id))
        except Exception:
            return None

    def save_view(self, username, view, shared=None, reserved_names=(), is_admin=False):
        """Upsert a SavedView into its ONE home — personal workspace or the shared bucket.

        `shared` defaults to reading the view's own permissions (`is_shared`). Whichever home
        it lands in, the view is REMOVED from the other, so a view can never exist as two
        copies that diverge on the next edit.

        ⚠ AUTHORISATION IS THE CALLER'S JOB and must happen BEFORE this is called — this layer
        moves data and does not know who is asking. `_cl_handle_one` is the wall.
        """
        view_id = str((view or {}).get('id') or '').strip()
        if not view_id:
            raise ValueError('view id is required')
        if username == SHARED_KEY:
            raise ValueError('reserved username')
        to_shared = is_shared(view) if shared is None else bool(shared)
        requested = dict(view)
        accepted = {}

        def _up(data):
            # View names are tenant-global: every personal workspace plus the shared bucket.
            # This deliberately includes views the actor cannot see.  The only disclosed fact
            # is that a display name is already taken, while the categorical "no duplicate
            # view names" contract remains true when a personal view is later shared.
            names = list(reserved_names or ())
            for workspace in data.values():
                if not isinstance(workspace, dict):
                    continue
                names.extend(
                    value.get('name')
                    for candidate_id, value in (workspace.get('views') or {}).items()
                    if candidate_id != view_id and isinstance(value, dict)
                )
            payload = dict(requested)
            payload['name'] = _unique_name(payload.get('name'), names)
            accepted.clear()
            accepted.update(payload)
            if to_shared:
                bucket = data.setdefault(SHARED_KEY, {})
                bucket.setdefault('views', {})[view_id] = payload
                # it may have lived in the creator's workspace before being shared
                owner = data.get(payload.get('createdBy') or username) or {}
                (owner.get('views') or {}).pop(view_id, None)
            else:
                ws = data.setdefault(username, {})
                ws.setdefault('views', {})[view_id] = payload
                (data.get(SHARED_KEY, {}).get('views') or {}).pop(view_id, None)
            return data

        self._st.update(self.table_key, _up, flush='async')
        return dict(accepted)

    def delete_view(self, username, view_id):
        """Delete a custom/list view override. The system all-rows view is guarded by caller.

        Removes from BOTH homes: the caller has already authorised the delete, and leaving a
        stale copy in the other bucket would resurrect the view on the next read.
        """
        vid = str(view_id)

        def _up(data):
            (data.get(username, {}).get('views') or {}).pop(vid, None)
            (data.get(SHARED_KEY, {}).get('views') or {}).pop(vid, None)
            return data

        self._st.update(self.table_key, _up, flush='async')

    def save_field(self, username, field, reserved_names=(), correction_id=None):
        """Persist a column note or a user-created (custom_/measure_) field definition."""
        key = str((field or {}).get('key') or '').strip()
        if not key:
            raise ValueError('field key is required')
        requested = dict(field)
        accepted = {}

        def _save(ws):
            names = list(reserved_names or ())
            names.extend(
                value.get('label')
                for candidate_key, value in (ws.get('fields') or {}).items()
                if candidate_key != key and isinstance(value, dict)
            )
            payload = dict(requested)
            payload.pop('labelCorrectedFrom', None)
            payload.pop('labelCorrectionId', None)
            requested_label = ' '.join(
                str(payload.get('label') or 'Untitled').split())[:120].rstrip()
            payload['label'] = _unique_name(requested_label, names)
            corrections = ws.setdefault('fieldCorrections', {})
            corrections.pop(key, None)
            if payload['label'] != requested_label and correction_id:
                corrections[key] = {
                    'label': payload['label'],
                    'labelCorrectedFrom': requested_label,
                    'labelCorrectionId': str(correction_id)[:180],
                }
            if not corrections:
                ws.pop('fieldCorrections', None)
            accepted.clear()
            accepted.update(payload)
            if source_override_is_empty(payload):
                ws['fields'].pop(key, None)
            else:
                ws['fields'][key] = payload

        self._update(username, _save)
        return dict(accepted)

    def duplicate_field(self, username, source_key, new_key, field,
                        reserved_names=(), correction_id=None):
        """Clone a user-created field in ONE store transaction (wave-5 item 1): the new
        definition plus — for `custom_` overlay sources only — every stored cell value under
        the source key. One transaction, because a def without its values (or values without a
        def) is exactly the orphan state delete_field exists to prevent, in reverse.
        The caller validated both keys (same created stratum) and stamped the clone's
        createdBy; this layer only moves data."""
        source_key = str(source_key or '').strip()
        new_key = str(new_key or '').strip()
        if not source_key or not new_key or source_key == new_key:
            raise ValueError('duplicate_field needs two distinct keys')
        requested = dict(field)
        accepted = {}

        def _dup(ws):
            names = list(reserved_names or ())
            names.extend(
                value.get('label')
                for candidate_key, value in (ws.get('fields') or {}).items()
                if candidate_key != new_key and isinstance(value, dict)
            )
            payload = dict(requested)
            payload.pop('labelCorrectedFrom', None)
            payload.pop('labelCorrectionId', None)
            requested_label = ' '.join(
                str(payload.get('label') or 'Untitled').split())[:120].rstrip()
            payload['label'] = _unique_name(requested_label, names)
            corrections = ws.setdefault('fieldCorrections', {})
            corrections.pop(new_key, None)
            if payload['label'] != requested_label and correction_id:
                corrections[new_key] = {
                    'label': payload['label'],
                    'labelCorrectedFrom': requested_label,
                    'labelCorrectionId': str(correction_id)[:180],
                }
            if not corrections:
                ws.pop('fieldCorrections', None)
            accepted.clear()
            accepted.update(payload)
            ws['fields'][new_key] = payload
            if source_key.startswith('custom_'):
                for row in ws['overlays'].values():
                    if isinstance(row, dict) and source_key in row:
                        row[new_key] = row[source_key]

        self._update(username, _dup)
        return dict(accepted)

    def delete_field(self, username, key):
        """Delete a USER-CREATED field definition outright (owner gap closed 2026-07-27).

        Only the created strata ever reach here (`custom_` overlay fields, `measure_` formula
        columns — the caller enforces the prefix). The stored overlay VALUES for the key are
        scrubbed with it: a deleted column's cells must not linger as orphan data that would
        silently resurface if the key were ever reused. Views referencing the key self-heal on
        their next autosave (an unknown colId is dropped) — the rule every stale key rides.
        """
        key = str(key or '').strip()
        if not key:
            return

        def _drop(ws):
            ws['fields'].pop(key, None)
            for row in ws['overlays'].values():
                if isinstance(row, dict):
                    row.pop(key, None)

        self._update(username, _drop)

    def patch_overlay(self, username, pid, updates):
        """Patch only the external editable stratum; never writes to the source system."""
        clean = dict(updates or {})
        if not clean:
            return

        def _patch(ws):
            ws['overlays'].setdefault(str(int(pid)), {}).update(clean)

        self._update(username, _patch)


def make(table_key, st=None):
    return TableStore(table_key, st=st)