Create usage.py

usage.py

@@ -0,0 +1,275 @@
+"""
+usage.py — ResearchBee usage counting
+
+Counts only. There is no query-text path through this module and no
+column to put one in: the manuscript tab receives unpublished abstracts,
+and the cheapest guarantee that those are never persisted is to have
+nowhere to persist them.
+
+Design notes
+------------
+1.  Counting happens in memory and costs a dict update. Nothing touches
+    the network on the request path.
+
+2.  Deltas are flushed to the Cloudflare Worker every FLUSH_SECONDS.
+    A failed flush is NOT lost: the counts are folded back into the
+    pending dict and go out with the next attempt. Because we send
+    accumulated deltas rather than one row per request, an outage costs
+    a delay in the dashboard, not data. This is the main reason the
+    aggregate design is worth having — LibBee's per-row logging had to
+    grow a 200-item retry buffer to get the same property, and still
+    drops rows past that bound.
+
+3.  HF Spaces egress to workers.dev is known to fail intermittently
+    (LibBee v3.9.2 moved its logging into the browser for exactly this
+    reason). Retry-with-accumulation handles that without needing a
+    client-side path, so nothing has to change in the LibGuides fragment.
+
+4.  Counters do NOT survive a container restart if a flush has not
+    happened yet — at most FLUSH_SECONDS of activity is lost on a hard
+    restart. The 6-hour canary keeps the Space warm, so restarts should
+    be rare.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+import os
+from collections import deque
+from datetime import datetime
+from typing import Dict, List, Tuple
+
+logger = logging.getLogger(__name__)
+
+WORKER_URL = (os.environ.get("CLOUDFLARE_WORKER_URL") or "").strip().strip('"').strip("'").rstrip("/")
+WORKER_TOKEN = (os.environ.get("CLOUDFLARE_WORKER_TOKEN") or "").strip()
+
+FLUSH_SECONDS = 60
+MAX_ATTEMPTS_LOG = 15
+
+# Endpoint → surface. Adding an endpoint here is the only step needed to
+# start counting it; the Worker validates the name and rejects unknowns,
+# so a typo surfaces as a visible 400 rather than a phantom column.
+SURFACE_BY_PATH: Dict[str, str] = {
+    "/api/browse-subject":  "journal_subject",
+    "/api/analyze-journal": "journal_manuscript",
+    "/api/check-license":   "licence_deposit",
+    "/api/find-repository": "data_repository",
+    "/api/cover-letter":    "cover_letter",
+    "/api/chat":            "chat",
+}
+
+# Key: (surface, outcome, gap_code)
+_Key = Tuple[str, str, str]
+
+
+class UsageCounters:
+    def __init__(self) -> None:
+        self._pending: Dict[_Key, int] = {}
+        self._lifetime: Dict[_Key, int] = {}
+        self._lock = asyncio.Lock()
+        self._attempts: deque = deque(maxlen=MAX_ATTEMPTS_LOG)
+        self._task: asyncio.Task | None = None
+        self._started_at = datetime.utcnow()
+
+    # ── Recording (sync, no I/O) ──────────────────────────────────────
+
+    def record(self, surface: str, outcome: str = "ok", gap_code: str = "") -> None:
+        """Count one event. Never raises — instrumentation must not be
+        able to fail a user request."""
+        try:
+            if surface not in SURFACE_BY_PATH.values():
+                return
+            if outcome not in ("ok", "miss", "error"):
+                outcome = "error"
+            key = (surface, outcome, (gap_code or "")[:40])
+            self._pending[key] = self._pending.get(key, 0) + 1
+            self._lifetime[key] = self._lifetime.get(key, 0) + 1
+        except Exception:
+            pass
+
+    def record_gap(self, surface: str, gap_code: str) -> None:
+        """A request that completed but could not be answered from the
+        available sources — a SCImago subject with no matches, an
+        OA.Works 404, a DOI that resolves to nothing. Recorded as its own
+        outcome so 'served a page' and 'answered the question' stay
+        distinguishable in the dashboard."""
+        self.record(surface, outcome="miss", gap_code=gap_code)
+
+    # ── Local snapshot (works with no Worker configured) ──────────────
+
+    def snapshot(self) -> dict:
+        by_surface: Dict[str, Dict[str, int]] = {}
+        gaps: Dict[str, int] = {}
+        for (surface, outcome, gap), n in self._lifetime.items():
+            s = by_surface.setdefault(
+                surface, {"total": 0, "ok": 0, "miss": 0, "error": 0}
+            )
+            s["total"] += n
+            s[outcome] = s.get(outcome, 0) + n
+            if gap:
+                gaps[gap] = gaps.get(gap, 0) + n
+        return {
+            "since": self._started_at.strftime("%Y-%m-%d %H:%M:%S") + " UTC",
+            "note": "Counts since this container last started. "
+                    "Durable totals come from the Worker.",
+            "by_surface": by_surface,
+            "gaps": dict(sorted(gaps.items(), key=lambda kv: -kv[1])),
+            "pending_deltas": len(self._pending),
+            "worker_configured": bool(WORKER_URL),
+            "recent_flushes": list(self._attempts)[::-1],
+        }
+
+    # ── Flushing ──────────────────────────────────────────────────────
+
+    def _record_attempt(self, outcome: str, detail: str = "") -> None:
+        self._attempts.append({
+            "at": datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S"),
+            "outcome": outcome,
+            "detail": detail[:200],
+        })
+
+    async def flush(self) -> bool:
+        """Send accumulated deltas. On failure the deltas are folded back
+        in and retried on the next pass, so nothing is dropped."""
+        if not WORKER_URL or not self._pending:
+            return True
+
+        async with self._lock:
+            batch = self._pending
+            self._pending = {}
+
+        deltas: List[dict] = [
+            {"surface": s, "outcome": o, "gap_code": g, "n": n}
+            for (s, o, g), n in batch.items()
+        ]
+
+        try:
+            import httpx
+            headers = {"Authorization": f"Bearer {WORKER_TOKEN}"} if WORKER_TOKEN else {}
+            async with httpx.AsyncClient(timeout=10) as client:
+                r = await client.post(
+                    f"{WORKER_URL}/rb/log",
+                    json={"deltas": deltas},
+                    headers=headers,
+                )
+            body = {}
+            try:
+                body = r.json()
+            except Exception:
+                pass
+
+            # The Worker answers 200 with logged=false when the D1 write
+            # itself failed, so the status code alone cannot tell a stored
+            # batch from a dropped one. Check the verdict, not the code.
+            if r.status_code < 400 and body.get("logged"):
+                self._record_attempt(
+                    "ok", f"{body.get('accepted', len(deltas))} delta(s) stored"
+                )
+                return True
+
+            self._requeue(batch)
+            if r.status_code == 401:
+                self._record_attempt(
+                    "rejected_401",
+                    "CLOUDFLARE_WORKER_TOKEN does not match the Worker's ANALYTICS_TOKEN",
+                )
+            else:
+                self._record_attempt(
+                    "refused",
+                    f"HTTP {r.status_code} {body.get('warning', '')}",
+                )
+            return False
+
+        except Exception as exc:
+            self._requeue(batch)
+            self._record_attempt("unreachable", f"{type(exc).__name__}: {exc}")
+            return False
+
+    def _requeue(self, batch: Dict[_Key, int]) -> None:
+        for key, n in batch.items():
+            self._pending[key] = self._pending.get(key, 0) + n
+
+    async def _loop(self) -> None:
+        while True:
+            await asyncio.sleep(FLUSH_SECONDS)
+            try:
+                await self.flush()
+            except Exception as exc:
+                logger.warning("usage flush loop error: %s", exc)
+
+    def start(self) -> None:
+        """Start the background flusher. Called once from app startup."""
+        if self._task is not None:
+            return
+        try:
+            loop = asyncio.get_running_loop()
+        except RuntimeError:
+            return
+        # A strong reference is held on the instance: asyncio keeps only
+        # weak references to tasks, and a dropped task can be collected
+        # mid-flight, silently cancelling the flush.
+        self._task = loop.create_task(self._loop())
+        logger.info("Usage flusher started (%ss interval, worker=%s)",
+                    FLUSH_SECONDS, bool(WORKER_URL))
+
+    async def stop(self) -> None:
+        """Final flush on shutdown so the last minute of activity lands."""
+        if self._task:
+            self._task.cancel()
+            self._task = None
+        await self.flush()
+
+
+usage = UsageCounters()
+
+
+# ── FastAPI middleware ────────────────────────────────────────────────
+
+def install(app) -> None:
+    """Wire counting into the app. No frontend change is required: the
+    tab a person used is implied by the endpoint their browser called,
+    which is why this survives the move to LibGuides and any future
+    frontend rewrite."""
+    from starlette.requests import Request
+
+    @app.middleware("http")
+    async def _count(request: Request, call_next):
+        surface = SURFACE_BY_PATH.get(request.url.path)
+        # Canary traffic exists to keep the Space awake and must never
+        # appear in usage figures — the same exclusion LibBee applies.
+        counted = surface and request.headers.get("X-ResearchBee-Canary") != "1"
+
+        try:
+            response = await call_next(request)
+        except Exception:
+            # An unhandled exception propagates out of call_next, so
+            # anything written after the await would never run. Without
+            # this branch the most important failures — the ones that
+            # crash a handler — would be the only ones missing from the
+            # dashboard.
+            if counted:
+                usage.record(surface, "error", "unhandled_exception")
+            raise
+
+        try:
+            if counted:
+                if response.status_code < 400:
+                    # A handler that recorded its own miss has already
+                    # counted this request; the flag stops it counting twice.
+                    if not getattr(request.state, "usage_recorded", False):
+                        usage.record(surface, "ok")
+                else:
+                    usage.record(surface, "error")
+        except Exception:
+            pass
+        return response
+
+    @app.on_event("startup")
+    async def _start_usage():
+        usage.start()
+
+    @app.on_event("shutdown")
+    async def _stop_usage():
+        await usage.stop()