thermograph/metrics.py

"""Traffic counters for the ops dashboard.

Cheap, best-effort counters for **inbound** HTTP requests (bucketed by category) and
**outbound** calls to external data sources (bucketed by source). Everything here is
since-start; durable daily history lives in the audit/error JSONL logs (see
``audit.py``). Recording never raises into the request path.

Read over the gated ``GET {BASE}/api/v2/metrics`` route (see ``app.py``) and rendered
by ``scripts/dashboard.py`` over SSH.

**Multi-worker.** Under a single uvicorn worker the counters live in process memory.
When the app runs several workers (``--workers N``), per-process memory would split the
tally across workers, so the dashboard — which polls one random worker — would see only
a fraction of the traffic. Set ``THERMOGRAPH_METRICS_DB`` to a file path and every worker
shares one SQLite store, so the dashboard reports the whole picture. Unset (dev, tests,
single worker) keeps the zero-dependency in-memory store.
"""
import os
import sqlite3
import threading
import time
import urllib.parse

# The ``phase=`` tag passed to ``climate._request`` maps 1:1 onto an external source.
# Keep this in sync with the phase names used at each call site in ``climate.py``.
_PHASE_SOURCE = {
    "history_fetch": "open-meteo-archive",
    "history_topup": "open-meteo-archive",
    "recent_forecast_fetch": "open-meteo-forecast",
    "geocode": "open-meteo-geocoding",
    "history_nasa": "nasa-power",
    "forecast_metno": "met-norway",
    "reverse_geocode": "nominatim",
}

# Every source we know about, so the dashboard shows a stable set of rows even before
# any traffic has hit a given source.
SOURCES = sorted(set(_PHASE_SOURCE.values()))

_OUTCOMES = ("ok", "retry", "error", "rate_limited")
_STATUS_BUCKETS = ("2xx", "3xx", "4xx", "5xx", "other")

# Rolling short-window view: per-minute buckets summed over the last WINDOW minutes on
# snapshot(). Cheap and bounded — minutes are pruned as they roll out of the window.
WINDOW_MINUTES = 10


def source_for_phase(phase: str) -> str:
    """Map a ``climate._request`` phase tag to an external-source label."""
    return _PHASE_SOURCE.get(phase, "other")


def _status_bucket(status) -> str:
    try:
        code = int(status)
    except (TypeError, ValueError):
        return "other"
    if 200 <= code < 300:
        return "2xx"
    if 300 <= code < 400:
        return "3xx"
    if 400 <= code < 500:
        return "4xx"
    if 500 <= code < 600:
        return "5xx"
    return "other"


def classify_inbound(path: str, base: str = "") -> str:
    """Bucket a request path into a dashboard category.

    ``path`` is ``request.url.path`` (still carrying the ``BASE`` prefix); ``base`` is
    the app's mount prefix (e.g. ``/thermograph`` or ``""``).
    """
    p = path or "/"
    # The dashboard polls the metrics endpoint; never count it as traffic, whatever base
    # prefix it arrives under — e.g. a `/thermograph/api/v2/metrics` probe against a
    # root-served prod app would otherwise land in "other" and show as an inbound error.
    if p.rstrip("/").endswith("/api/v2/metrics"):
        return "metrics"
    if base and p.startswith(base):
        p = p[len(base):] or "/"
    if p.startswith("/api/"):
        rest = p[len("/api/"):]
        if rest.startswith(("v1/", "v2/")):
            rest = rest[3:]
        seg = rest.split("/", 1)[0].split("?", 1)[0]
        if seg in ("auth", "users"):
            return "auth"
        if seg in ("subscriptions", "notifications", "push"):
            return "accounts"
        if seg == "metrics":
            return "metrics"
        # The event beacon reports traffic; it is not itself traffic. Its own
        # category keeps record_inbound from double-counting every interaction.
        if seg == "event":
            return "event"
        return f"api:{seg}" if seg else "api:other"
    if p.endswith((".js", ".css", ".html", ".webmanifest", ".png", ".svg", ".ico",
                   ".json", ".woff", ".woff2", ".map", ".txt", ".xml")):
        # robots.txt / sitemap.xml are crawler SEO endpoints, not page views.
        if p in ("/robots.txt", "/sitemap.xml"):
            return "seo"
        return "static"
    if p in ("", "/", "/calendar", "/day", "/compare", "/legend", "/alerts", "/privacy"):
        return "page"
    if p == "/about" or p.startswith("/climate") or p.startswith("/glossary"):
        return "page"
    return "other"


# --- product events ----------------------------------------------------------
# Distinct from inbound traffic: these count *intent* (tapped "use my location",
# submitted the digest form) rather than requests. Keyed by (event, referrer
# domain, UTC day) so a launch post's traffic can be told apart from search.
#
# Everything here is aggregate — no cookies, no per-visitor id, no full URLs.

EVENTS = frozenset({
    "home.locate", "home.digest_signup", "home.records_click", "home.share",
})
# home.nav_{surface}: only these surfaces, so the key space stays bounded.
NAV_SURFACES = frozenset({
    "calendar", "compare", "day", "alerts", "city", "cities", "records",
})
# Anything unrecognized collapses here, with no referrer dimension — visible
# enough to notice abuse, bounded enough to be harmless.
EVENT_OTHER = "event:other"

EVENT_DAYS = 7          # how many days of event history to keep
MAX_REFERRERS = 200     # distinct referrer domains tracked before overflow
REFERRER_OTHER = "other"

_REF_OK = set("abcdefghijklmnopqrstuvwxyz0123456789.-")


def normalize_event(name: str) -> str:
    """An allowlisted event name, or EVENT_OTHER. The allowlist is what keeps an
    unauthenticated endpoint from being turned into unbounded storage."""
    name = (name or "").strip()
    if name in EVENTS:
        return name
    surface = name.partition("home.nav_")[2] if name.startswith("home.nav_") else ""
    if surface and surface in NAV_SURFACES:
        return name
    return EVENT_OTHER


def normalize_referrer(referer: "str | None", host: "str | None" = None) -> str:
    """The bare registrable-ish domain of a Referer header — never the full URL,
    which can carry a search query or a private path. 'direct' when absent,
    'self' for our own pages."""
    if not referer:
        return "direct"
    try:
        netloc = urllib.parse.urlsplit(referer).netloc
    except ValueError:
        return REFERRER_OTHER
    domain = (netloc.rpartition("@")[2].partition(":")[0] or "").lower()
    if domain.startswith("www."):
        domain = domain[4:]
    if host and domain == (host.partition(":")[0] or "").lower().removeprefix("www."):
        return "self"
    domain = "".join(ch for ch in domain if ch in _REF_OK)[:64]
    return domain or REFERRER_OTHER


def _utc_day() -> str:
    return time.strftime("%Y-%m-%d", time.gmtime())


def _recent_days() -> "set[str]":
    """The EVENT_DAYS UTC days still worth keeping."""
    now = time.time()
    return {time.strftime("%Y-%m-%d", time.gmtime(now - d * 86400))
            for d in range(EVENT_DAYS)}


def _nest_events(rows) -> dict:
    """Flat (event, referrer, day, count) rows -> {event: {referrer: {day: n}}}."""
    out: "dict[str, dict[str, dict[str, int]]]" = {}
    for key, count in rows:
        event, referrer, day = key
        out.setdefault(event, {}).setdefault(referrer, {})[day] = count
    return out


def _now_minute() -> int:
    """Current epoch-minute. Uses ``time.time()`` so tests can monkeypatch the clock."""
    return int(time.time() // 60)


def _empty_inbound_row() -> "dict[str, int]":
    return {"total": 0, **{b: 0 for b in _STATUS_BUCKETS}}


class _MemStore:
    """In-process counters. One per worker — fine for a single-worker deployment."""

    def __init__(self) -> None:
        self._lock = threading.Lock()
        self.started_at = time.time()
        self._inbound: "dict[str, dict[str, int]]" = {}
        self._outbound: "dict[str, dict[str, int]]" = {
            s: {o: 0 for o in _OUTCOMES} for s in SOURCES
        }
        # epoch-minute -> {key: count}, pruned as minutes roll out of the window.
        self._recent_in: "dict[int, dict[str, int]]" = {}
        self._recent_out: "dict[int, dict[str, int]]" = {}
        # name -> {"ts": last-beat epoch, "interval_s": expected cadence}. Lets the
        # dashboard judge a background daemon by liveness, not per-worker thread presence.
        self._heartbeats: "dict[str, dict[str, float]]" = {}
        # (event, referrer, day) -> count, pruned to EVENT_DAYS on write.
        self._events: "dict[tuple[str, str, str], int]" = {}
        self._recent_ev: "dict[int, dict[str, int]]" = {}

    def _bump_recent(self, buckets, key) -> None:
        """Count one event for ``key`` in the current-minute bucket (caller holds lock)."""
        minute = _now_minute()
        row = buckets.get(minute)
        if row is None:
            row = buckets[minute] = {}
            cutoff = minute - WINDOW_MINUTES
            for old in [m for m in buckets if m <= cutoff]:
                del buckets[old]   # drop minutes that have rolled out of the window
        row[key] = row.get(key, 0) + 1

    @staticmethod
    def _recent_totals(buckets) -> "dict[str, int]":
        """Sum each key's count over the last WINDOW minutes (caller holds lock)."""
        cutoff = _now_minute() - WINDOW_MINUTES + 1
        totals: "dict[str, int]" = {}
        for minute, row in buckets.items():
            if minute >= cutoff:
                for key, count in row.items():
                    totals[key] = totals.get(key, 0) + count
        return totals

    def record_inbound(self, category, bucket) -> None:
        with self._lock:
            row = self._inbound.get(category)
            if row is None:
                row = self._inbound[category] = _empty_inbound_row()
            row["total"] += 1
            row[bucket] = row.get(bucket, 0) + 1
            self._bump_recent(self._recent_in, category)

    def record_outbound(self, source, outcome) -> None:
        with self._lock:
            row = self._outbound.get(source)
            if row is None:
                row = self._outbound[source] = {o: 0 for o in _OUTCOMES}
            row[outcome] = row.get(outcome, 0) + 1
            self._bump_recent(self._recent_out, source)

    def record_heartbeat(self, name, ts, interval_s) -> None:
        with self._lock:
            self._heartbeats[name] = {"ts": ts, "interval_s": interval_s}

    def record_event(self, event, referrer, day) -> None:
        with self._lock:
            # Cap distinct referrers so a spray of forged Referer headers can't
            # grow the key space without bound.
            if referrer not in ("direct", "self", REFERRER_OTHER):
                seen = {r for _, r, _ in self._events}
                if referrer not in seen and len(seen) >= MAX_REFERRERS:
                    referrer = REFERRER_OTHER
            key = (event, referrer, day)
            self._events[key] = self._events.get(key, 0) + 1
            keep = _recent_days()
            for k in [k for k in self._events if k[2] not in keep]:
                del self._events[k]
            self._bump_recent(self._recent_ev, event)

    def read(self) -> dict:
        with self._lock:
            inbound = {k: dict(v) for k, v in self._inbound.items()}
            outbound = {k: dict(v) for k, v in self._outbound.items()}
            return {
                "started_at": self.started_at,
                "inbound": inbound,
                "outbound": outbound,
                "inbound_recent": self._recent_totals(self._recent_in),
                "outbound_recent": self._recent_totals(self._recent_out),
                "heartbeats": {k: dict(v) for k, v in self._heartbeats.items()},
                "events": _nest_events(self._events.items()),
                "events_recent": self._recent_totals(self._recent_ev),
            }


class _SqliteStore:
    """Cross-process counters in a shared SQLite file, so every uvicorn worker tallies
    into one place and the dashboard sees the whole deployment. WAL mode lets the workers
    write concurrently; the counts are tiny UPSERTs well within SQLite's throughput.
    """

    def __init__(self, path: str) -> None:
        self._lock = threading.Lock()
        # check_same_thread=False: the connection is shared across a worker's threads,
        # serialized by self._lock. isolation_level=None: autocommit each statement.
        self._db = sqlite3.connect(path, check_same_thread=False, isolation_level=None)
        self._db.execute("PRAGMA journal_mode=WAL")
        self._db.execute("PRAGMA synchronous=NORMAL")
        self._db.execute("PRAGMA busy_timeout=3000")
        self._db.executescript(
            """
            CREATE TABLE IF NOT EXISTS meta(key TEXT PRIMARY KEY, value REAL);
            CREATE TABLE IF NOT EXISTS inbound_cume(
                category TEXT, bucket TEXT, count INTEGER, PRIMARY KEY(category, bucket));
            CREATE TABLE IF NOT EXISTS outbound_cume(
                source TEXT, outcome TEXT, count INTEGER, PRIMARY KEY(source, outcome));
            CREATE TABLE IF NOT EXISTS inbound_minute(
                minute INTEGER, category TEXT, count INTEGER, PRIMARY KEY(minute, category));
            CREATE TABLE IF NOT EXISTS outbound_minute(
                minute INTEGER, source TEXT, count INTEGER, PRIMARY KEY(minute, source));
            CREATE TABLE IF NOT EXISTS event_cume(
                event TEXT, referrer TEXT, day TEXT, count INTEGER,
                PRIMARY KEY(event, referrer, day));
            CREATE TABLE IF NOT EXISTS event_minute(
                minute INTEGER, event TEXT, count INTEGER, PRIMARY KEY(minute, event));
            """
        )
        # First worker to touch the DB stamps the shared start time; the rest inherit it.
        self._db.execute(
            "INSERT OR IGNORE INTO meta(key, value) VALUES('started_at', ?)", (time.time(),)
        )
        self._last_pruned_minute = -1

    @property
    def started_at(self) -> float:
        row = self._db.execute(
            "SELECT value FROM meta WHERE key='started_at'").fetchone()
        return row[0] if row else time.time()

    def _prune(self, minute: int) -> None:
        """Drop per-minute rows that have rolled out of the window (idempotent across
        workers). Cheap, and only run when the epoch-minute advances."""
        if minute == self._last_pruned_minute:
            return
        self._last_pruned_minute = minute
        cutoff = minute - WINDOW_MINUTES
        self._db.execute("DELETE FROM inbound_minute WHERE minute <= ?", (cutoff,))
        self._db.execute("DELETE FROM outbound_minute WHERE minute <= ?", (cutoff,))
        self._db.execute("DELETE FROM event_minute WHERE minute <= ?", (cutoff,))
        # Event history is kept by day, not by minute.
        self._db.execute(
            "DELETE FROM event_cume WHERE day NOT IN (%s)"
            % ",".join("?" * EVENT_DAYS), tuple(sorted(_recent_days())),
        )

    def record_inbound(self, category, bucket) -> None:
        minute = _now_minute()
        with self._lock:
            self._db.execute(
                "INSERT INTO inbound_cume(category, bucket, count) VALUES(?, ?, 1) "
                "ON CONFLICT(category, bucket) DO UPDATE SET count = count + 1",
                (category, bucket),
            )
            self._db.execute(
                "INSERT INTO inbound_minute(minute, category, count) VALUES(?, ?, 1) "
                "ON CONFLICT(minute, category) DO UPDATE SET count = count + 1",
                (minute, category),
            )
            self._prune(minute)

    def record_outbound(self, source, outcome) -> None:
        minute = _now_minute()
        with self._lock:
            self._db.execute(
                "INSERT INTO outbound_cume(source, outcome, count) VALUES(?, ?, 1) "
                "ON CONFLICT(source, outcome) DO UPDATE SET count = count + 1",
                (source, outcome),
            )
            self._db.execute(
                "INSERT INTO outbound_minute(minute, source, count) VALUES(?, ?, 1) "
                "ON CONFLICT(minute, source) DO UPDATE SET count = count + 1",
                (minute, source),
            )
            self._prune(minute)

    def record_event(self, event, referrer, day) -> None:
        minute = _now_minute()
        with self._lock:
            if referrer not in ("direct", "self", REFERRER_OTHER):
                row = self._db.execute(
                    "SELECT COUNT(DISTINCT referrer) FROM event_cume").fetchone()
                known = self._db.execute(
                    "SELECT 1 FROM event_cume WHERE referrer = ? LIMIT 1", (referrer,)
                ).fetchone()
                if not known and row and row[0] >= MAX_REFERRERS:
                    referrer = REFERRER_OTHER
            self._db.execute(
                "INSERT INTO event_cume(event, referrer, day, count) VALUES(?, ?, ?, 1) "
                "ON CONFLICT(event, referrer, day) DO UPDATE SET count = count + 1",
                (event, referrer, day),
            )
            self._db.execute(
                "INSERT INTO event_minute(minute, event, count) VALUES(?, ?, 1) "
                "ON CONFLICT(minute, event) DO UPDATE SET count = count + 1",
                (minute, event),
            )
            self._prune(minute)

    def record_heartbeat(self, name, ts, interval_s) -> None:
        # Reuse the meta table (two keys per daemon) so every worker's beats land in the
        # one shared file — the dashboard then sees the notifier's liveness whichever
        # worker answers its poll, even though only the leader worker runs it.
        with self._lock:
            self._db.execute(
                "INSERT INTO meta(key, value) VALUES(?, ?) "
                "ON CONFLICT(key) DO UPDATE SET value = excluded.value",
                (f"hb_ts:{name}", ts),
            )
            self._db.execute(
                "INSERT INTO meta(key, value) VALUES(?, ?) "
                "ON CONFLICT(key) DO UPDATE SET value = excluded.value",
                (f"hb_int:{name}", interval_s),
            )

    def _read_heartbeats(self) -> "dict[str, dict[str, float]]":
        beats: "dict[str, dict[str, float]]" = {}
        for key, value in self._db.execute(
                "SELECT key, value FROM meta WHERE key LIKE 'hb_ts:%' OR key LIKE 'hb_int:%'"):
            field, _, name = key.partition(":")
            beats.setdefault(name, {})["ts" if field == "hb_ts" else "interval_s"] = value
        return beats

    def read(self) -> dict:
        cutoff = _now_minute() - WINDOW_MINUTES + 1
        with self._lock:
            inbound: "dict[str, dict[str, int]]" = {}
            for category, bucket, count in self._db.execute(
                    "SELECT category, bucket, count FROM inbound_cume"):
                row = inbound.setdefault(category, _empty_inbound_row())
                row[bucket] = row.get(bucket, 0) + count
                row["total"] += count
            # Start from the stable source rows so the dashboard shows them even at zero.
            outbound: "dict[str, dict[str, int]]" = {
                s: {o: 0 for o in _OUTCOMES} for s in SOURCES
            }
            for source, outcome, count in self._db.execute(
                    "SELECT source, outcome, count FROM outbound_cume"):
                outbound.setdefault(source, {o: 0 for o in _OUTCOMES})[outcome] = count
            recent_in = dict(self._db.execute(
                "SELECT category, SUM(count) FROM inbound_minute WHERE minute >= ? "
                "GROUP BY category", (cutoff,)))
            recent_out = dict(self._db.execute(
                "SELECT source, SUM(count) FROM outbound_minute WHERE minute >= ? "
                "GROUP BY source", (cutoff,)))
            events = _nest_events(
                ((event, referrer, day), count)
                for event, referrer, day, count in self._db.execute(
                    "SELECT event, referrer, day, count FROM event_cume")
            )
            recent_ev = dict(self._db.execute(
                "SELECT event, SUM(count) FROM event_minute WHERE minute >= ? "
                "GROUP BY event", (cutoff,)))
            return {
                "started_at": self.started_at,
                "inbound": inbound,
                "outbound": outbound,
                "inbound_recent": recent_in,
                "outbound_recent": recent_out,
                "heartbeats": self._read_heartbeats(),
                "events": events,
                "events_recent": recent_ev,
            }


def _make_store():
    """Pick the store from the environment. A path in ``THERMOGRAPH_METRICS_DB`` selects
    the shared SQLite store (needed when running multiple workers); otherwise in-memory.
    A store that fails to open falls back to memory — metrics must never break boot."""
    path = os.environ.get("THERMOGRAPH_METRICS_DB", "").strip()
    if path:
        try:
            return _SqliteStore(path)
        except Exception:  # noqa: BLE001 - never let instrumentation break startup
            pass
    return _MemStore()


_store = _make_store()


def record_inbound(category: str, status) -> None:
    """Count one inbound request in ``category``, bucketed by status class."""
    # The metrics endpoint is polled by the dashboard itself — counting it would just
    # show the monitor watching itself, so it's excluded from traffic entirely.
    # The event beacon is likewise not traffic: it is the *record* of traffic, and
    # counting it would double every interaction it reports.
    if category in ("metrics", "event"):
        return
    try:
        _store.record_inbound(category, _status_bucket(status))
    except Exception:  # noqa: BLE001 - instrumentation must never break a request
        pass


def record_outbound(phase: str, outcome: str) -> None:
    """Count one outbound call. ``outcome`` in {ok, retry, error, rate_limited}."""
    try:
        _store.record_outbound(source_for_phase(phase), outcome)
    except Exception:  # noqa: BLE001
        pass


# Per-IP token bucket for the unauthenticated event beacon. Bounded: the map is
# cleared whenever the minute rolls, so it holds at most one minute of clients.
_EVENT_RATE_PER_MIN = 30
_rate_lock = threading.Lock()
_rate_minute = -1
_rate_counts: "dict[str, int]" = {}


def _rate_ok(ip: "str | None") -> bool:
    global _rate_minute
    key = ip or "?"
    minute = _now_minute()
    with _rate_lock:
        if minute != _rate_minute:
            _rate_minute = minute
            _rate_counts.clear()
        count = _rate_counts.get(key, 0) + 1
        _rate_counts[key] = count
        return count <= _EVENT_RATE_PER_MIN


def record_event(name: str, referer: "str | None" = None,
                 host: "str | None" = None, ip: "str | None" = None) -> None:
    """Count one product event. The name is allowlisted and the referrer is taken
    from the request's own Referer header (never client-supplied, which would be
    trivially forgeable). Best-effort and silent: the caller always answers 204,
    so a rejected or rate-limited event tells a prober nothing."""
    try:
        if not _rate_ok(ip):
            return
        event = normalize_event(name)
        # An unrecognized event carries no referrer dimension — one bucket total.
        referrer = "direct" if event == EVENT_OTHER else normalize_referrer(referer, host)
        _store.record_event(event, referrer, _utc_day())
    except Exception:  # noqa: BLE001 - instrumentation must never break a request
        pass


def record_heartbeat(name: str, interval_s: float) -> None:
    """Stamp ``name``'s liveness. ``interval_s`` is the daemon's expected beat cadence, so
    a reader can tell fresh from stale without knowing the schedule. Best-effort."""
    try:
        _store.record_heartbeat(name, time.time(), interval_s)
    except Exception:  # noqa: BLE001 - instrumentation must never break the daemon
        pass


def snapshot() -> dict:
    """A JSON-serializable copy of all counters plus process facts."""
    try:
        data = _store.read()
    except Exception:  # noqa: BLE001 - a locked/broken store must not 500 the dashboard
        data = {"started_at": time.time(), "inbound": {}, "outbound": {},
                "inbound_recent": {}, "outbound_recent": {}, "heartbeats": {},
                "events": {}, "events_recent": {}}
    inbound = data["inbound"]
    outbound = data["outbound"]
    started_at = data["started_at"]
    # Convert each heartbeat's absolute timestamp to an age here, on the server clock, so
    # a reader (dashboard) never has to reconcile clock skew to judge freshness.
    now = time.time()
    heartbeats = {
        name: {"age_s": round(now - hb["ts"], 1), "interval_s": hb.get("interval_s")}
        for name, hb in (data.get("heartbeats") or {}).items()
        if hb.get("ts") is not None
    }
    return {
        "pid": os.getpid(),
        "started_at": started_at,
        "uptime_s": round(time.time() - started_at, 1),
        "inbound": inbound,
        "outbound": outbound,
        "inbound_total": sum(v.get("total", 0) for v in inbound.values()),
        "outbound_total": sum(sum(v.values()) for v in outbound.values()),
        # Traffic over the last WINDOW_MINUTES, per category / source.
        "window_minutes": WINDOW_MINUTES,
        "inbound_recent": data["inbound_recent"],
        "outbound_recent": data["outbound_recent"],
        "heartbeats": heartbeats,
        # Product events: {event: {referrer_domain: {day: count}}}, aggregate only.
        "events": data.get("events") or {},
        "events_recent": data.get("events_recent") or {},
        "events_total": sum(
            n
            for refs in (data.get("events") or {}).values()
            for days in refs.values()
            for n in days.values()
        ),
    }