platform/core/alerts.py

"""core/alerts.py — "tell me when a record ENTERS this view" (wave 20, owner item 25).

THE OWNER'S SHAPE, verbatim: *"a user click a view and click 'Create alert'. This way when a
Record gets into that Filter's criteria, a user gets notified in that module … Read/Unread like an
inbox type of thing so they can confirm whether they have seen the Record that got filtered in."*

So an alert is **a view plus a remembered MATCHED SET**, and a notification is a **new entrant** —
a pid that matches now and did not last time. That definition is the whole design, and it is what
makes the module trustworthy rather than noisy:

  * a record already in the view when the alert is created is NOT news. The first evaluation
    SEEDS the set silently (`seed=True`); without that, creating an alert on a 400-row view
    would announce 400 "new" records, and the user would turn the feature off in one click;
  * a record that leaves and comes back IS news again — it re-entered the criteria, which is the
    event the owner described;
  * a record that merely CHANGES while staying in the view is not an entrant. "Still matching" is
    not an event. (An on-change alert is a different feature with a different noise budget; it is
    NOT smuggled in here under the same name.)

⚠ **WHY THE MATCHED SET IS STORED AND NOT RECOMPUTED FROM HISTORY.** There is no row-level audit
log to diff against — `store` holds current state, and Odoo is read-only. The remembered set IS
the history, so it is written on every evaluation, in the same transaction that queues the
notifications. Losing that write while keeping the notifications would re-announce the same
records forever.

⚠ **WHAT AN UNANSWERABLE LEAF ACTUALLY DOES HERE — CHECKED, NOT ASSUMED.** `harness.filter_eval`'s
`EvalCtx` is explicit that each of its four members "absent, the condition matches NOTHING rather
than everything" — the engine NARROWS on an unresolved measure/cohort/rank set rather than
widening. So the failure mode is not a false alert; it is a **missed** one, and a missed entrant
is re-detected on the next evaluation because `matched` only ever records what genuinely matched.

That is the safe direction, and it is why `partial` is a caller-supplied flag rather than
something inferred in here: the caller (`routes_alerts`) is the only layer that knows whether it
could BUILD a complete context. When it says so, `evaluate()` records the attempt and changes
nothing — updating `matched` from a narrowed evaluation would drop rows out of the remembered set
and then re-announce every one of them as an "entrant" on the next complete pass. That is the
real hazard: not a phantom alert, but a storm of stale ones.

⚠ **EVERY STAMP CARRIES A UTC OFFSET** (DEBT D-18). A browser can only render a timestamp
relatively if it can subtract it from `now`, and a naive container-local stamp is unsubtractable —
it renders VERBATIM or, worse, silently as the reader's own zone. `_now_iso()` is the one place
that is decided.
"""
import datetime as _dt

import core.store as store

#: Alert DEFINITIONS, per tenant: {alert_id: {...}}. Separate from the notifications bucket so a
#: busy inbox never rewrites the definitions (and a definition edit never rewrites the inbox).
ALERTS_KEY = 'alerts'
#: The inbox: {username: [notification, ...]}, newest last.
NOTIFICATIONS_KEY = 'alert_notifications'

#: Per user. An inbox is a WORKING queue, not an archive — the store is a JSON blob read whole on
#: every request, so an unbounded inbox is a payload that grows without limit for a user who
#: never clicks. Oldest READ entries are dropped first; unread ones survive the cap because the
#: whole point is that the user has not seen them yet.
MAX_NOTIFICATIONS = 200


def _now_iso():
    """UTC, WITH the offset — `2026-08-05T09:41:07.123456+00:00`. See the module note (D-18)."""
    return _dt.datetime.now(_dt.timezone.utc).isoformat()


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


def list_alerts(user=None, st=None, is_admin=False):
    """Alert definitions this user may see: their own. Admins see all (they support them)."""
    try:
        recs = _st(st).get(ALERTS_KEY) or {}
    except Exception:
        return []
    out = []
    for aid, rec in recs.items():
        if not isinstance(rec, dict):
            continue
        if user and not is_admin and str(rec.get('owner') or '') != str(user):
            continue
        out.append({**rec, 'id': str(aid)})
    return sorted(out, key=lambda r: str(r.get('createdAt') or ''))


def create(alert_id, *, view_id, topic, owner, label='', st=None):
    """Register an alert on a view. Returns the record.

    The view is stored BY ID, never by a copy of its filter tree. An alert whose criteria were
    snapshotted at creation would silently stop matching the view the moment its owner edited it —
    and the user's mental model is "alert me on THIS VIEW", not "on this view as it was in
    August".
    """
    rec = {'id': str(alert_id), 'viewId': str(view_id), 'topic': str(topic),
           'owner': str(owner), 'label': str(label or '')[:160],
           'createdAt': _now_iso(), 'matched': [], 'seeded': False,
           'lastRunAt': None, 'lastError': None}

    def _apply(data):
        data[str(alert_id)] = rec
        return data

    _st(st).update(ALERTS_KEY, _apply, flush='sync')
    return rec


def drop_topic(topic, st=None):
    """Delete every alert DEFINITION on a topic (wave 21, item 6a / C3 — a deleted database's
    alerts die with it; an alert on a view that no longer exists can only ever error).
    Notifications already delivered are left in inboxes: they are history, and history is the
    one thing a delete must not rewrite."""
    t = str(topic or '')
    if not t:
        return

    def _apply(data):
        for aid in [k for k, r in (data or {}).items()
                    if isinstance(r, dict) and str(r.get('topic') or '') == t]:
            data.pop(aid, None)
        return data

    _st(st).update(ALERTS_KEY, _apply, flush='async')


def delete(alert_id, st=None):
    def _apply(data):
        data.pop(str(alert_id), None)
        return data

    _st(st).update(ALERTS_KEY, _apply, flush='sync')
    return True


def evaluate(alert_id, matching_pids, *, labels=None, partial=False, st=None):
    """Fold a fresh evaluation of one alert's view into its state; queue any NEW entrants.

    `matching_pids` is what the view matches NOW — the caller owns running the filter, because
    only it knows the topic's rows and the reader's scope. `partial` says the evaluation could not
    answer every leaf; see the module note on why that suppresses everything.

    Returns `{'new': [...], 'seeded': bool}` or `{'skipped': 'partial'|'missing'}`.
    """
    now_set = {str(p) for p in (matching_pids or ())}
    outcome = {}

    def _apply(data):
        rec = data.get(str(alert_id))
        if not isinstance(rec, dict):
            outcome['skipped'] = 'missing'
            return data
        if partial:
            # Record the ATTEMPT (so "last checked" is honest) but change nothing else. Writing
            # `matched` here would bake a widened set in as truth, and the next COMPLETE
            # evaluation would then report every genuinely-absent row as an entrant.
            rec['lastRunAt'] = _now_iso()
            rec['lastError'] = ('the view could not be fully evaluated (a filter leaf was '
                                'unanswerable); no alert was raised')
            outcome['skipped'] = 'partial'
            data[str(alert_id)] = rec
            return data
        prior = {str(p) for p in (rec.get('matched') or ())}
        seeding = not rec.get('seeded')
        entrants = [] if seeding else sorted(now_set - prior, key=lambda s: (len(s), s))
        rec['matched'] = sorted(now_set, key=lambda s: (len(s), s))
        rec['seeded'] = True
        rec['lastRunAt'] = _now_iso()
        rec['lastError'] = None
        data[str(alert_id)] = rec
        outcome['new'] = entrants
        outcome['seeded'] = seeding
        outcome['_rec'] = rec
        return data

    _st(st).update(ALERTS_KEY, _apply, flush='sync')
    if outcome.get('skipped'):
        return {'skipped': outcome['skipped']}
    rec = outcome.pop('_rec', {}) or {}
    if outcome.get('new'):
        _queue(rec, outcome['new'], labels or {}, st=st)
    return outcome


def _queue(rec, pids, labels, st=None):
    """Append one notification per entrant to the alert owner's inbox."""
    at = _now_iso()
    items = [{'id': f"{rec.get('id')}:{pid}:{at}",
              'alertId': str(rec.get('id')), 'viewId': str(rec.get('viewId')),
              'topic': str(rec.get('topic')), 'rowId': str(pid),
              'label': str(labels.get(str(pid)) or labels.get(pid) or pid),
              'alertLabel': str(rec.get('label') or ''),
              'at': at, 'read': False}
             for pid in pids]
    owner = str(rec.get('owner') or '')

    def _apply(data):
        inbox = list(data.get(owner) or [])
        inbox.extend(items)
        if len(inbox) > MAX_NOTIFICATIONS:
            # Drop READ entries oldest-first; keep every unread one. A cap that dropped unread
            # notifications would silently lose exactly the records the user has not confirmed —
            # the one thing this module promises not to do.
            unread = [n for n in inbox if not n.get('read')]
            read = [n for n in inbox if n.get('read')]
            keep_read = read[max(0, len(unread) + len(read) - MAX_NOTIFICATIONS):]
            inbox = sorted(unread + keep_read, key=lambda n: str(n.get('at') or ''))
        data[owner] = inbox
        return data

    _st(st).update(NOTIFICATIONS_KEY, _apply, flush='async')


def notify(owner, label, *, topic='automation', key='', row_id='', detail='', st=None):
    """⭐ WAVE 23 (C6) — put ONE notification in a named user's inbox, for a producer that is not
    a view-alert evaluation.

    The automation engine needs the bell to ring when records arrive at a review gate, and that
    is not an alert over a view: there is no filter tree, no `matched` set, no entrant diff. It
    reached for `_queue` directly during the wave, which would have made a private function a
    cross-module contract — so this is the public door instead, and `_queue` stays the internal
    half of `evaluate()`.

    The inbox shape is UNCHANGED, deliberately: `AlertsPane` reads one list, and a second row
    shape would mean a client that must branch on where a notification came from. `alertId`
    carries the producer's key so a click-through can route (W23-W5).
    """
    owner = str(owner or '').strip()
    if not owner or not label:
        return None
    at = _now_iso()
    item = {'id': f"{key or topic}:{row_id or 'n'}:{at}",
            'alertId': str(key or ''), 'viewId': '', 'topic': str(topic),
            'rowId': str(row_id or ''), 'label': str(detail or label),
            'alertLabel': str(label), 'at': at, 'read': False}

    def _apply(data):
        box = list(data.get(owner) or [])
        box.append(item)
        if len(box) > MAX_NOTIFICATIONS:
            unread = [n for n in box if not n.get('read')]
            read = [n for n in box if n.get('read')]
            keep_read = read[max(0, len(unread) + len(read) - MAX_NOTIFICATIONS):]
            box = sorted(unread + keep_read, key=lambda n: str(n.get('at') or ''))
        data[owner] = box
        return data

    _st(st).update(NOTIFICATIONS_KEY, _apply, flush='async')
    return item


def inbox(user, st=None, limit=100):
    """`{'unread': int, 'items': [...]}` — newest first, for the badge and the pane."""
    try:
        items = list((_st(st).get(NOTIFICATIONS_KEY) or {}).get(str(user)) or [])
    except Exception:
        return {'unread': 0, 'items': []}
    items = [n for n in items if isinstance(n, dict)]
    items.sort(key=lambda n: str(n.get('at') or ''), reverse=True)
    return {'unread': sum(1 for n in items if not n.get('read')), 'items': items[:max(0, limit)]}


def mark_read(user, ids, read=True, st=None):
    """Mark specific notifications read/unread. `ids=None` marks every one (mark-all-read)."""
    want = None if ids is None else {str(i) for i in ids}

    def _apply(data):
        inbox_ = [dict(n) for n in (data.get(str(user)) or []) if isinstance(n, dict)]
        for n in inbox_:
            if want is None or str(n.get('id')) in want:
                n['read'] = bool(read)
        data[str(user)] = inbox_
        return data

    _st(st).update(NOTIFICATIONS_KEY, _apply, flush='async')
    return inbox(user, st=st)


def after_write(topic_key, st=None, runner=None):
    """The write-path HOOK: re-evaluate every alert whose view belongs to `topic_key`.

    ⚠ CALLED FROM A WRITE PATH, SO IT MUST NOT RAISE AND MUST NOT BLOCK ON A FULL POOL BUILD.
    A failed alert evaluation must never fail the edit that triggered it — the user typed into a
    cell; whether an alert fires is not their problem. `runner` is injected by the caller
    (`routes_alerts` passes one that can resolve the topic's rows for the alert's owner), so this
    module never reaches for a pool itself and stays unit-testable with a fake.
    """
    if runner is None:
        return {'evaluated': 0}
    n = 0
    for rec in list_alerts(st=st):
        if str(rec.get('topic')) != str(topic_key):
            continue
        try:
            runner(rec)
            n += 1
        except Exception:                      # noqa: BLE001 — see the docstring
            continue
    return {'evaluated': n}