"""Data observability (Disease #6 — Traceability / observability).

A lightweight background monitor that, per business table, tracks over time:

  volume      – row count and its delta between cycles
  freshness   – age of the newest record
  schema      – the column set; a change raises a drift alert

It derives alerts (critical/warning/info) for volume drops, stalled ingestion,
stale data and schema drift, and keeps a rolling time-series per dataset for the
trend charts in the UI.

Endpoints:
  GET  /api/observability/metrics   -> per-dataset series + current state
  GET  /api/observability/alerts    -> active + recent alerts
  POST /api/observability/run       -> run one probe cycle now
  POST /api/observability/config    -> {enabled, interval_s, freshness_min}
"""

from __future__ import annotations

import threading
import time
import uuid
from collections import deque
from datetime import datetime, timezone
from typing import Any

from fastapi import APIRouter, Body
from fastapi.responses import JSONResponse

from lake_meta import DATASETS, trino_scalar, discover_columns

router = APIRouter(prefix="/api/observability", tags=["observability"])

_lock = threading.Lock()
_state: dict[str, Any] = {
    "enabled": True,
    "interval_s": 90.0,
    "freshness_min": 30.0,
    "running": False,
    "cycles": 0,
    "last_cycle_ts": 0.0,
    "next_run_ts": 0.0,
    "started_at": None,
    "datasets": {
        d["key"]: {"label": d["label"], "engine": d["engine"], "color": d["color"],
                   "table": d["fqtn"], "rows": None, "prev_rows": None, "delta": None,
                   "freshness_age_min": None, "columns": None, "schema_hash": None,
                   "series": deque(maxlen=80), "deltas": deque(maxlen=20),
                   "stalled_cycles": 0, "ts": None, "error": None}
        for d in DATASETS
    },
    "alerts_active": {},        # dedup_key -> alert
    "alerts_history": deque(maxlen=120),
    "feed": deque(maxlen=40),
}


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


def _term(text: str, level: str = "info", phase: str = "observe") -> None:
    try:
        from agent_terminal import emit_threadsafe
        emit_threadsafe("infra-sentinel", text, level=level, phase=phase)
    except Exception:
        pass


def _raise_alert(dataset: str, atype: str, severity: str, message: str) -> None:
    dedup = f"{dataset}:{atype}"
    existing = _state["alerts_active"].get(dedup)
    if existing:
        existing["count"] += 1
        existing["last_ts"] = _now().isoformat()
        existing["message"] = message
        return
    alert = {"id": uuid.uuid4().hex[:8], "dataset": dataset, "type": atype,
             "severity": severity, "message": message, "count": 1,
             "ts": _now().isoformat(), "last_ts": _now().isoformat(), "resolved": False}
    _state["alerts_active"][dedup] = alert
    _state["alerts_history"].appendleft(dict(alert))
    lvl = "err" if severity == "critical" else ("warn" if severity == "warning" else "info")
    _term(f"  ⚠ ALERT [{severity}] {dataset}: {message}", level=lvl, phase="alert")
    _state["feed"].appendleft({"ts": _now().isoformat(), "text": f"[{severity}] {dataset}: {message}", "level": lvl})


def _clear_alert(dataset: str, atype: str) -> None:
    dedup = f"{dataset}:{atype}"
    a = _state["alerts_active"].pop(dedup, None)
    if a:
        a["resolved"] = True
        a["resolved_ts"] = _now().isoformat()
        _state["alerts_history"].appendleft({**a, "message": f"resolved: {a['message']}"})


def _native_freshness(ds: dict[str, Any]) -> float | None:
    """Epoch seconds of the newest record, read cheaply via the PK index
    (ORDER BY <pk> DESC LIMIT 1) on the source DB. Returns None if unavailable."""
    nc = ds.get("native_count")
    ts = ds.get("ts_col")
    key = ds.get("key_col")
    if not nc or not ts:
        return None
    eng = nc[0]
    try:
        import sql_console as s
        if eng == "postgres" and key:
            import psycopg2
            conn = psycopg2.connect(host=s.DB_HOST, port=s.PG_PORT, user=s.PG_USER,
                                    password=s.PG_PASS, dbname=s.PG_DB, connect_timeout=6)
            try:
                cur = conn.cursor()
                cur.execute("SET statement_timeout = 6000")
                cur.execute(f'SELECT extract(epoch FROM "{ts}") FROM public."{ds["table"]}" '
                            f'WHERE "{ts}" IS NOT NULL ORDER BY "{key}" DESC LIMIT 1')
                row = cur.fetchone()
                return float(row[0]) if row and row[0] is not None else None
            finally:
                conn.close()
        if eng == "mysql" and key:
            import pymysql
            conn = pymysql.connect(host=s.DB_HOST, port=s.MYSQL_PORT, user=s.MYSQL_USER,
                                   password=s.MYSQL_PASS, database=s.MYSQL_DB, connect_timeout=6)
            try:
                cur = conn.cursor()
                cur.execute("SET SESSION MAX_EXECUTION_TIME = 6000")
                cur.execute(f"SELECT UNIX_TIMESTAMP(`{ts}`) FROM `{ds['table']}` "
                            f"WHERE `{ts}` IS NOT NULL ORDER BY `{key}` DESC LIMIT 1")
                row = cur.fetchone()
                return float(row[0]) if row and row[0] is not None else None
            finally:
                conn.close()
        if eng == "mongodb":
            cli = s._mongo_client()
            try:
                doc = list(cli[ds["schema"]][ds["table"]].find({}, {ts: 1, "_id": 0}).sort("_id", -1).limit(1))
                if doc:
                    import datetime as _dt
                    v = doc[0].get(ts)
                    if isinstance(v, _dt.datetime):
                        return v.timestamp()
            finally:
                cli.close()
    except Exception:
        return None
    return None


def _probe(ds: dict[str, Any]) -> None:
    key = ds["key"]
    st = _state["datasets"][key]
    fq = ds["fqtn"]

    # volume — prefer a fast native planner estimate (postgres/mysql/mongo); use
    # Trino metadata count for Iceberg; treat a slow/failed count as a soft miss.
    rows = None
    estimated = False
    nc = ds.get("native_count")
    if nc:
        try:
            import sql_console as _s
            rows = _s._table_row_count(nc[0], nc[1])
            estimated = True
        except Exception:
            rows = None
    if rows is None:
        try:
            rows = int(trino_scalar(f"SELECT count(*) FROM {fq}", timeout=30.0) or 0)
        except Exception as exc:
            st["error"] = str(exc)[:140]
            st["count_fails"] = st.get("count_fails", 0) + 1
            return
    st["error"] = None
    st["count_fails"] = 0
    st["estimated"] = estimated

    prev = st["rows"]
    st["prev_rows"] = prev
    st["rows"] = rows
    delta = (rows - prev) if prev is not None else None
    st["delta"] = delta
    st["series"].append({"t": _now().strftime("%H:%M:%S"), "rows": rows, "delta": delta or 0})
    st["ts"] = _now().isoformat()

    if delta is not None:
        st["deltas"].append(delta)
        # Guard against planner-estimate jitter: only alert on a material drop.
        drop_threshold = max(2000, int(0.02 * (prev or 0)))
        if delta < -drop_threshold:
            _raise_alert(key, "volume_drop", "warning",
                         f"row count dropped by {abs(delta):,} ({prev:,} → {rows:,})")
        else:
            _clear_alert(key, "volume_drop")
        # stalled ingestion: no growth for several cycles on CDC sources. Skip for
        # estimate-based counts (planner stats update lazily → false stalls).
        if delta == 0 and not ds.get("curated") and not estimated:
            st["stalled_cycles"] += 1
            if st["stalled_cycles"] >= 4:
                _raise_alert(key, "stalled", "warning",
                             f"no new rows for {st['stalled_cycles']} cycles")
        else:
            st["stalled_cycles"] = 0
            _clear_alert(key, "stalled")
        # spike (informational) — only for exact counts
        pos = [d for d in st["deltas"] if d > 0]
        if not estimated and pos and delta > (sum(pos) / len(pos)) * 6 and len(pos) >= 4:
            _raise_alert(key, "spike", "info", f"volume spike +{delta:,} rows this cycle")

    # freshness — cheap PK-indexed newest-row lookup for the CDC sources, with a
    # short Trino max(ts) fallback for Iceberg tables.
    if ds.get("ts_col"):
        epoch = _native_freshness(ds)
        if epoch is None:
            try:
                v = trino_scalar(f'SELECT to_unixtime(max("{ds["ts_col"]}")) FROM {fq}', timeout=10.0)
                epoch = float(v) if v is not None else None
            except Exception:
                epoch = None
        if epoch:
            age = max(0.0, (time.time() - float(epoch)) / 60.0)
            st["freshness_age_min"] = round(age, 1)
            if age > float(_state["freshness_min"]) and not ds.get("curated"):
                _raise_alert(key, "stale", "warning",
                             f"newest record is {age:.0f}m old (> {_state['freshness_min']:.0f}m)")
            else:
                _clear_alert(key, "stale")

    # schema drift
    cols = discover_columns(ds)
    if cols:
        names = sorted(c["name"] for c in cols)
        h = hash(tuple(names))
        if st["schema_hash"] is not None and h != st["schema_hash"]:
            old = set(st["columns"] or [])
            new = set(names)
            added = sorted(new - old)
            removed = sorted(old - new)
            parts = []
            if added:
                parts.append(f"+{', '.join(added)}")
            if removed:
                parts.append(f"-{', '.join(removed)}")
            _raise_alert(key, "schema_drift", "critical", "schema changed: " + " ".join(parts))
        st["columns"] = names
        st["schema_hash"] = h


def run_cycle() -> dict[str, Any]:
    if _state["running"]:
        return {"ok": True, "skipped": "already running"}
    _state["running"] = True
    try:
        _term(f"═══ observability sweep {_state['cycles'] + 1} — volume · freshness · schema ═══",
              level="info", phase="cycle")
        for ds in DATASETS:
            if not _state["enabled"]:
                break
            try:
                _probe(ds)
            except Exception as exc:
                _state["datasets"][ds["key"]]["error"] = str(exc)[:140]
        _state["cycles"] += 1
        _state["last_cycle_ts"] = time.time()
        _state["next_run_ts"] = time.time() + float(_state["interval_s"])
        active = len(_state["alerts_active"])
        _term(f"═══ sweep {_state['cycles']} done — {active} active alert(s) ═══",
              level=("warn" if active else "ok"), phase="cycle")
    finally:
        _state["running"] = False
    return {"ok": True, "active_alerts": len(_state["alerts_active"])}


def _loop() -> None:
    _state["started_at"] = _now().isoformat()
    time.sleep(28)
    while True:
        try:
            if _state["enabled"]:
                run_cycle()
        except Exception:
            pass
        time.sleep(max(30.0, float(_state["interval_s"])))


threading.Thread(target=_loop, daemon=True, name="observability").start()


def metrics_view() -> dict[str, Any]:
    with _lock:
        datasets = []
        for ds in DATASETS:
            st = _state["datasets"][ds["key"]]
            datasets.append({
                "key": ds["key"], "label": st["label"], "engine": st["engine"],
                "color": st["color"], "table": st["table"], "rows": st["rows"],
                "delta": st["delta"], "freshness_age_min": st["freshness_age_min"],
                "columns": len(st["columns"]) if st["columns"] else None,
                "stalled_cycles": st["stalled_cycles"], "error": st["error"],
                "ts": st["ts"], "series": list(st["series"]),
            })
        return {
            "ok": True, "enabled": _state["enabled"], "interval_s": _state["interval_s"],
            "freshness_min": _state["freshness_min"], "running": _state["running"],
            "cycles": _state["cycles"], "last_cycle_ts": _state["last_cycle_ts"],
            "next_run_ts": _state["next_run_ts"],
            "alert_counts": _alert_counts(),
            "datasets": datasets, "feed": list(_state["feed"])[:20],
        }


def _alert_counts() -> dict[str, int]:
    counts = {"critical": 0, "warning": 0, "info": 0}
    for a in _state["alerts_active"].values():
        counts[a["severity"]] = counts.get(a["severity"], 0) + 1
    counts["total"] = len(_state["alerts_active"])
    return counts


def summary_for_llm() -> dict[str, Any]:
    with _lock:
        return {
            "active_alerts": _alert_counts(),
            "alerts": [{"dataset": a["dataset"], "type": a["type"], "severity": a["severity"],
                        "message": a["message"]} for a in _state["alerts_active"].values()],
        }


@router.get("/metrics")
async def get_metrics() -> JSONResponse:
    return JSONResponse(metrics_view())


@router.get("/alerts")
async def get_alerts() -> JSONResponse:
    with _lock:
        return JSONResponse({
            "ok": True,
            "counts": _alert_counts(),
            "active": sorted(_state["alerts_active"].values(),
                             key=lambda a: {"critical": 0, "warning": 1, "info": 2}.get(a["severity"], 3)),
            "history": list(_state["alerts_history"])[:60],
        })


@router.post("/run")
async def post_run() -> JSONResponse:
    from starlette.concurrency import run_in_threadpool
    res = await run_in_threadpool(run_cycle)
    return JSONResponse({**res, "metrics": metrics_view()})


@router.post("/config")
async def post_config(body: dict = Body(default={})) -> JSONResponse:
    if "enabled" in body:
        _state["enabled"] = bool(body["enabled"])
    if "interval_s" in body:
        try:
            _state["interval_s"] = max(30.0, min(900.0, float(body["interval_s"])))
        except Exception:
            pass
    if "freshness_min" in body:
        try:
            _state["freshness_min"] = max(1.0, min(1440.0, float(body["freshness_min"])))
        except Exception:
            pass
    return JSONResponse({"ok": True, "enabled": _state["enabled"],
                         "interval_s": _state["interval_s"], "freshness_min": _state["freshness_min"]})