thermograph/backend/core/audit.py

"""Lightweight file-based audit + error logging for Thermograph.

Two log streams, written as JSON Lines (one self-contained JSON object per line):

* ``logs/audit/audit-<date>.jsonl`` — one line per graded run, recording the
  total wall-clock duration, a per-phase timing breakdown, how many days the run
  covered (the recent-window size and the historical span), and a ``run_type``
  tag of **full** (a fresh ~45-year history fetch happened) or **partial** (the
  history came from cache, so only the recent days were fetched).

* ``logs/errors/errors-<date>.jsonl`` — a *separate folder* for anything that
  went wrong: each upstream failure is tagged ``retry`` (another attempt will be
  made) or ``error`` (final failure), and whole-run failures are tagged ``error``.

Everything is best-effort: logging never raises into the request path.
"""
import contextlib
import contextvars
import datetime
import json
import os
import threading
import time
import uuid

import paths

LOG_ROOT = paths.LOGS_DIR
AUDIT_DIR = os.path.join(LOG_ROOT, "audit")
ERROR_DIR = os.path.join(LOG_ROOT, "errors")
ACCESS_DIR = os.path.join(LOG_ROOT, "access")
HEARTBEAT_DIR = os.path.join(LOG_ROOT, "heartbeat")
ACTIVITY_DIR = os.path.join(LOG_ROOT, "activity")

_lock = threading.Lock()

# Set for the duration of a run so error/retry logs from deep in the fetch layer
# can be correlated back to the run that triggered them.
current_run_id: "contextvars.ContextVar[str | None]" = contextvars.ContextVar(
    "thermograph_run_id", default=None
)


def _now_iso() -> str:
    return datetime.datetime.now().astimezone().isoformat(timespec="milliseconds")


def _today() -> str:
    return datetime.date.today().isoformat()


def _append(path: str, record: dict) -> None:
    try:
        os.makedirs(os.path.dirname(path), exist_ok=True)
        line = json.dumps(record, default=str)
        with _lock:
            with open(path, "a", encoding="utf-8") as f:
                f.write(line + "\n")
    except Exception:  # noqa: BLE001 - logging must never break the request
        pass


def log_event(tag: str, record: dict) -> dict:
    """Record an error or retry in the separate errors folder.

    ``tag`` is conventionally ``"retry"`` or ``"error"``.
    """
    rec = {"ts": _now_iso(), "tag": tag, "run_id": current_run_id.get(), **record}
    _append(os.path.join(ERROR_DIR, f"errors-{_today()}.jsonl"), rec)
    return rec


def log_activity(tag: str, record: dict) -> dict:
    """Record a non-error application event — auth, subscriptions, notification
    delivery, app lifecycle, slow requests — to a dedicated *activity* stream,
    kept separate from the errors folder so these never pollute error dashboards.
    ``tag`` names the event (e.g. ``"auth.login"``, ``"notify.pass"``). Carries no
    PII: user ids are UUIDs, and no emails/request bodies/push endpoints. Best-effort."""
    rec = {"ts": _now_iso(), "tag": tag, "run_id": current_run_id.get(), **record}
    _append(os.path.join(ACTIVITY_DIR, f"activity-{_today()}.jsonl"), rec)
    return rec


def log_access(record: dict) -> None:
    """Append one request line (client IP, method, path, status) to the daily access
    log, so request/IP data is retained for later analysis. Best-effort; never raises
    into the request path."""
    _append(os.path.join(ACCESS_DIR, f"access-{_today()}.jsonl"), {"ts": _now_iso(), **record})


def log_heartbeat(daemon: str, interval_s: float) -> None:
    """Emit one liveness beat for a background daemon (e.g. the subscription
    notifier) to the heartbeat log stream. This is the log-shipping counterpart to
    ``metrics.record_heartbeat``: the observability stack tails these JSONL files, so
    a beat here becomes a queryable ``tag="heartbeat"`` signal in Loki/Grafana — the
    dashboard can tell a live daemon from a dead one by whether beats keep arriving
    within ``interval_s``, without polling the in-process metrics endpoint. Best-effort."""
    _append(
        os.path.join(HEARTBEAT_DIR, f"heartbeat-{_today()}.jsonl"),
        {"ts": _now_iso(), "level": "info", "tag": "heartbeat",
         "daemon": daemon, "interval_s": interval_s},
    )


class RunAudit:
    """Times a graded run and writes one audit line when the ``with`` block exits.

    Usage::

        with audit.RunAudit(endpoint="grade", lat=..., lon=...) as run:
            with run.phase("history"):
                ...
            run.set(run_type="full", history_rows=n)
            return response
    """

    def __init__(self, **params):
        self.run_id = uuid.uuid4().hex[:12]
        self.fields = {"run_id": self.run_id, **params}
        self.phases: dict[str, float] = {}
        self._t0 = time.perf_counter()
        self._token = None

    def __enter__(self) -> "RunAudit":
        self._token = current_run_id.set(self.run_id)
        return self

    def set(self, **kw) -> "RunAudit":
        """Attach extra fields (e.g. run_type, history span) to the audit record."""
        self.fields.update(kw)
        return self

    @contextlib.contextmanager
    def phase(self, name: str):
        """Time a named part of the run; the duration (ms) lands in ``phases``."""
        start = time.perf_counter()
        try:
            yield
        finally:
            self.phases[name] = round((time.perf_counter() - start) * 1000, 1)

    def __exit__(self, exc_type, exc, tb) -> bool:
        total = round((time.perf_counter() - self._t0) * 1000, 1)
        status = "ok"
        if exc is not None:
            status = "error"
            self.fields["error"] = repr(exc)
            code = getattr(exc, "status_code", None)
            if code is not None:
                self.fields["http_status"] = code
            # Mirror whole-run failures into the separate errors folder.
            log_event("error", {
                "phase": "run",
                "error": repr(exc),
                **{k: self.fields.get(k) for k in ("endpoint", "lat", "lon", "cell_id")},
            })
        record = {
            "ts": _now_iso(),
            "status": status,
            "duration_ms": total,
            "phases": self.phases,
            **self.fields,
        }
        _append(os.path.join(AUDIT_DIR, f"audit-{_today()}.jsonl"), record)
        if self._token is not None:
            current_run_id.reset(self._token)
        return False  # never suppress the exception