"""Topic Feed Lab: browse and follow new Hub items by topic, from precomputed scores (FastAPI on Granian, Docker Space).

Scores come from the private bucket davanstrien/hub-firehose (GLiNER2.5-Decide, 48-topic catalogue cat48-v1),
mounted in the Space at FIREHOSE_DIR (default /data). The page itself runs no model; a background worker process
(firehose_worker.py, see supervise_worker) scores new Hub items on CPU and writes them to the same bucket.

Data in memory: a Polars table of item metadata, a NumPy float32 matrix of effective scores (broad topic =
max(broad, narrow), computed once per refresh) and NumPy arrays for the columns requests filter and sort on.
"""
import glob
import json
import os
import threading
import time
from contextlib import asynccontextmanager
from datetime import datetime, timedelta, timezone

import numpy as np
import orjson
import pandas as pd
import polars as pl
from fastapi import FastAPI, Query
from fastapi.middleware.gzip import GZipMiddleware
from fastapi.responses import HTMLResponse, Response

HERE = os.path.dirname(os.path.abspath(__file__))
DATA = os.environ.get("FIREHOSE_DIR", "/data")
REFRESH_S = 20  # items: the worker writes a new parquet file every few minutes
EVENTS_REFRESH_S = 8
ACTIVE_WITHIN = timedelta(minutes=10)  # worker counts as active if it logged an event this recently
CACHE_TTL_S = 20
TRENDING_REFRESH_S = 15 * 60
TRENDING_TOP_N = 1000
# Custom (user-defined) topics need a model. They are planned for the worker, not the Space.
CUSTOM_TOPICS_ENABLED = False

META_COLS = ["id", "source", "url", "title", "created_at", "last_modified", "snippet", "tags", "likes", "downloads",
             "upvotes", "scored_at", "scorer", "has_files"]
PAGE_COLS = ["id", "source", "url", "title", "snippet", "tags", "likes", "downloads", "upvotes"]
SOURCES = ["model", "dataset", "paper"]
NEW_WITHIN = timedelta(days=7)
NS = 1_000_000_000


def _version(v):
    """'cat48-v1.1' -> (1, 1): compare catalogue versions."""
    return tuple(int(x) for x in v.split("-v")[-1].split("."))


def load_catalogue():
    """Newest catalogue with the same labels: the bucket's copies, or the one shipped with the Space."""
    paths = glob.glob(os.path.join(DATA, "catalogue", "cat48-v*.json")) + glob.glob(os.path.join(HERE, "cat48-v*.json"))
    cats = []
    for p in paths:
        with open(p, encoding="utf-8") as f:
            cats.append(json.load(f))
    labels = cats[0]["labels"]
    cats = [c for c in cats if c["labels"] == labels]
    return max(cats, key=lambda c: _version(c["catalogue_version"]))


CAT = load_catalogue()
LABELS = CAT["labels"]
L = {t: i for i, t in enumerate(LABELS)}
DEFAULT_THR = np.array([CAT["thresholds"][t] for t in LABELS], dtype=np.float32)
NARROW = {}  # broad index -> [narrow indices]; a broad topic's effective score is max(broad, narrow)
for b, n in CAT["pairs"]:
    NARROW.setdefault(L[b], []).append(L[n])


def _ns(col):
    """ISO-8601 string column -> Int64 ns since epoch (UTC), null when unparsable."""
    return (pl.col(col).str.to_datetime(time_zone="UTC", strict=False, time_unit="ns").dt.epoch("ns"))


class Data:
    """One immutable snapshot of all items; requests read it without locks."""

    def __init__(self, meta, raw, eff, version):
        n = len(meta)
        self.meta, self.raw, self.eff, self.version, self.n = meta, raw, eff, version, n
        self.src = meta["source"].to_numpy()
        self.is_paper = self.src == "paper"
        self.t_ns = meta["t_ns"].fill_null(np.iinfo(np.int64).min).to_numpy()
        self.created_ns = meta["created_ns"].fill_null(0).to_numpy()
        self.modified_ns = meta["modified_ns"].fill_null(0).to_numpy()
        self.likes = meta["likes"].fill_null(0).cast(pl.Float64).to_numpy()
        self.downloads = meta["downloads"].fill_null(0).cast(pl.Float64).to_numpy()
        self.upvotes = meta["upvotes"].fill_null(0).cast(pl.Float64).to_numpy()
        self.order_t = np.argsort(-self.t_ns, kind="stable")  # newest first, computed once per refresh
        # card only = the worker saw no data/weight files (has_files False). Null (backfill, papers) = unknown.
        self.card_only = (~meta["has_files"].cast(pl.Boolean)).fill_null(False).to_numpy()
        self.scorer = meta["scorer"].fill_null("").to_numpy()
        self.scored_ns = meta.select(_ns("scored_at"))["scored_at"].fill_null(0).to_numpy()
        self.search = meta["_search"]
        self.index = {(s, i): r for r, (s, i) in enumerate(zip(self.src, meta["id"].to_list()))}
        paper_t = self.t_ns[self.is_paper]
        self.latest_paper_ns = int(paper_t.max()) if paper_t.size else None


class Store:
    """Loads items/date=*/*.parquet from the bucket mount; re-reads only new or changed files."""

    def __init__(self):
        self.files = {}  # path -> (mtime, size, polars DataFrame, raw scores)
        z = np.zeros((0, len(LABELS)), np.float32)
        self.data = Data(self._empty(), z, z, 0)
        self.status = {}
        self.status_cache = None
        self.loaded_at = None
        self.load_seconds = None
        self.error = None

    @staticmethod
    def _empty():
        types = {"tags": pl.List(pl.String), "likes": pl.Int64, "downloads": pl.Int64, "upvotes": pl.Int64,
                 "has_files": pl.Boolean}
        return pl.DataFrame(schema={c: types.get(c, pl.String) for c in META_COLS}).with_columns(
            pl.lit(None, pl.Int64).alias("t_ns"), pl.lit(None, pl.Int64).alias("created_ns"),
            pl.lit(None, pl.Int64).alias("modified_ns"), pl.lit("").alias("_search"))

    def _read(self, path):
        df = pl.read_parquet(path)
        raw = np.asarray(df["scores"].list.to_array(len(LABELS)).to_numpy(), np.float32) if len(df) else \
            np.zeros((0, len(LABELS)), np.float32)
        for c in META_COLS:
            if c not in df.columns:
                df = df.with_columns(pl.lit(None).alias(c))
        df = df.select(META_COLS).with_columns(
            pl.col("likes").cast(pl.Int64, strict=False), pl.col("downloads").cast(pl.Int64, strict=False),
            pl.col("upvotes").cast(pl.Int64, strict=False), pl.col("tags").cast(pl.List(pl.String), strict=False),
            pl.col("has_files").cast(pl.Boolean, strict=False))
        return df, raw

    def refresh(self):
        t0 = time.time()
        try:
            paths = sorted(glob.glob(os.path.join(DATA, "items", "date=*", "*.parquet")))
            changed = False
            for p in paths:
                st = os.stat(p)
                old = self.files.get(p)
                if old and old[0] == st.st_mtime and old[1] == st.st_size:
                    continue
                df, raw = self._read(p)
                self.files[p] = (st.st_mtime, st.st_size, df, raw)
                changed = True
            for p in set(self.files) - set(paths):
                del self.files[p]
                changed = True
            if changed or self.loaded_at is None:
                self._rebuild()
            self.status = self._read_status()
            self.error = None if paths else f"No parquet files under {DATA}/items (is the bucket mounted?)"
        except Exception as e:  # keep serving the last good data
            self.error = f"{type(e).__name__}: {e}"
        self.loaded_at = datetime.now(timezone.utc)
        self.load_seconds = round(time.time() - t0, 2)
        self.status_cache = None

    def _rebuild(self):
        parts = [p for p in self.files.values() if len(p[2])]
        if not parts:
            return
        meta = pl.concat([p[2] for p in parts], how="vertical_relaxed").with_row_index("_row")
        raw = np.concatenate([p[3] for p in parts])
        # One row per item: the newest score wins (the worker re-scores items when their card changes).
        meta = meta.sort("scored_at", nulls_last=False).unique(["source", "id"], keep="last", maintain_order=True)
        raw = raw[meta["_row"].to_numpy()]
        meta = meta.drop("_row").with_columns(_ns("created_at").alias("created_ns"),
                                              _ns("last_modified").alias("modified_ns"))
        meta = meta.with_columns(
            pl.coalesce("modified_ns", "created_ns").alias("t_ns"),
            pl.concat_str([pl.col("title").fill_null(""), pl.col("id").fill_null(""), pl.col("snippet").fill_null(""),
                           pl.col("tags").list.join(" ").fill_null("")], separator=" ").str.to_lowercase()
            .alias("_search"))
        eff = raw.copy()
        for b, ns in NARROW.items():
            eff[:, b] = np.max(raw[:, [b, *ns]], axis=1)
        self.data = Data(meta, raw, eff, self.data.version + 1)

    def _read_status(self):
        out = {}
        for name in ("worker_status", "backfill_stats", "cursor"):
            p = os.path.join(DATA, "state", f"{name}.json")
            if os.path.exists(p):
                try:
                    with open(p, encoding="utf-8") as f:
                        out[name] = json.load(f)
                except Exception as e:
                    out[name] = {"error": str(e)}
        return out


STAGES = {"skip_same_etag": "unchanged", "skip_unchanged_etag": "unchanged", "skip_yaml_only": "unchanged",
          "skip_no_readme": "filtered", "skip_thin_card": "filtered", "skip_private_or_disabled": "filtered",
          "superseded_in_queue": "superseded", "enqueued": "queued", "queued": "queued", "scored": "scored",
          "detected": "detected", "error": "error"}


class Events:
    """Recent worker events (last ~2 days of files): state/worker_events/date=*/*.parquet, or
    state/events/YYYY-MM-DD/*.jsonl. Columns normalised to t, id, source, event, decide_s, queue_len."""

    def __init__(self):
        self.files = {}
        self.df = pd.DataFrame(columns=["t", "id", "source", "event", "decide_s", "queue_len"])
        self.lock = threading.Lock()
        self.version = 0

    def _read(self, p):
        df = pd.read_parquet(p) if p.endswith(".parquet") else pd.read_json(p, lines=True)
        df = df.rename(columns={"ts": "t", "stage": "event"})
        for c in ("t", "id", "source", "event", "decide_s", "queue_len"):
            if c not in df:
                df[c] = None
        if "decide_ms" in df and df["decide_s"].isna().all():
            df["decide_s"] = df["decide_ms"] / 1000
        return df[["t", "id", "source", "event", "decide_s", "queue_len"]]

    def refresh(self):
        try:
            days = {(datetime.now(timezone.utc) - timedelta(days=d)).strftime("%Y-%m-%d") for d in (0, 1)}
            paths = []
            for d in days:
                paths += glob.glob(os.path.join(DATA, "state", "worker_events", f"date={d}", "*.parquet"))
                paths += glob.glob(os.path.join(DATA, "state", "events", d, "*.jsonl"))
            changed = False
            for p in paths:
                st = os.stat(p)
                old = self.files.get(p)
                if old and old[0] == st.st_mtime and old[1] == st.st_size:
                    continue
                self.files[p] = (st.st_mtime, st.st_size, self._read(p))
                changed = True
            for p in set(self.files) - set(paths):
                del self.files[p]
                changed = True
            if changed and self.files:
                df = pd.concat([f[2] for f in self.files.values()], ignore_index=True)
                df["t_dt"] = pd.to_datetime(df["t"], utc=True, errors="coerce", format="ISO8601")
                # The worker may write the same event to both logs (parquet + jsonl): keep one.
                df = df.drop_duplicates(["t", "source", "id", "event"])
                df = df.dropna(subset=["t_dt"]).sort_values("t_dt").tail(20000).reset_index(drop=True)
                with self.lock:
                    self.df = df
                    self.version += 1
        except Exception as e:
            print("events refresh failed:", e)

    def snapshot(self):
        with self.lock:
            return self.df


class Trending:
    """Hub trending scores, refreshed every 15 min (two listing calls). Not stored per row: it changes.
    Items outside the top-N get 0; papers use their Daily Papers upvotes instead."""

    def __init__(self):
        self.scores = {}  # (source, id) -> trendingScore
        self.version = 0
        self.updated_at = None
        self.error = None
        self._arr = (None, None)  # (data version, trending version) -> array aligned with STORE.data

    def refresh(self):
        try:
            from huggingface_hub import HfApi

            api, scores = HfApi(), {}
            for source, lister in (("model", api.list_models), ("dataset", api.list_datasets)):
                for x in lister(sort="trending_score", limit=TRENDING_TOP_N, expand=["trendingScore"]):
                    scores[(source, x.id)] = float(getattr(x, "trending_score", 0) or 0)
            self.scores, self.updated_at, self.error = scores, datetime.now(timezone.utc), None
            self.version += 1
        except Exception as e:  # keep the last good map
            self.error = f"{type(e).__name__}: {e}"
            print("trending refresh failed:", self.error, flush=True)

    def array(self, d):
        """Trending key per row of `d`: Hub trendingScore, or upvotes for papers."""
        key = (d.version, self.version)
        if self._arr[0] != key:
            hub = np.fromiter((self.scores.get(k, 0.0) for k in zip(d.src, d.meta["id"].to_list())), np.float64, d.n)
            self._arr = (key, np.where(d.is_paper, d.upvotes, hub))
        return self._arr[1]


STORE = Store()
EVENTS = Events()
TRENDING = Trending()


def refresher(obj, every):
    while True:
        time.sleep(every)
        obj.refresh()


# ---------------------------------------------------------------- in-Space ingest worker
# The worker (firehose_worker.py) runs GLiNER2.5-Decide on CPU and appends scores to the bucket mounted at DATA.
# It runs as a child process at lower CPU priority so page requests stay responsive. RUN_WORKER=0 disables it.
RUN_WORKER = os.environ.get("RUN_WORKER", "1") == "1"
# --flush-secs 60 (worker default 300): scored rows land in items/ within a minute, so Live shows them sooner.
WORKER_ARGS = ["--out", DATA, "--model-cap", "1500", "--inherit", "quantized", "--sweep-secs", "1800",
               "--sweep-hours", "6", "--canary-secs", "600", "--flush-secs", "60"]


def wait_for_previous_worker(stale_s=150, max_wait_s=900):
    """On a redeploy the old container keeps running until the new one is up. Its worker writes
    state/worker_status.json every ~60 s; wait until that goes stale so two workers never write at once."""
    path = os.path.join(DATA, "state", "worker_status.json")
    t0 = time.time()
    while time.time() - t0 < max_wait_s:
        try:
            with open(path, encoding="utf-8") as f:
                upd = pd.to_datetime(json.load(f).get("updated_at"), utc=True)
            age = (datetime.now(timezone.utc) - upd).total_seconds()
        except Exception:
            return
        if age > stale_s:
            return
        print(f"worker: previous worker active {age:.0f} s ago; waiting before starting", flush=True)
        time.sleep(20)
    print("worker: previous worker still reporting after 15 min; starting anyway", flush=True)


def supervise_worker():
    import subprocess
    import sys

    while not os.path.isdir(os.path.join(DATA, "state")):
        print("worker: waiting for the bucket mount at", DATA, flush=True)
        time.sleep(10)
    wait_for_previous_worker()
    backoff = 30
    while True:
        started = time.time()
        proc = subprocess.Popen([sys.executable, os.path.join(HERE, "firehose_worker.py"), *WORKER_ARGS],
                                preexec_fn=lambda: os.nice(10))
        print(f"worker: started pid {proc.pid}", flush=True)
        rc = proc.wait()
        print("worker exited with", rc, flush=True)
        backoff = 30 if time.time() - started > 600 else min(600, backoff * 2)
        time.sleep(backoff)


_WORKER_LOCK = None  # held for the life of the process that supervises the worker


def start_worker_once():
    """Single-writer rule: at most one ingest worker per container. Granian runs one process (--workers 1);
    the file lock also guards against a second app process (e.g. a reload) starting another worker."""
    global _WORKER_LOCK
    import fcntl

    f = open(os.environ.get("WORKER_LOCK", "/tmp/topic-feed-worker.lock"), "w")
    try:
        fcntl.flock(f, fcntl.LOCK_EX | fcntl.LOCK_NB)
    except OSError:
        print(f"worker: another process (pid file {f.name}) supervises the worker; not starting one", flush=True)
        f.close()
        return False
    f.write(str(os.getpid()))
    f.flush()
    _WORKER_LOCK = f
    print(f"worker: supervisor started in pid {os.getpid()}", flush=True)
    threading.Thread(target=supervise_worker, daemon=True).start()
    return True


# ---------------------------------------------------------------- helpers
def thr_vector(thresholds):
    th = DEFAULT_THR.copy()
    for t, v in (thresholds or {}).items():
        if t in L and v is not None:
            try:
                th[L[t]] = float(v)
            except (TypeError, ValueError):
                pass
    return th


def topic_idx(topics):
    return [L[t] for t in (topics or []) if t in L]


def iso_ns(v):
    return None if v is None or v <= 0 else datetime.fromtimestamp(v / NS, timezone.utc).isoformat()


def items_payload(d, rows):
    """The fields the page uses, for one page of rows (vectorised; scores as a list in catalogue label order)."""
    if len(rows) == 0:
        return []
    page = d.meta[rows].select(PAGE_COLS).to_dicts()
    window_start = (datetime.now(timezone.utc) - NEW_WITHIN).timestamp() * NS
    created, modified, t = d.created_ns[rows], d.modified_ns[rows], d.t_ns[rows]
    has_both = (created > 0) & (modified > 0)
    new = ~has_both | ((modified - created) <= NEW_WITHIN.total_seconds() * NS) | (created >= window_start)
    new |= d.is_paper[rows]
    date_only = d.is_paper[rows] & (t % (86400 * NS) == 0)
    scores = np.round(d.raw[rows], 3)  # raw: the page applies max(broad, narrow) and marks where it came from
    trend = TRENDING.array(d)[rows]
    for i, it in enumerate(page):
        it["snip"] = it.pop("snippet")
        it["tags"] = (it["tags"] or [])[:8]
        it["t"] = iso_ns(int(t[i]))
        it["created"] = iso_ns(int(created[i]))
        it["modified"] = iso_ns(int(modified[i]))
        it["date_only"] = bool(date_only[i])
        it["badge"] = "new" if new[i] else "updated"
        it["s"] = scores[i]
        if d.card_only[rows[i]]:
            it["card_only"] = True
        if trend[i] > 0 and not d.is_paper[rows[i]]:
            it["trend"] = int(trend[i])
    return page


def slim_worker_status(ws):
    """Only the worker fields the page shows."""
    if not ws or "error" in ws:
        return ws
    return {"updated_at": ws.get("updated_at"), "state": ws.get("state"), "queue": {"total": (ws.get("queue") or {}).get("total"),
                                                                              "inflight": (ws.get("queue") or {}).get("inflight")},
            "sources": {s: {"last_hour": {"scored": (v.get("last_hour") or {}).get("scored")},
                            "latency_modified_to_scored_s": v.get("latency_modified_to_scored_s")}
                        for s, v in (ws.get("sources") or {}).items()}}


def _rss_mb():
    """Resident memory of this app process (Linux), MB."""
    try:
        with open("/proc/self/status") as f:
            for line in f:
                if line.startswith("VmRSS:"):
                    return round(int(line.split()[1]) / 1024)
    except OSError:
        return None


def card_only_share(d, now_ns, hours=1):
    """Share of card-only items among worker-scored Hub items in the last `hours` (by scored_at)."""
    recent = (d.scored_ns >= now_ns - hours * 3600 * NS) & ~d.is_paper & np.isin(d.scorer, ["worker", "inherited"])
    known = recent & (d.meta["has_files"].is_not_null().to_numpy())
    n, c = int(known.sum()), int((known & d.card_only).sum())
    by = {s: {"known": int((known & (d.src == s)).sum()), "card_only": int((known & d.card_only & (d.src == s)).sum())}
          for s in ("model", "dataset")}
    return {"hours": hours, "known": n, "card_only": c, "share": round(c / n, 3) if n else None, "by_source": by}


def status_payload():
    """Cached per data refresh (the refresher clears it)."""
    if STORE.status_cache is not None:
        return STORE.status_cache
    d = STORE.data
    now_ns = int(datetime.now(timezone.utc).timestamp() * NS)
    valid = d.t_ns > 0
    bf = (STORE.status.get("backfill_stats") or {}).get("fetch") or {}
    out = {"rows": d.n, "by_source": {s: int((d.src == s).sum()) for s in SOURCES},
           "newest": iso_ns(int(d.t_ns[valid].max())) if valid.any() else None,
           "last_24h": int((d.t_ns >= now_ns - 86400 * NS).sum()), "last_1h": int((d.t_ns >= now_ns - 3600 * NS).sum()),
           "worker_status": slim_worker_status(STORE.status.get("worker_status")),
           "backfill_stats": {"fetch": {s: {"listed": v.get("listed"), "kept": v.get("kept")} for s, v in bf.items()
                                        if isinstance(v, dict)}},
           "loaded_at": STORE.loaded_at.isoformat() if STORE.loaded_at else None, "load_seconds": STORE.load_seconds,
           "files": len(STORE.files), "error": STORE.error, "catalogue_version": CAT["catalogue_version"],
           "latest_paper_date": iso_ns(d.latest_paper_ns), "app_rss_mb": _rss_mb(),
           "card_only_last_hour": card_only_share(d, now_ns),
           "trending": {"items": len(TRENDING.scores), "updated_at": TRENDING.updated_at.isoformat() if TRENDING.updated_at
                        else None, "error": TRENDING.error}}
    STORE.status_cache = out
    return out


# ---------------------------------------------------------------- endpoints (pure functions)
def catalogue() -> dict:
    """Topic catalogue: labels, notes, theme groups, broad/narrow pairs, default thresholds."""
    return {k: CAT[k] for k in ("catalogue_version", "labels", "notes", "groups", "pairs", "thresholds")} | {
        "thresholds_v1": CAT.get("thresholds_v1"), "custom_topics_enabled": CUSTOM_TOPICS_ENABLED}


def status() -> dict:
    """Data status: row counts, newest item, worker status (if any)."""
    return status_payload()


def live(topics, thresholds, limit=50, offset=0, since="", all_items=False, prof=None, card_only=False) -> dict:
    """Newest items first (by last_modified) where any of `topics` passes its threshold (all_items: every item).
    Also returns the last 200 arrivals across all topics with their lane (for the replay)."""
    tick = _Tick(prof)
    d = STORE.data
    th, idx = thr_vector(thresholds), topic_idx(topics)
    if all_items:
        passing = np.ones(d.n, bool)
    elif idx:
        passing = (d.eff[:, idx] >= th[idx]).any(axis=1)
    else:
        passing = np.zeros(d.n, bool)
    if not card_only:
        passing = passing & ~d.card_only
    tick("thresholds+filter")
    rows = d.order_t[passing[d.order_t]]
    tick("sort")
    new_since = None
    if since:
        cut = pd.to_datetime(since, utc=True, errors="coerce")
        if not pd.isna(cut):
            new_since = int((d.t_ns[rows] > cut.value).sum())
    top = d.order_t[:200]
    if idx:
        margins = d.eff[np.ix_(top, idx)] - th[idx]
        best = margins.argmax(axis=1)
        ok = margins[np.arange(len(top)), best] >= 0
        lanes = [topics_in(idx, j) if k else None for j, k in zip(best, ok)]
    else:
        lanes = [None] * len(top)
    arrivals = [{"t": iso_ns(int(t)), "source": s, "lane": ln} for t, s, ln in zip(d.t_ns[top], d.src[top], lanes)]
    tick("arrivals")
    page = items_payload(d, rows[offset:offset + limit])
    tick("pagination+payload")
    out = {"items": page, "total": int(len(rows)), "new_since": new_since, "arrivals": arrivals,
           "status": status_payload()}
    tick("status")
    return out


def topics_in(idx, j):
    return LABELS[idx[j]]


SORTS = {"score", "newest", "likes", "downloads", "trending"}


def browse(topics, thresholds, days=7, sources=None, q="", sort="score", limit=30, offset=0, match="any",
           prof=None, card_only=False) -> dict:
    """Items in one or more topics. days: 1, 3, 7 or 0 (all). sources: model/dataset/paper. q: text search over
    title, id, snippet and tags. sort: score, newest, likes, downloads (papers: upvotes). match: 'any' or 'all'."""
    tick = _Tick(prof)
    d = STORE.data
    th, idx = thr_vector(thresholds), topic_idx(topics)
    mask = np.ones(d.n, bool)
    if idx:
        hits = d.eff[:, idx] >= th[idx]
        mask &= hits.all(axis=1) if match == "all" else hits.any(axis=1)
    tick("thresholds")
    if days and days > 0:
        # Hub items: the last `days` x 24 h. Papers have date-only times: their window ends at the latest paper day.
        now_ns = int(datetime.now(timezone.utc).timestamp() * NS)
        ok = d.t_ns >= now_ns - int(days) * 86400 * NS
        if d.latest_paper_ns is not None:
            day0 = d.latest_paper_ns - d.latest_paper_ns % (86400 * NS) - (int(days) - 1) * 86400 * NS
            ok = np.where(d.is_paper, d.t_ns >= day0, ok)
        mask &= ok
    if sources:
        mask &= np.isin(d.src, list(sources))
    if not card_only:
        mask &= ~d.card_only
    if q and q.strip():
        expr = pl.lit(True)
        for word in q.lower().split():
            expr = expr & pl.col("_search").str.contains(word, literal=True)
        mask &= d.meta.select(expr.alias("m"))["m"].to_numpy()
    tick("filter")
    rows = np.flatnonzero(mask)
    sort = sort if sort in SORTS else "score"
    if sort == "score" and idx:
        key = -d.eff[np.ix_(rows, idx)].max(axis=1)
    elif sort == "likes":
        key = -d.likes[rows]
    elif sort == "trending":  # Hub trendingScore (top 1000, refreshed every 15 min); papers: upvotes
        key = -TRENDING.array(d)[rows]
    elif sort == "downloads":  # papers have no downloads: rank them by upvotes
        key = -np.where(d.is_paper[rows], d.upvotes[rows], d.downloads[rows])
    else:
        key = -d.t_ns[rows]
    rows = rows[np.argsort(key, kind="stable")]
    tick("sort")
    # Summary: counts per source, and which other topics the matched items are about (catalogue defaults).
    co = (d.eff[rows] >= DEFAULT_THR).sum(axis=0)
    co_topics = [{"topic": LABELS[j], "n": int(co[j])} for j in np.argsort(-co) if co[j] and j not in idx][:10]
    per_topic = {LABELS[j]: int((d.eff[rows, j] >= th[j]).sum()) for j in idx}
    srcs = d.src[rows]
    summary = {"total": int(len(rows)), "by_source": {s: int((srcs == s).sum()) for s in SOURCES if (srcs == s).any()},
               "co_topics": co_topics, "per_topic": per_topic}
    tick("summary")
    page = items_payload(d, rows[offset:offset + limit])
    tick("pagination+payload")
    return {"items": page, "total": int(len(rows)), "summary": summary, "status": status_payload()}


def events(topics, thresholds, since="", limit=400) -> dict:
    """Recent worker events, newest last, with each scored item's lane among `topics` and its top topics."""
    ev = EVENTS.snapshot()
    d = STORE.data
    th, idx = thr_vector(thresholds), topic_idx(topics)
    now = datetime.now(timezone.utc)
    last = ev["t_dt"].max() if len(ev) else None
    recent = ev[ev["t_dt"] >= now - timedelta(minutes=10)] if len(ev) else ev
    counts = {}
    for e in recent["event"].tolist():
        k = STAGES.get(e, e)
        counts[k] = counts.get(k, 0) + 1
    sel = ev
    if since:
        cut = pd.to_datetime(since, utc=True, errors="coerce")
        if not pd.isna(cut):
            sel = ev[ev["t_dt"] >= cut]  # same-second events: the client de-duplicates
    out = []
    for e in sel.tail(int(limit)).itertuples():
        stage = STAGES.get(e.event, e.event)
        o = {"t": e.t, "id": e.id, "source": e.source, "stage": stage, "event": e.event}
        r = d.index.get((e.source, e.id))
        if stage == "scored" and r is not None:
            top = np.argsort(-d.eff[r])[:3]
            o["top"] = [LABELS[j] for j in top if d.eff[r, j] >= DEFAULT_THR[j]]
            if idx:
                margins = d.eff[r, idx] - th[idx]
                j = int(np.argmax(margins))
                o["lane"] = LABELS[idx[j]] if margins[j] >= 0 else None
        out.append(o)
    return {"events": out, "last_event_at": last.isoformat() if last is not None else None,
            "active": bool(last is not None and last >= now - ACTIVE_WITHIN), "last_10min": counts,
            "worker_status": slim_worker_status(STORE.status.get("worker_status")), "rows": d.n}


class _Tick:
    """Optional per-stage timer: fills `prof` (dict) with ms per stage."""

    def __init__(self, prof):
        self.prof, self.t = prof, time.perf_counter()

    def __call__(self, stage):
        if self.prof is not None:
            now = time.perf_counter()
            self.prof[stage] = round((now - self.t) * 1000, 2)
            self.t = now


# ---------------------------------------------------------------- app
def dumps(obj):
    return orjson.dumps(obj, option=orjson.OPT_SERIALIZE_NUMPY)


class Cache:
    """Serialised responses for CACHE_TTL_S, keyed by route + normalised params + data versions."""

    def __init__(self, maxsize=512):
        self.d, self.maxsize, self.lock = {}, maxsize, threading.Lock()

    def get_or_make(self, key, make):
        key = (key, STORE.data.version, EVENTS.version, TRENDING.version)
        now = time.time()
        with self.lock:
            hit = self.d.get(key)
            if hit and now - hit[0] < CACHE_TTL_S:
                return hit[1], True
        body = dumps(make())
        with self.lock:
            if len(self.d) >= self.maxsize:
                for k in sorted(self.d, key=lambda k: self.d[k][0])[: self.maxsize // 4]:
                    del self.d[k]
            self.d[key] = (now, body)
        return body, False


CACHE = Cache()


def cached_json(route, params, make, max_age=0):
    key = route + "?" + json.dumps(params, sort_keys=True, default=str)
    body, hit = CACHE.get_or_make(key, make)
    headers = {"X-Cache": "hit" if hit else "miss",
               "Cache-Control": f"public, max-age={max_age}" if max_age else "no-cache"}
    return Response(body, media_type="application/json", headers=headers)


@asynccontextmanager
async def lifespan(_app):
    STORE.refresh()
    EVENTS.refresh()
    threading.Thread(target=refresher, args=(STORE, REFRESH_S), daemon=True).start()
    threading.Thread(target=refresher, args=(EVENTS, EVENTS_REFRESH_S), daemon=True).start()
    threading.Thread(target=lambda: (TRENDING.refresh(), refresher(TRENDING, TRENDING_REFRESH_S)), daemon=True).start()
    if RUN_WORKER:
        start_worker_once()
    yield


app = FastAPI(title="Topic Feed Lab", lifespan=lifespan, docs_url="/api/docs", openapi_url="/api/openapi.json")
app.add_middleware(GZipMiddleware, minimum_size=1000)


@app.middleware("http")
async def timing(request, call_next):
    t0 = time.perf_counter()
    resp = await call_next(request)
    resp.headers["Server-Timing"] = f"app;dur={(time.perf_counter() - t0) * 1000:.1f}"
    return resp


def _thresholds(s):
    """thresholds query param: a JSON object {topic: value}."""
    try:
        d = json.loads(s) if s else {}
        return d if isinstance(d, dict) else {}
    except ValueError:
        return {}


@app.get("/favicon.ico")
def favicon():
    return Response(status_code=204)


@app.get("/", response_class=HTMLResponse)
async def homepage():
    with open(os.path.join(HERE, "index.html"), encoding="utf-8") as f:
        return f.read()


# ---------------------------------------------------------------- JSON API
# Lists (topics, sources) are repeated query params: ?topics=robotics&topics=video. thresholds is a JSON object.
# Item scores come as `s`: a list of 48 raw scores in catalogue label order (the page applies max(broad, narrow)).
@app.get("/api/catalogue")
def api_catalogue():
    return cached_json("catalogue", {}, catalogue, max_age=3600)


NO_STORE = {"Cache-Control": "no-store"}


@app.get("/api/status")
def api_status():
    return Response(dumps(status()), media_type="application/json", headers=NO_STORE)


@app.get("/api/live")
def api_live(topics: list[str] = Query(default=[]), thresholds: str = "", limit: int = 50, offset: int = 0,
             since: str = "", all_items: bool = False, card_only: bool = False):
    th = _thresholds(thresholds)
    p = {"topics": topics, "th": th, "limit": limit, "offset": offset, "since": since, "all": all_items, "co": card_only}
    return cached_json("live", p, lambda: live(topics, th, limit, offset, since, all_items, card_only=card_only))


@app.get("/api/events")
def api_events(topics: list[str] = Query(default=[]), thresholds: str = "", since: str = "", limit: int = 400):
    return Response(dumps(events(topics, _thresholds(thresholds), since, limit)), media_type="application/json",
                    headers=NO_STORE)


@app.get("/api/browse")
def api_browse(topics: list[str] = Query(default=[]), thresholds: str = "", days: int = 7,
               sources: list[str] = Query(default=[]), q: str = "", sort: str = "score", limit: int = 30,
               offset: int = 0, match: str = "any", card_only: bool = False):
    th = _thresholds(thresholds)
    p = {"topics": topics, "th": th, "days": days, "sources": sorted(sources), "q": q.strip().lower(), "sort": sort,
         "limit": limit, "offset": offset, "match": match, "co": card_only}
    return cached_json("browse", p, lambda: browse(topics, th, days, sources or None, q, sort, limit, offset, match,
                                                   card_only=card_only))