platform/modules/cohort.py

"""Cohort — a FIXED set of customers the user curates by hand (owner item 6, 2026-07-26).

The difference from the Data module, and the only one: **a Cohort's membership is not a query.**

  Data    one row per customer across the whole scoped book. Which rows you see is whatever the
          view's FILTER matches, so the set moves when the data moves.
  Cohort  a list of customer ids somebody chose. You can still filter, sort, group and add
          columns exactly as on Data — but those only narrow what is DISPLAYED. The set itself
          changes only when a person adds or removes a row.

That distinction is the whole point: a formula list silently re-populates ("who is overdue
today"), a cohort does not ("the 40 accounts we agreed to call in Q3"). If a cohort could drift,
it would not be worth having — you could just save a view.

Rows come from `modules.customer_data.pool()` — the SAME validated metric surface the Data module
renders, so a customer's numbers cannot differ between the two pages. This module owns only
MEMBERSHIP.

Persistence: HF store key `customer_cohorts`, shaped {username: {cohort_id: {...}}}. Per-user
like the saved views it sits beside, and written through `store.update` (strict
read-modify-write — never get+put around a mutation; that is the restart-wipe bug).

⭐ WAVE 19 / OWNER RULING R9 — A COHORT BELONGS TO ITS DATABASE. "It should stay in Product
database … for ANY database new/old": until relational features land, database functionalities
are purely independent. So this module is now SCOPE-PARAMETERIZED rather than restricted — one
bucket per topic (`<scope>_cohorts`), reached through `scoped(scope)`.

  ⛔ THE ID SPACES ARE WHY, not tidiness. A cohort stores the TOPIC's row ids: customer pids are
     Odoo partner ids, product pids are CRC32 hashes of the SKU code, a user table's are its own
     row numbers. One shared bucket and a hash collision silently puts a product in somebody's
     customer list — the same argument `routes_products` makes about the overlay bucket, and the
     reason wave 16 answered it with `with_cohorts=False` (a refusal) instead of a bucket.

  ⛔ THE CUSTOMER TOPIC KEEPS THE LEGACY KEY — zero migration. `customer` and `cohort` are ONE
     topic (the Cohort surface is the customer table over hand-curated sets), so both resolve to
     `customer_cohorts` and every shipped list is found exactly where it already lives.

  ⛔ `scoped('customer')` RETURNS THIS MODULE, deliberately, so the module-level functions stay
     the customer implementation rather than becoming a second copy of it. Callers that patch
     `modules.cohort.visible` (the gates do) keep working through the facade, and every existing
     positional call site is byte-unaffected.

TENANCY, STATED NOT HIDDEN (booked as DEBT, wave 19): this module writes through the MODULE-LEVEL
`core.store`, i.e. tenant #0's dataset repo — the same place `customer_docs` and the customer /
product table workspaces already live. It is safe rather than correct: the outer key is the
USERNAME, and `core/users.py`'s registry is ONE global bucket, so a username belongs to exactly
one tenant and no cross-tenant READ is reachable. What is wrong is RESIDENCY (tenant B's lists
sit in tenant #0's repo). Fixing it means threading the runtime handle through `EventCtx`, which
needs `routes_grid._ctx` — another session's file this wave.

No validate(): there is no Odoo aggregate to reconcile a hand-picked list against. Its counts
derive from the rows it renders (rule 8b), and the metrics themselves are already validated by
the customers module they come from.
"""
import datetime as _dt
import sys as _sys

import core.store as store

KEY = 'customer_cohorts'

#: The scope names that mean THE CUSTOMER BOOK. `cohort` is the customer table over curated sets
#: — the same rows, the same pids, the same lists — so it must resolve to the same bucket, and ''
#: is what an unset `scope_key` looks like. Anything else is another database.
LEGACY_SCOPES = frozenset({'', 'customer', 'cohort'})

#: A cohort is a list of ids, not a query — but ids are cheap and a runaway list would bloat the
#: per-user store blob, which is read on every page render. Generous enough to never bind in
#: practice (the whole book is ~1,550) and small enough to bound the payload.
MAX_MEMBERS = 5000
MAX_COHORTS = 100


def key_for(scope):
    """The store bucket ONE topic's cohorts live in (R9).

    Customer (and its `cohort` surface) keep `customer_cohorts` — the legacy key, so no shipped
    list moves. Every other database gets `<scope>_cohorts`, which for a user table is
    `ut_<slug>_cohorts`: the same naming the table's own workspace bucket already uses
    (`<scope>_table_workspace`), so one database's durable state reads as one family.
    """
    s = str(scope or 'customer').strip().lower()
    return KEY if s in LEGACY_SCOPES else f'{s}_cohorts'


def noun_for(scope):
    """What ONE row of this topic is called — the "Added 12 customers to …" toast's noun.

    A hardcoded "customer" was the visible half of the leak this ruling closes: adding twelve
    SKUs to a product list said "Added 12 customers", which is the surface telling the user the
    wrong thing about where their list went. Unknown topics say "record" rather than guessing a
    label from a slug — a user table is called whatever its creator named it, and inventing a
    singular from `ut_supplier_quotes` would be worse than the honest generic.
    """
    s = str(scope or 'customer').strip().lower()
    if s in LEGACY_SCOPES:
        return 'customer'
    if s == 'product':
        return 'product'
    return 'record'


def _now():
    return _dt.datetime.now().strftime('%Y-%m-%d %H:%M')


def _all_for(key, username):
    """{cohort_id: {name, members, note, created, updated}} for one user, from ONE bucket.

    Returns {} when the store is unavailable rather than raising: a cohort list that cannot load
    should degrade to "no cohorts", never to a broken page.
    """
    try:
        return dict((store.get(key) or {}).get(username, {}) or {})
    except Exception:
        return {}


def all_for(username):
    """{cohort_id: {...}} for one user on the CUSTOMER book — see `_all_for`."""
    return _all_for(KEY, username)


def _clean_members(raw, allowed_pids=None):
    """Ints only, de-duplicated, ORDER PRESERVED, and intersected with what this user may see.

    ⚠ The `allowed_pids` intersection is a PERMISSION boundary, not tidying. A cohort is stored
    per user but the ids in it are customer ids: if an agent's scope narrows (or a cohort is ever
    shared), replaying stored ids unchecked would render rows outside their book. Filtering on
    READ means a scope change takes effect immediately, without rewriting anyone's stored data.
    """
    out, seen = [], set()
    for v in list(raw or [])[:MAX_MEMBERS * 2]:
        try:
            pid = int(v)
        except (TypeError, ValueError):
            continue
        if pid in seen:
            continue
        if allowed_pids is not None and pid not in allowed_pids:
            continue
        seen.add(pid)
        out.append(pid)
    return out[:MAX_MEMBERS]


def _visible(key, username, allowed_pids=None):
    """Every cohort in ONE bucket, with membership already scoped to what this user may open."""
    return {cid: {**c, 'members': _clean_members(c.get('members'), allowed_pids)}
            for cid, c in _all_for(key, username).items()}


def visible(username, allowed_pids=None):
    """Every CUSTOMER cohort, membership scoped to what this user may open."""
    return _visible(KEY, username, allowed_pids)


def new_id(name):
    """A stable id from the name plus a suffix, so two cohorts called "Q3 calls" can coexist.

    ⚠ Topic-FREE on purpose. The id is a handle stored inside the topic's own bucket, and a saved
    view's `cohortLock` names it — so encoding the topic in the string would only add a second
    place for the two to disagree. Two databases may each hold a `cohort_q3_calls_101112`; they
    are different objects in different buckets and neither read can reach the other.
    """
    slug = ''.join(ch if ch.isalnum() else '_' for ch in str(name).lower()).strip('_')[:40]
    return f"cohort_{slug or 'list'}_{_dt.datetime.now().strftime('%H%M%S')}"


def _create(key, username, name, members=(), note=''):
    """Create a cohort in ONE bucket and return its id. Raises ValueError on a blank name or when
    the per-user cap is reached — refusing loudly beats silently dropping somebody's list.

    The cap is PER BUCKET, i.e. per database, which is the honest reading of R9: a hundred
    product lists must not be spent by a hundred customer lists somebody else's page created.
    """
    name = str(name or '').strip()[:120]
    if not name:
        raise ValueError('a cohort needs a name')
    existing = _all_for(key, username)
    if len(existing) >= MAX_COHORTS:
        raise ValueError(f'you already have {MAX_COHORTS} cohorts — delete one first')
    cid = new_id(name)
    record = {'name': name, 'members': _clean_members(members), 'note': str(note or '')[:2000],
              'created': _now(), 'updated': _now()}

    def _up(data):
        data.setdefault(username, {})[cid] = record
        return data
    store.update(key, _up)
    return cid


def create(username, name, members=(), note=''):
    """Create a CUSTOMER cohort and return its id — see `_create`."""
    return _create(KEY, username, name, members=members, note=note)


def _mutate(key, username, cohort_id, change):
    def _up(data):
        book = data.setdefault(username, {})
        c = book.get(cohort_id)
        if c is None:
            return data                      # deleted concurrently — nothing to do, not an error
        change(c)
        c['updated'] = _now()
        return data
    store.update(key, _up)


def _add_members(key, username, cohort_id, pids):
    """Union new ids into a cohort, PRESERVING the existing order and appending the new ones.

    Additive on purpose: "add to list" from a view can be clicked twice, and the second click
    must be a no-op rather than a reorder or a duplicate.
    """
    def _change(c):
        c['members'] = _clean_members(list(c.get('members') or []) + list(pids or []))
    _mutate(key, username, cohort_id, _change)


def add_members(username, cohort_id, pids):
    """Union new ids into a CUSTOMER cohort — see `_add_members`."""
    _add_members(KEY, username, cohort_id, pids)


def _remove_members(key, username, cohort_id, pids):
    drop = {int(p) for p in (pids or []) if str(p).lstrip('-').isdigit()}

    def _change(c):
        c['members'] = [p for p in _clean_members(c.get('members')) if p not in drop]
    _mutate(key, username, cohort_id, _change)


def remove_members(username, cohort_id, pids):
    _remove_members(KEY, username, cohort_id, pids)


def _rename(key, username, cohort_id, name):
    name = str(name or '').strip()[:120]
    if not name:
        raise ValueError('a cohort needs a name')
    _mutate(key, username, cohort_id, lambda c: c.__setitem__('name', name))


def rename(username, cohort_id, name):
    _rename(KEY, username, cohort_id, name)


def _set_note(key, username, cohort_id, note):
    _mutate(key, username, cohort_id, lambda c: c.__setitem__('note', str(note or '')[:2000]))


def set_note(username, cohort_id, note):
    _set_note(KEY, username, cohort_id, note)


def _delete(key, username, cohort_id):
    def _up(data):
        (data.get(username) or {}).pop(cohort_id, None)
        return data
    store.update(key, _up)


def delete(username, cohort_id):
    _delete(KEY, username, cohort_id)


def rows_for(cohort, pool_rows):
    """The cohort's rows, in MEMBERSHIP order, from an already-built pool.

    Membership order, not pool order: the user built this list, and the order they built it in is
    information. Ids with no matching pool row are dropped silently — a customer can fall out of
    the 24-month pool without the cohort being wrong, and showing a blank row would be worse than
    showing fewer. `missing_count` lets the page say so honestly rather than hiding it.
    """
    by_pid = {r['pid']: r for r in pool_rows}
    members = _clean_members(cohort.get('members'))
    rows = [by_pid[p] for p in members if p in by_pid]
    return rows, len(members) - len(rows)


# ─────────────────────────────────────────────── R9: ONE topic's cohorts, as an object
class CohortStore:
    """The cohort verbs for ONE non-customer database, closed over its bucket key.

    Same method names and same argument order as the module functions above, so a caller written
    against `modules.cohort` reads identically whichever it holds — which is the point: the write
    handler resolves the store once (`scoped(ctx.scope_key)`) and every branch below it stays the
    code it already was.
    """

    __slots__ = ('scope', 'key', 'noun')

    def __init__(self, scope):
        self.scope = str(scope or 'customer').strip().lower()
        self.key = key_for(self.scope)
        self.noun = noun_for(self.scope)

    def __repr__(self):
        return f'CohortStore({self.scope!r} -> {self.key!r})'

    def all_for(self, username):
        return _all_for(self.key, username)

    def visible(self, username, allowed_pids=None):
        return _visible(self.key, username, allowed_pids)

    def create(self, username, name, members=(), note=''):
        return _create(self.key, username, name, members=members, note=note)

    def add_members(self, username, cohort_id, pids):
        _add_members(self.key, username, cohort_id, pids)

    def remove_members(self, username, cohort_id, pids):
        _remove_members(self.key, username, cohort_id, pids)

    def rename(self, username, cohort_id, name):
        _rename(self.key, username, cohort_id, name)

    def set_note(self, username, cohort_id, note):
        _set_note(self.key, username, cohort_id, note)

    def delete(self, username, cohort_id):
        _delete(self.key, username, cohort_id)


def scoped(scope=None):
    """The cohort store for ONE topic — THE resolution point (R9).

    ⛔ Returns THIS MODULE for the customer topic, not a `CohortStore(KEY)`. Two reasons, and both
    are about not creating a second implementation of the shipped behaviour:
      1. the module-level functions ARE the customer store, so the legacy path keeps executing
         exactly the code the gates measure — no wrapper to drift from;
      2. tests patch `modules.cohort.visible` (verify_grid_events' folder legs do), and a facade
         that returned a bound object would silently ignore the patch, leaving a gate green over
         code it is no longer exercising.
    """
    if str(scope or 'customer').strip().lower() in LEGACY_SCOPES:
        return _sys.modules[__name__]
    return CohortStore(scope)


#: What ONE row of the customer topic is called — the module facade's answer to `CohortStore.noun`,
#: so `scoped(x).noun` reads the same whichever object `x` resolved to.
noun = 'customer'