thermograph/backend/core/events.py

"""Product-event instrumentation: what people actually *do* in the UI.

This is the durable, dimensioned tier that sits above ``core/metrics.py``'s
since-start counters. metrics.py answers "is traffic flowing right now"; this
module answers "did anyone ever get a location, and did search work" — over
weeks, across deploys.

**Two tiers, two flags, both OFF by default.** Everything here is inert unless
the matching flag is set, so shipping it is a no-op.

* **Tier A — anonymous aggregates (``THERMOGRAPH_EVENTS``).** Hourly counts keyed
  by allowlisted enums. No identifier of any kind, nothing written to the
  visitor's device, so it needs no consent. This runs for *every* visitor and is
  the unbiased signal.
* **Tier B — consented session rows (``THERMOGRAPH_EVENT_SESSIONS``).** The same
  events, additionally carrying an ephemeral per-tab session id, written one row
  per event so a funnel ("of the sessions that opened the map, how many reached a
  grade?") can actually be computed. **Only ever sent by a browser whose visitor
  gave opt-in consent** — see frontend/static/consent.js. This is a behavioural
  sequence dataset, which is exactly what Tier A was designed to avoid, so it
  carries a much shorter retention and extra minimisation.

Tier B depends on Tier A: with ``THERMOGRAPH_EVENTS`` off, nothing records at all.

Design constraints, in priority order:

1. **No stable identifier.** No cookie, no cross-session id, no fingerprint, and
   never the account id — see ``validate_session``. The Tier B id is minted in
   the browser, lives in ``sessionStorage`` (per tab), rotates on a 30-minute idle
   / 2-hour absolute cap, and is unlinkable to any other session, device, or
   account. Tier A has no id column at all.
2. **No free text, no coordinates.** Every event field is drawn from a fixed
   allowlist (see ``SCHEMA``); an unrecognised name or value collapses to a
   bounded "other" bucket. The one deliberate exception is the failed-search text
   captured server-side by ``record_search_miss`` — heavily normalised, rejected
   on any personal-data smell, and pruned unless several people typed the same
   thing. It is never accepted from the browser.
3. **No IP.** The caller's IP is used only as an in-memory rate-limit key for the
   current minute (see ``metrics.rate_ok``) and is never passed in here, never
   stored, and never joined to a session id.
4. **Never breaks a request.** Every entry point swallows its own errors.

Storage (Postgres; see ``alembic/versions/0003_ui_events.py`` and ``0004``):

* ``ui_event_hourly`` — Tier A hourly aggregate, 400-day retention.
* ``ui_event_session`` — Tier B per-event rows, **30-day** retention, minute-
  granularity timestamps and an in-session sequence number instead of precise
  clock times.
* ``ui_search_miss`` — normalised zero-result queries, 90-day retention, with
  singletons pruned at 7 days (see ``SEARCH_MISS_MIN_COUNT``).

Plus one JSONL line per event into the activity stream so Loki/Grafana can show
the live picture. Off Postgres (dev, tests, offline tooling) the SQL sinks
degrade to no-ops and only the JSONL line is written — the same dialect split
``data/store.py`` and ``core/metrics.py`` already use.
"""
import datetime
import os
import re
import threading
import unicodedata

from core import audit

# --- feature flags -------------------------------------------------------------
# Default OFF everywhere. Turn on per-environment (dev first, then beta, then
# prod) once the privacy copy in frontend/templates/privacy.html.j2 matches what
# is actually recorded.


def _flag(name: str) -> bool:
    return os.environ.get(name, "").strip().lower() in ("1", "true", "yes", "on")


#: Tier A. Nothing at all records without this.
ENABLED = _flag("THERMOGRAPH_EVENTS")

#: Tier B. Consented, session-scoped rows. Separate flag on purpose: Tier A is
#: the thing worth running permanently, and Tier B should be switchable off
#: (e.g. at the end of a time-boxed study) without losing the aggregate history.
SESSIONS_ENABLED = _flag("THERMOGRAPH_EVENT_SESSIONS")

#: Zero-result search text. Separate flag again — it is the only free-text field
#: in the system and deserves its own on/off switch and its own decision.
SEARCH_MISS_ENABLED = _flag("THERMOGRAPH_SEARCH_MISS")

DATABASE_URL = os.environ.get("THERMOGRAPH_DATABASE_URL", "").strip()
IS_POSTGRES = DATABASE_URL.startswith("postgresql")


def _libpq_dsn(url: str) -> str:
    """A plain libpq URL psycopg accepts (mirrors climate_store/metrics)."""
    return url.replace("+asyncpg", "").replace("+psycopg", "")


_PG_DSN = _libpq_dsn(DATABASE_URL) if IS_POSTGRES else ""

# --- schema -------------------------------------------------------------------
# Seven events. Every one answers a product question that cannot be answered from
# the request log alone; anything the request log *can* answer (page views, API
# usage, error rates) is deliberately NOT duplicated here.
#
# Three dimension slots are stored — ``view``, ``prop``, ``value`` — plus a
# referrer domain on ``view.open`` only (acquisition), so the key space stays
# small enough to eyeball. Each event declares which slots it uses and the exact
# allowlist for each.

VIEWS = frozenset({
    "home", "calendar", "day", "compare", "score", "alerts", "legend",
    "city", "cities", "records", "month", "glossary", "about", "privacy",
})

#: Where a location came from. The activation question: do visitors ever get
#: past "no place picked", and by which route?
PICK_METHODS = frozenset({"geolocate", "map", "search", "restore", "link"})

#: Did the location search actually work? ``hit`` = results shown, ``typo`` =
#: results only after respelling (places.suggest's ``corrected``), ``miss`` =
#: nothing found. The query string itself is never sent or stored.
SEARCH_OUTCOMES = frozenset({"hit", "typo", "miss"})

#: In-view controls worth knowing about. Anything not listed is dropped, so a
#: new control has to be added here deliberately (and reviewed for privacy).
CONTROLS = frozenset({
    "chart_metric",     # value: the metric key
    "unit",             # value: F | C
    "date",             # value: today | past
    "range",            # value: the span bucket
    "season",           # value: season key or "custom"
    "totals",           # calendar totals mode
    "dist_metric",      # compare distribution metric
    "comfort",          # compare comfort slicer moved
    "compare_add",      # value: the location-count bucket after the add
    "compare_remove",
    "filter_sheet",     # mobile filter sheet opened
    "day_step",         # day view prev/next
})

#: Bounded value vocabulary, shared across controls. A value outside this set is
#: stored as "other" — never verbatim.
VALUES = frozenset({
    # chart / distribution metrics
    "tmax", "tmin", "feels", "humid", "wind", "gust", "precip", "dry",
    # units
    "F", "C",
    # dates
    "today", "past", "future",
    # spans / counts (bucketed client-side, never a raw number)
    "1", "2", "3", "4", "5plus",
    "1y", "2y", "5y", "10y", "max",
    # seasons
    "winter", "spring", "summer", "fall", "custom", "all",
    # generic
    "on", "off", "other",
})

SHARE_KINDS = frozenset({"link", "png", "permalink"})

#: The alerts funnel. ``ok`` records whether the step succeeded, so a high
#: start/low create ratio is visible without any per-person tracking.
ALERT_STEPS = frozenset({"open", "start", "create", "push_on", "push_off", "test"})

#: Dead ends — the highest-signal events here. A spike in any of these is a bug
#: report nobody filed.
DEADENDS = frozenset({
    "grade_error",   # /api/grade returned an error the user saw
    "no_data",       # a cell with no usable history
    "geo_denied",    # browser geolocation refused / timed out
    "search_miss",   # search returned nothing (paired with place.search=miss)
    "not_found",     # a 404 page was rendered
    "offline",       # the fetch failed with no network
})

OK = frozenset({"y", "n"})

#: name -> {slot: allowlist}. A slot absent from an event is always stored "".
SCHEMA: "dict[str, dict[str, frozenset]]" = {
    # A view was opened (page load or SPA view). ``prop`` is how it was reached.
    "view.open":    {"view": VIEWS, "prop": frozenset({"direct", "nav", "link", "restore"})},
    # A location was chosen. NEVER carries coordinates — only the route taken.
    "place.pick":   {"view": VIEWS, "prop": PICK_METHODS},
    # A location search completed. Recorded server-side from api_suggest as well
    # as client-side, so searches from every surface are covered.
    "place.search": {"prop": SEARCH_OUTCOMES},
    # An in-view control was used.
    "view.control": {"view": VIEWS, "prop": CONTROLS, "value": VALUES},
    # Share / export.
    "share":        {"view": VIEWS, "prop": SHARE_KINDS},
    # The alerts funnel.
    "alert":        {"prop": ALERT_STEPS, "value": OK},
    # A dead end the visitor actually hit.
    "deadend":      {"view": VIEWS, "prop": DEADENDS},
}

#: Anything unrecognised collapses here with every dimension blank — visible
#: enough to notice abuse, bounded enough to be harmless.
EVENT_OTHER = "other"
VALUE_OTHER = "other"

SLOTS = ("view", "prop", "value")

#: Max events per batched beacon payload. The client batches bursts (chart-metric
#: taps, slider drags) rather than firing a request each.
MAX_BATCH = 20

#: Max JSON body the beacon will read, in bytes. Must comfortably exceed
#: MAX_BATCH full-sized events (~120 bytes each with a session id) or the byte
#: cap, not the batch cap, becomes the binding limit and a legitimate full batch
#: is silently dropped. Still far too small to be worth abusing as storage.
MAX_BODY_BYTES = 4096

# --- Tier B: the ephemeral session id ------------------------------------------
# The id is minted in the browser (16 random bytes, base64url, 22 chars) and is
# the ONE field a client could otherwise use to smuggle arbitrary text into the
# pipeline — a free-text column with a different name. So it is validated by
# shape, not merely length-capped: anything that is not exactly 22 base64url
# characters is discarded and the event still records to Tier A. There is
# nothing to gain by sending a "clever" id, and nothing gets stored if you try.
_SESSION_RE = re.compile(r"\A[A-Za-z0-9_-]{22}\Z")

#: Rows per session before Tier B stops recording (Tier A keeps counting). Bounds
#: both storage and how long/detailed any one behavioural sequence can get — a
#: 200-step sequence is already far more than any funnel question needs.
MAX_SEQ = 200


def validate_session(sid) -> str:
    """A well-formed ephemeral session id, or "" — never anything else.

    Deliberately NOT a sanitiser: an id that does not match the expected shape is
    dropped entirely rather than trimmed into shape, because the only thing a
    malformed id can be is either a bug or an attempt to use the field as a
    free-text channel, and both should lose their payload completely.

    The account id is never accepted here and is never read on this route: see
    web/app.py's api_event, which does not touch the session cookie. Tying an
    anonymous behavioural sequence to an identified person would convert this
    whole dataset into personal data about a known individual, for essentially
    no analytical gain (the questions are about the product, not about who).
    """
    if not isinstance(sid, str):
        return ""
    return sid if _SESSION_RE.match(sid) else ""


def validate_seq(seq) -> int:
    """An in-session sequence number in [0, MAX_SEQ], or -1 for "not in a
    session". Ordering is carried by this counter rather than by a precise
    timestamp, so the stored time can stay coarse (see ``_minute_bucket``)."""
    if isinstance(seq, bool) or not isinstance(seq, int):
        return -1
    return seq if 0 <= seq <= MAX_SEQ else -1


# --- the one free-text field: zero-result search queries ------------------------
# Captured server-side only (api_suggest), never accepted from a browser. What
# people searched for and did NOT find is the single highest-value product signal
# here — it names the gaps in the place index — but a search box is a field a
# person can type anything into, so this is the only place where personal data
# could arrive incidentally. Four layers of mitigation, in order of strength:
#
#   1. Only ZERO-RESULT queries are considered at all. A query that found a place
#      is already answered by the `hit` counter and is never stored as text.
#   2. Reject (never sanitise) anything with a personal-data smell — an address,
#      an email, a URL, a long digit run, a sentence. Stripping bad characters
#      would keep the text; rejecting loses it, which is the point.
#   3. Store as a COUNT per normalised string per day, not as a row per search.
#   4. Prune anything only ever typed by a couple of people (see
#      SEARCH_MISS_MIN_COUNT). A string three different people typed is a product
#      signal; a string one person typed once is the risky case, and it is the
#      one that gets deleted.

#: Longer than this is a sentence or an address, not a place name.
SEARCH_MISS_MAX_LEN = 40
#: More words than this likewise.
SEARCH_MISS_MAX_WORDS = 5
#: A run of digits this long is a phone number / ID, not a postcode.
SEARCH_MISS_MAX_DIGIT_RUN = 6
#: Rows below this count are deleted once they are older than
#: SEARCH_MISS_PRUNE_DAYS — a k-anonymity floor on the free-text field.
SEARCH_MISS_MIN_COUNT = 3
SEARCH_MISS_PRUNE_DAYS = 7

# Letters (any script), marks, digits, and the punctuation that genuinely occurs
# in place names. Anything else is a smell, not a typo.
_PLACE_OK = re.compile(r"\A[^\W_]+(?:[ \-'.,()·’][^\W_]+)*\Z", re.UNICODE)
_DIGIT_RUN = re.compile(r"\d{%d,}" % (SEARCH_MISS_MAX_DIGIT_RUN + 1))
_URLISH = re.compile(r"(://|www\.|\.[a-z]{2,}(/|\Z))", re.IGNORECASE)


def normalize_search_miss(q) -> str:
    """A failed query reduced to a storable form, or "" to store nothing.

    NFKC-normalised, casefolded, whitespace-collapsed, length- and word-capped.
    Returns "" — meaning *drop it* — for anything that smells like personal data
    rather than a place name.
    """
    if not isinstance(q, str):
        return ""
    q = unicodedata.normalize("NFKC", q).strip()
    q = " ".join(q.split())          # collapse all internal whitespace runs
    if not q:
        return ""
    # Casefold, not lower(): correct for ß/İ and friends, and it collapses the
    # trivial "same query, different capitalisation" variants into one row.
    q = q.casefold()
    if len(q) > SEARCH_MISS_MAX_LEN:
        return ""
    if len(q.split(" ")) > SEARCH_MISS_MAX_WORDS:
        return ""
    if "@" in q or _URLISH.search(q):        # email / URL / hostname
        return ""
    if _DIGIT_RUN.search(q):                 # phone number, account id, coords
        return ""
    if not _PLACE_OK.match(q):               # punctuation soup, not a place name
        return ""
    return q


def normalize(name: str, props: "dict | None" = None) -> "tuple[str, dict[str, str]]":
    """An allowlisted ``(event, {view, prop, value})`` pair.

    The allowlist is the whole security model of an unauthenticated write
    endpoint: it is what stops the beacon from becoming unbounded storage, a
    spam sink, or an accidental PII channel. Unknown event -> ``EVENT_OTHER``
    with blank dimensions; unknown value in a known slot -> ``VALUE_OTHER``;
    a slot the event does not declare -> dropped entirely.
    """
    name = (name or "").strip()
    spec = SCHEMA.get(name)
    if spec is None:
        return EVENT_OTHER, {s: "" for s in SLOTS}
    props = props if isinstance(props, dict) else {}
    out: "dict[str, str]" = {}
    for slot in SLOTS:
        allowed = spec.get(slot)
        if allowed is None:
            out[slot] = ""
            continue
        raw = props.get(slot)
        raw = raw.strip() if isinstance(raw, str) else ""
        out[slot] = raw if raw in allowed else (VALUE_OTHER if raw else "")
    return name, out


# --- durable aggregate sink ---------------------------------------------------
# One row per (hour, event, view, prop, value, referrer). No row is ever written
# per interaction, so the table cannot hold a sequence of one person's actions
# even in principle.

_local = threading.local()


def _conn():
    """Thread-local autocommit psycopg connection, or None when Postgres is off
    or unreachable. Mirrors data/climate_store.py."""
    if not IS_POSTGRES:
        return None
    conn = getattr(_local, "conn", None)
    if conn is not None and not conn.closed:
        return conn
    try:
        import psycopg  # local import: only the Postgres path needs it
        _local.conn = psycopg.connect(_PG_DSN, autocommit=True)
    except Exception:  # noqa: BLE001 - degrade to "no durable sink", never raise
        _local.conn = None
    return _local.conn


def _hour_bucket() -> datetime.datetime:
    now = datetime.datetime.now(datetime.timezone.utc)
    return now.replace(minute=0, second=0, microsecond=0)


def _minute_bucket() -> datetime.datetime:
    """Tier B timestamps are truncated to the minute. Precise clock times are a
    fingerprint (inter-event millisecond gaps are individuating, and they let two
    rotated sessions be stitched back together by adjacency); the *order* of a
    session's events is what a funnel needs, and that is carried by ``seq``."""
    now = datetime.datetime.now(datetime.timezone.utc)
    return now.replace(second=0, microsecond=0)


def _utc_day() -> datetime.date:
    return datetime.datetime.now(datetime.timezone.utc).date()


_UPSERT = """
    INSERT INTO ui_event_hourly (bucket, event, view, prop, value, referrer, count)
    VALUES (%s, %s, %s, %s, %s, %s, 1)
    ON CONFLICT (bucket, event, view, prop, value, referrer)
    DO UPDATE SET count = ui_event_hourly.count + 1
"""


def _upsert(event: str, dims: "dict[str, str]", referrer: str) -> None:
    conn = _conn()
    if conn is None:
        return
    try:
        conn.execute(_UPSERT, (_hour_bucket(), event, dims["view"], dims["prop"],
                               dims["value"], referrer))
    except Exception:  # noqa: BLE001 - a broken connection retires itself below
        try:
            conn.close()
        except Exception:  # noqa: BLE001
            pass
        _local.conn = None


_INSERT_SESSION = """
    INSERT INTO ui_event_session (ts, session, seq, event, view, prop, value, referrer)
    VALUES (%s, %s, %s, %s, %s, %s, %s, %s)
"""


def _insert_session(session: str, seq: int, event: str,
                    dims: "dict[str, str]", referrer: str) -> None:
    conn = _conn()
    if conn is None:
        return
    try:
        conn.execute(_INSERT_SESSION, (_minute_bucket(), session, seq, event,
                                       dims["view"], dims["prop"], dims["value"], referrer))
    except Exception:  # noqa: BLE001 - a broken connection retires itself
        try:
            conn.close()
        except Exception:  # noqa: BLE001
            pass
        _local.conn = None


_UPSERT_SEARCH_MISS = """
    INSERT INTO ui_search_miss (day, q, count) VALUES (%s, %s, 1)
    ON CONFLICT (day, q) DO UPDATE SET count = ui_search_miss.count + 1
"""


def _upsert_search_miss(q: str) -> None:
    conn = _conn()
    if conn is None:
        return
    try:
        conn.execute(_UPSERT_SEARCH_MISS, (_utc_day(), q))
    except Exception:  # noqa: BLE001
        try:
            conn.close()
        except Exception:  # noqa: BLE001
            pass
        _local.conn = None


#: Deletes the free-text rows that only one or two people ever produced, once
#: they are old enough that the count has stopped growing. Run daily (see the
#: rollout section of UI-EVENTS.md); parameterised here rather than typed by hand
#: on the box so the retention rule is reviewable in the repo.
PRUNE_SEARCH_MISS_SQL = (
    "DELETE FROM ui_search_miss "
    "WHERE day < CURRENT_DATE - INTERVAL '%d days' AND count < %d"
    % (SEARCH_MISS_PRUNE_DAYS, SEARCH_MISS_MIN_COUNT)
)


# --- public entry points -------------------------------------------------------

def record(name: str, props: "dict | None" = None,
           referer: "str | None" = None, host: "str | None" = None,
           session: "str | None" = None, seq=None) -> None:
    """Record one product event. Best-effort, silent, and a no-op unless the
    feature flag is on.

    ``referer``/``host`` come from the request's own headers (a client-supplied
    referrer would be forgeable and buys nothing) and are reduced to a bare
    domain by ``metrics.normalize_referrer`` — never a full URL, which can carry
    a search query or a private path. The referrer dimension is kept only for
    ``view.open``; every other event stores "" there, which is what keeps the
    key space small.

    ``session``/``seq`` are the Tier B additions. They are only ever populated by
    a browser whose visitor opted in, they are validated by shape (see
    ``validate_session``), and Tier A records identically whether they are
    present or not — so a dropped or refused session id costs the aggregate
    nothing.
    """
    if not ENABLED:
        return
    try:
        # Import here, not at module scope: metrics imports cleanly on its own and
        # this keeps the dependency one-directional at import time.
        from core import metrics
        event, dims = normalize(name, props)
        referrer = metrics.normalize_referrer(referer, host) if event == "view.open" else ""
        # Tier A first and unconditionally: the anonymous aggregate is the signal
        # that must never depend on consent rates, client behaviour, or Tier B.
        _upsert(event, dims, referrer)
        sid = validate_session(session) if SESSIONS_ENABLED else ""
        n = validate_seq(seq) if sid else -1
        if sid and n >= 0:
            _insert_session(sid, n, event, dims, referrer)
        # Second sink: one structured line per event into the activity stream, so
        # the Loki/Grafana stack can show the live picture without waiting on an
        # hourly rollup. Carries the same allowlisted enums and nothing else —
        # NOT the session id, which would put a behavioural sequence into a log
        # stream with its own retention and its own access rules.
        audit.log_activity("ui.event", {"event": event, **dims, "referrer": referrer})
    except Exception:  # noqa: BLE001 - instrumentation must never break a request
        pass


def record_search_miss(q) -> None:
    """Record one zero-result search query, server-side only.

    Never call this with a query that returned results, and never accept the
    string from a browser: the browser sending its own search text would put it
    in a request body, in the client, and in anything that logs bodies — all of
    which this deliberately avoids. Silently stores nothing when the query does
    not survive ``normalize_search_miss``.
    """
    if not (ENABLED and SEARCH_MISS_ENABLED):
        return
    try:
        norm = normalize_search_miss(q)
        if norm:
            _upsert_search_miss(norm)
    except Exception:  # noqa: BLE001 - instrumentation must never break a request
        pass