app/read_model.py

"""In-process read model over the central bucket (§16.1).

Two layers, per central-bucket folder:

- **Listing cache** — the folder's tree listing, refreshed at most once per
  ``LISTING_TTL_S`` behind a per-folder lock (single-flight): any number of
  concurrent readers costs at most one bucket listing per TTL window.
- **Content cache** — parsed ``{frontmatter, body}`` per file, keyed by the
  listing's ``xet_hash`` so byte-identical files (inbox copies) share one
  cached entry. Bounded by ``CONTENT_CACHE_MAX_BYTES`` with LRU eviction;
  eviction means a refetch, never an error. Cold misses are fetched in one
  **batch** download, not per file.

Coherence: the Space is the only writer to the central bucket (§2), so every
API write is inserted synchronously (``write_through``) — agents always
observe their own writes immediately, independent of TTL. Locally written
entries live in an overlay merged over bucket listings for a grace window, so
a lagging bucket listing can never make a fresh write disappear. The TTL
exists only to pick up out-of-band admin edits (verification verdicts, force
re-registrations); the per-file hash check then refreshes exactly the changed
entries, so mutable files need no special handling.

All state here is cache — restart-safe by loss (§1).
"""
from __future__ import annotations

import json
import logging
import re
import threading
import time
from collections import OrderedDict
from dataclasses import dataclass, field, replace
from typing import Any, Callable

from app.config import Settings
from app.frontmatter import parse
from app.hub import HubClient, ListedFile
from app.naming import (
    BROADCASTS_FOLDER,
    CHANNELS_FOLDER,
    VERIFICATION_STATUS_PATH,
    channel_readme_path,
)
from app.validation import NOTIFY_ALL, NOTIFY_MENTIONS, stored_notify_level


log = logging.getLogger(__name__)

_README_RE = re.compile(r"(?:^|/)README\.md$", re.IGNORECASE)

# A channel *message*: channels/{name}/{stamp}_{author}.md — depth exactly 2
# under channels/, stamped leaf. Excludes the README (the theme) and the
# members/ markers by shape, not by convention.
_CHANNEL_MSG_RE = re.compile(
    r"^channels/([a-z0-9](?:[a-z0-9-]{0,38}[a-z0-9])?)/(\d{8}-\d{6}-\d{3}_[^/]+\.md)$"
)
# A subscription marker: channels/{name}/members/{handle}.md.
_CHANNEL_MEMBER_RE = re.compile(
    r"^channels/([a-z0-9](?:[a-z0-9-]{0,38}[a-z0-9])?)/members/([^/]+)\.md$"
)

# How long a write-through entry shadows the bucket before we trust the bucket
# listing to have caught up. Generous; a write normally appears immediately.
_OVERLAY_GRACE_S = 300.0


@dataclass
class Record:
    filename: str
    path: str
    frontmatter: dict[str, Any]
    body: str
    size: int
    parse_error: bool = False
    # Why this record is in the caller's unified watch stream; set only by
    # ``updates_records`` (WATCH_DESIGN.md §4.2) and carried through the list
    # grammar into the expanded item. Every other view leaves it None.
    reasons: list[str] | None = None


@dataclass
class _Folder:
    files: dict[str, ListedFile] = field(default_factory=dict)
    fetched_at: float = float("-inf")
    overlay: dict[str, tuple[ListedFile, float]] = field(default_factory=dict)
    lock: threading.Lock = field(default_factory=threading.Lock)


def _safe_parse(raw: bytes) -> tuple[dict[str, Any], str, bool]:
    """Parse a bucket file, never raising: a malformed historical file must
    degrade to an empty-frontmatter record, not 4xx/5xx a GET."""
    try:
        text = raw.decode("utf-8")
    except UnicodeDecodeError:
        return {}, raw.decode("utf-8", errors="replace"), True
    try:
        fm, body = parse(text)
    except Exception:
        return {}, text, True
    return fm, body, False


class ReadModel:
    def __init__(
        self,
        hub: HubClient,
        settings: Settings,
        clock: Callable[[], float] = time.monotonic,
    ):
        self._hub = hub
        self._settings = settings
        self._clock = clock
        self._folders: dict[str, _Folder] = {}
        self._folders_lock = threading.Lock()
        # Parsed content keyed by xet_hash: (frontmatter, body, size, parse_error).
        self._content: OrderedDict[str, tuple[dict, str, int, bool]] = OrderedDict()
        self._content_bytes = 0
        # Write-through entries whose xet_hash isn't known yet, keyed by path.
        self._local: dict[str, tuple[dict, str, int, float]] = {}
        self._verification: tuple[str, dict[str, str]] | None = None
        self._content_lock = threading.Lock()

    # ───────────────────────── listings ─────────────────────────

    def _folder(self, folder: str) -> _Folder:
        with self._folders_lock:
            return self._folders.setdefault(folder, _Folder())

    def listing(self, folder: str) -> list[ListedFile]:
        """The folder's current listing: TTL-cached bucket truth merged with
        the local write-through overlay (overlay fills gaps, never overrides)."""
        f = self._folder(folder)
        with f.lock:
            now = self._clock()
            if now - f.fetched_at >= self._settings.listing_ttl_s:
                fresh = self._hub.list_central_dir(folder)
                if not fresh and f.files:
                    # The hub flattens listing errors to []; nothing is ever
                    # deleted from these folders, so an empty result for a
                    # previously non-empty folder is a transient failure.
                    log.warning(
                        "listing(%s) came back empty; keeping %d cached entries",
                        folder, len(f.files),
                    )
                else:
                    f.files = {e.rel_path: e for e in fresh}
                f.fetched_at = now
                f.overlay = {
                    p: (e, ts)
                    for p, (e, ts) in f.overlay.items()
                    if p not in f.files and now - ts < _OVERLAY_GRACE_S
                }
            merged = dict(f.files)
            for p, (e, _ts) in f.overlay.items():
                merged.setdefault(p, e)
            return list(merged.values())

    def _md_entries(self, folder: str) -> list[ListedFile]:
        return [
            e
            for e in self.listing(folder)
            if e.rel_path.endswith(".md") and not _README_RE.search(e.rel_path)
        ]

    # ───────────────────────── records ─────────────────────────

    def records(self, folder: str) -> list[Record]:
        """Parsed records for every .md file under ``folder`` (READMEs
        excluded), ascending by filename. Cold misses are batch-fetched."""
        out = self._resolve_many(self._md_entries(folder))
        return [out[p] for p in sorted(out)]

    def records_for(self, folder: str, paths: list[str]) -> dict[str, Record]:
        """Resolve specific files from ``folder``'s listing through the content
        cache, keyed by rel_path. For files ``records`` excludes by convention
        (READMEs) or selective reads over a tree listing (taskforces, §18).
        Unlisted paths are silently absent from the result."""
        by_path = {e.rel_path: e for e in self.listing(folder)}
        return self._resolve_many([by_path[p] for p in paths if p in by_path])

    def _resolve_many(self, entries: list[ListedFile]) -> dict[str, Record]:
        out: dict[str, Record] = {}
        misses: list[ListedFile] = []
        with self._content_lock:
            for e in entries:
                rec = self._resolve_cached(e)
                if rec is not None:
                    out[e.rel_path] = rec
                else:
                    misses.append(e)
        if misses:
            fetched = self._hub.download_many(
                self._settings.central_bucket, [e.rel_path for e in misses]
            )
            with self._content_lock:
                for e in misses:
                    raw = fetched.get(e.rel_path)
                    if raw is None:
                        continue  # transient download failure; heals next pass
                    out[e.rel_path] = self._insert(e, raw)
        return out

    def record(self, folder: str, filename: str) -> Record | None:
        """One file, resolved through the cache; None if it isn't listed."""
        path = f"{folder}/{filename}"
        entry = next((e for e in self.listing(folder) if e.rel_path == path), None)
        if entry is None:
            return None
        with self._content_lock:
            rec = self._resolve_cached(entry)
        if rec is not None:
            return rec
        raw = self._hub.download_many(self._settings.central_bucket, [path]).get(path)
        if raw is None:
            return None
        with self._content_lock:
            return self._insert(entry, raw)

    def _resolve_cached(self, e: ListedFile) -> Record | None:
        """Caller holds ``_content_lock``."""
        filename = e.rel_path.rsplit("/", 1)[-1]
        if e.xet_hash and e.xet_hash in self._content:
            self._content.move_to_end(e.xet_hash)
            fm, body, size, perr = self._content[e.xet_hash]
            return Record(filename, e.rel_path, fm, body, size, perr)
        if e.rel_path in self._local:
            fm, body, size, _ts = self._local[e.rel_path]
            return Record(filename, e.rel_path, fm, body, size, False)
        return None

    def _insert(self, e: ListedFile, raw: bytes) -> Record:
        """Caller holds ``_content_lock``."""
        fm, body, perr = _safe_parse(raw)
        if e.xet_hash:
            if e.xet_hash not in self._content:
                self._content[e.xet_hash] = (fm, body, len(raw), perr)
                self._content_bytes += len(raw)
                while (
                    self._content_bytes > self._settings.content_cache_max_bytes
                    and len(self._content) > 1
                ):
                    _, (_f, _b, sz, _p) = self._content.popitem(last=False)
                    self._content_bytes -= sz
            else:
                self._content.move_to_end(e.xet_hash)
        filename = e.rel_path.rsplit("/", 1)[-1]
        return Record(filename, e.rel_path, fm, body, len(raw), perr)

    # ───────────────────────── write-through ─────────────────────────

    def write_through(
        self, path: str, frontmatter: dict, body: str, size: int,
        folder: str | None = None,
    ) -> None:
        """Insert a just-written central-bucket file so read-after-write is
        exact regardless of listing TTL. Call right after the bucket write.

        ``folder`` pins which folder cache gets the listing overlay when it is
        not the file's immediate parent — taskforce files live under one shared
        ``taskforces`` tree listing whatever their subdirectory (§18.4)."""
        if folder is None:
            folder, _, _filename = path.rpartition("/")
        f = self._folder(folder)
        now = self._clock()
        with f.lock:
            f.overlay[path] = (ListedFile(rel_path=path, size=size, xet_hash=None), now)
        with self._content_lock:
            self._local[path] = (frontmatter, body, size, now)
            stale = [
                p for p, (_f, _b, _s, ts) in self._local.items()
                if now - ts >= _OVERLAY_GRACE_S
            ]
            for p in stale:
                del self._local[p]

    # ───────────────────────── derived views ─────────────────────────

    def registered_agents(self) -> set[str]:
        return {
            e.rel_path.rsplit("/", 1)[-1].removesuffix(".md")
            for e in self._md_entries("agents")
        }

    def inbox_records(self, handle: str) -> list[Record]:
        """The handle's inbox view: its mention/refs fan-out copies UNION every
        organizer broadcast. Broadcasts are stored once under broadcasts/ and
        merged here at read time, so a handle that never registered or joined
        after the broadcast still sees it. Deduped by filename (the same
        server-stamped name is unique), ascending by filename; callers apply
        the list grammar (order, cursor, limit)."""
        by_name: dict[str, Record] = {}
        for r in self.records(f"inbox/{handle}"):
            by_name[r.filename] = r
        for r in self.records(BROADCASTS_FOLDER):
            by_name.setdefault(r.filename, r)
        return [by_name[f] for f in sorted(by_name)]

    # ───────────────────────── channels ─────────────────────────
    # All channel reads run over the ONE recursive channels/ listing (the
    # taskforce FOLDER pattern): summaries, rosters, subscriptions, and the
    # cross-channel feed each cost at most one bucket listing per TTL window.

    def channel_exists(self, name: str) -> bool:
        """A channel exists iff its README (the theme) is listed — the same
        structural invariant as taskforces. Shared by the channels router and
        the POST /v1/messages channel gate (import-cycle-free)."""
        readme = channel_readme_path(name)
        return any(e.rel_path == readme for e in self.listing(CHANNELS_FOLDER))

    def channel_message_records(self, name: str) -> list[Record]:
        """One channel's messages (stamped files only — README and member
        markers excluded by shape), ascending by filename."""
        paths = [
            e.rel_path
            for e in self.listing(CHANNELS_FOLDER)
            if (m := _CHANNEL_MSG_RE.match(e.rel_path)) and m.group(1) == name
        ]
        recs = self.records_for(CHANNELS_FOLDER, paths)
        return [recs[p] for p in sorted(recs)]

    def channel_subscriptions(self, handle: str) -> list[str]:
        """Channel names the handle subscribes to — derived by filtering the
        cached listing for its member markers; zero content reads."""
        return sorted(
            {
                m.group(1)
                for e in self.listing(CHANNELS_FOLDER)
                if (m := _CHANNEL_MEMBER_RE.match(e.rel_path))
                and m.group(2) == handle
            }
        )

    def channel_notify_levels(self, handle: str) -> dict[str, str]:
        """``{channel: notify level}`` for every channel the handle is a member
        of, name-sorted. ``all`` when the marker carries ``notify: all``,
        ``mentions`` otherwise — an absent or unrecognised value reads as the
        quiet default, so every pre-existing (and backfilled) membership is
        correct without a migration.

        The sibling ``channel_subscriptions`` answers membership from marker
        *paths* alone at zero content reads; levels need marker *content*, so
        this costs one read per marker — resolved through the same
        hash-keyed content cache as every other record, so a steady state
        downloads nothing. Callers that only need membership keep the free
        path."""
        markers: dict[str, str] = {}
        for e in self.listing(CHANNELS_FOLDER):
            m = _CHANNEL_MEMBER_RE.match(e.rel_path)
            if m and m.group(2) == handle:
                markers[e.rel_path] = m.group(1)
        recs = self.records_for(CHANNELS_FOLDER, list(markers))
        levels: dict[str, str] = {}
        for path, name in markers.items():
            rec = recs.get(path)
            levels[name] = (
                stored_notify_level(rec.frontmatter) if rec else NOTIFY_MENTIONS
            )
        return dict(sorted(levels.items()))

    def updates_records(self, handle: str) -> list[Record]:
        """The handle's unified watch stream (WATCH_DESIGN.md §4.2): its inbox
        (mentions/refs wherever they were posted, plus organizer broadcasts)
        UNION the full traffic of only those channels it has flipped to
        ``notify: all``. Channels left at the quiet default contribute nothing
        here — their @mentions still arrive via the inbox side.

        Deduped by filename: a channel post that also @mentions you exists twice
        in the bucket (the channel copy and the inbox fan-out copy) and must be
        delivered exactly once, carrying BOTH reasons. Sorted by
        (filename, path) like ``channel_feed_records``, so one filename cursor
        covers the whole union — stamps are server-issued and per-author
        monotonic, which makes filenames globally unique and lexical order
        chronological order."""
        reasons: dict[str, list[str]] = {}
        by_name: dict[str, Record] = {}
        for r in self.inbox_records(handle):
            by_name[r.filename] = r
            # Provenance is the path: a broadcast is the one shared copy under
            # broadcasts/, everything else got here by @mention or refs.
            reasons[r.filename] = [
                "broadcast" if r.path.startswith(f"{BROADCASTS_FOLDER}/") else "mention"
            ]
        for name, level in self.channel_notify_levels(handle).items():
            if level != NOTIFY_ALL:
                continue
            for r in self.channel_message_records(name):
                by_name.setdefault(r.filename, r)
                reasons.setdefault(r.filename, []).append(f"channel:{name}")
        return sorted(
            (replace(r, reasons=reasons[fn]) for fn, r in by_name.items()),
            key=lambda r: (r.filename, r.path),
        )

    def channel_feed_records(self, handle: str) -> list[Record]:
        """The handle's cross-channel feed: the union of every subscribed
        channel's messages (CHANNELS_DESIGN.md §4). Records are keyed by
        rel_path — two channels can mint the same {stamp}_{author} filename,
        and both must survive the union — then sorted (filename, path) so the
        list grammar's filename cursors stay chronological."""
        subs = set(self.channel_subscriptions(handle))
        if not subs:
            return []
        paths = [
            e.rel_path
            for e in self.listing(CHANNELS_FOLDER)
            if (m := _CHANNEL_MSG_RE.match(e.rel_path)) and m.group(1) in subs
        ]
        recs = self.records_for(CHANNELS_FOLDER, paths)
        return sorted(recs.values(), key=lambda r: (r.filename, r.path))

    # ───────────────────────── delete-through ─────────────────────────

    def delete_through(self, path: str, folder: str | None = None) -> None:
        """Remove a just-deleted central-bucket file from the caches so
        read-after-delete is exact regardless of listing TTL — the inverse of
        ``write_through``, and like it called right after the bucket write.
        Without this, a recently written overlay entry (grace window 300s)
        would resurrect the file long after the bucket forgot it. Channel
        unsubscribe is the only caller (nothing else deletes)."""
        if folder is None:
            folder, _, _filename = path.rpartition("/")
        f = self._folder(folder)
        with f.lock:
            f.files.pop(path, None)
            f.overlay.pop(path, None)
        with self._content_lock:
            self._local.pop(path, None)

    def invalidate_verification_index(self) -> None:
        """Drop the cached verification index after the Space itself rewrites
        it (automated verdicts, §5.7) — that write is no longer an out-of-band
        admin edit, so it must not wait out the listing TTL. The next
        ``verification_index()`` call refetches the file (one download)."""
        with self._content_lock:
            self._verification = None

    def verification_index(self) -> dict[str, str]:
        """Parsed ``results/verification_status.json``, cached by its listing
        hash. Absent or unreadable → {} (every result then reads as pending —
        the truthful default for an unreviewed result)."""
        entry = next(
            (e for e in self.listing("results") if e.rel_path == VERIFICATION_STATUS_PATH),
            None,
        )
        if entry is None:
            return {}
        with self._content_lock:
            if (
                entry.xet_hash
                and self._verification is not None
                and self._verification[0] == entry.xet_hash
            ):
                return self._verification[1]
        raw = self._hub.download_many(
            self._settings.central_bucket, [VERIFICATION_STATUS_PATH]
        ).get(VERIFICATION_STATUS_PATH)
        if raw is None:
            return {}
        try:
            data = json.loads(raw.decode("utf-8"))
        except (json.JSONDecodeError, UnicodeDecodeError) as exc:
            log.error("verification index unparseable: %s", exc)
            return {}
        if not isinstance(data, dict):
            log.error("verification index is not a JSON object")
            return {}
        index = {str(k): str(v) for k, v in data.items()}
        if entry.xet_hash:
            with self._content_lock:
                self._verification = (entry.xet_hash, index)
        return index