"""Live CDC consumer for the Command Center "Changes" dashboard.

A background aiokafka consumer subscribes to the Debezium CDC topics on the
Kafka broker, parses each change event (insert/update/delete with before/after
images), keeps a bounded in-memory ring buffer, and fans every change out over
the WebSocket bus as a ``cdc_change`` event so the UI can render new & changed
data in real time.

Endpoints:
  GET /api/changes        -> recent changes from the ring buffer (filterable)
  GET /api/changes/stats  -> volume per source/table/op + per-minute buckets
"""

from __future__ import annotations

import asyncio
import json
import os
import re
import uuid
from collections import defaultdict, deque
from datetime import datetime, timezone
from typing import Any

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

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

KAFKA_BOOTSTRAP = os.getenv("KAFKA_BOOTSTRAP", "10.0.21.36:9092")
RING_SIZE = int(os.getenv("CDC_RING_SIZE", "5000"))
METRICS_RETENTION_MIN = int(os.getenv("CDC_METRICS_RETENTION_MIN", str(24 * 60)))

# Active CDC topic prefixes -> logical source. Matches Debezium topic.prefix.
PREFIX_SOURCE = {
    "postgres_sales": "postgres",
    "mysql_hr": "mysql",
    "mongodb_supplychain": "mongodb",
    "cassandra_telemetry": "cassandra",
    "neo4j_graph": "neo4j",
}
TOPIC_PATTERN = re.compile(
    r"^(postgres_sales|mysql_hr|mongodb_supplychain|cassandra_telemetry|neo4j_graph)\..+"
)

OP_MAP = {"c": "insert", "u": "update", "d": "delete", "r": "snapshot"}

_ring: deque[dict[str, Any]] = deque(maxlen=RING_SIZE)
# Minute-level rollups — accurate counts beyond the display ring buffer cap.
_minute_rollups: dict[int, dict[str, Any]] = {}
_state: dict[str, Any] = {
    "started": False,
    "connected": False,
    "consumed": 0,
    "last_error": None,
    "last_ts": None,
    "topics": [],
}


def _source_of(topic: str) -> tuple[str, str]:
    prefix = topic.split(".", 1)[0]
    source = PREFIX_SOURCE.get(prefix, prefix)
    table = topic.split(".")[-1]
    return source, table


def _coerce(v: Any) -> Any:
    """Debezium/Mongo nests the document as a JSON string sometimes."""
    if isinstance(v, str):
        try:
            return json.loads(v)
        except Exception:
            return v
    return v


def _key_fields(source: str, table: str, after: Any, before: Any) -> str:
    row = after or before or {}
    if not isinstance(row, dict):
        return ""
    prefer = ["order_id", "event_id", "_id", "id", "region", "order_status",
              "event_type", "type", "amount", "salary_change"]
    parts = []
    for k in prefer:
        if k in row and row[k] is not None:
            val = row[k]
            if isinstance(val, (dict, list)):
                continue
            sval = str(val)
            if len(sval) > 40:
                sval = sval[:40] + "…"
            parts.append(f"{k}={sval}")
        if len(parts) >= 4:
            break
    return " ".join(parts)


def _parse(topic: str, value: bytes | None) -> dict[str, Any] | None:
    if value is None:  # tombstone
        return None
    try:
        payload = json.loads(value.decode("utf-8"))
    except Exception:
        return None
    # Debezium envelope: {schema, payload:{before,after,op,source,ts_ms}}
    if isinstance(payload, dict) and "payload" in payload and isinstance(payload["payload"], dict):
        payload = payload["payload"]
    if not isinstance(payload, dict):
        return None
    op = OP_MAP.get(payload.get("op"), payload.get("op") or "change")
    before = _coerce(payload.get("before"))
    after = _coerce(payload.get("after"))
    src = payload.get("source") or {}
    source, table = _source_of(topic)
    if isinstance(src, dict) and src.get("table"):
        table = src.get("table")
    ts_ms = payload.get("ts_ms")
    ts = datetime.fromtimestamp(ts_ms / 1000, timezone.utc).isoformat() if ts_ms else datetime.now(timezone.utc).isoformat()
    return {
        "id": uuid.uuid4().hex[:10],
        "ts": ts,
        "source": source,
        "table": table,
        "topic": topic,
        "op": op,
        "summary": _key_fields(source, table, after, before),
        "before": before if isinstance(before, dict) else None,
        "after": after if isinstance(after, dict) else None,
    }


def _record_metrics(entry: dict[str, Any]) -> None:
    """Track per-minute aggregates for accurate window stats (not ring-limited)."""
    try:
        ts = datetime.fromisoformat(entry["ts"].replace("Z", "+00:00"))
    except Exception:
        ts = datetime.now(timezone.utc)
    minute_key = int(ts.timestamp()) // 60
    bucket = _minute_rollups.setdefault(
        minute_key,
        {"total": 0, "by_source": defaultdict(int), "by_op": defaultdict(int), "by_table": defaultdict(int)},
    )
    bucket["total"] += 1
    bucket["by_source"][entry["source"]] += 1
    bucket["by_op"][entry["op"]] += 1
    bucket["by_table"][entry["table"]] += 1

    cutoff = minute_key - METRICS_RETENTION_MIN
    for old_key in list(_minute_rollups.keys()):
        if old_key < cutoff:
            del _minute_rollups[old_key]


def _aggregate_window(minutes: int) -> dict[str, Any]:
    """Aggregate rollup buckets for the requested time window."""
    now = datetime.now(timezone.utc)
    cutoff_minute = int((now.timestamp() - minutes * 60) // 60)
    use_hour_buckets = minutes >= 120

    by_source: dict[str, int] = defaultdict(int)
    by_op: dict[str, int] = defaultdict(int)
    by_table: dict[str, int] = defaultdict(int)
    buckets: dict[str, int] = defaultdict(int)
    total = 0

    for minute_key, bucket in _minute_rollups.items():
        if minute_key < cutoff_minute:
            continue
        total += bucket["total"]
        for k, v in bucket["by_source"].items():
            by_source[k] += v
        for k, v in bucket["by_op"].items():
            by_op[k] += v
        for k, v in bucket["by_table"].items():
            by_table[k] += v
        ts = datetime.fromtimestamp(minute_key * 60, timezone.utc)
        label = ts.strftime("%H:00") if use_hour_buckets else ts.strftime("%H:%M")
        buckets[label] += bucket["total"]

    # Fill chart timeline with zero buckets so the window span is visible.
    if use_hour_buckets:
        span = max(1, (minutes + 59) // 60)
        filled: dict[str, int] = {}
        for i in range(span):
            t = datetime.fromtimestamp(now.timestamp() - (span - 1 - i) * 3600, timezone.utc)
            filled[t.strftime("%H:00")] = buckets.get(t.strftime("%H:00"), 0)
        buckets = filled
    else:
        span = max(1, minutes)
        filled = {}
        for i in range(span):
            t = datetime.fromtimestamp(now.timestamp() - (span - 1 - i) * 60, timezone.utc)
            filled[t.strftime("%H:%M")] = buckets.get(t.strftime("%H:%M"), 0)
        buckets = filled

    return {
        "total": total,
        "by_source": dict(by_source),
        "by_op": dict(by_op),
        "by_table": dict(by_table),
        "buckets": [{"t": k, "n": v} for k, v in sorted(buckets.items())],
        "inserts": by_op.get("insert", 0),
        "updates": by_op.get("update", 0),
        "deletes": by_op.get("delete", 0),
        "rate_per_min": round(total / max(1, minutes), 1),
    }


async def cdc_consumer_loop() -> None:
    """Background loop: consume CDC topics and publish each change live."""
    _state["started"] = True
    await asyncio.sleep(6)
    try:
        from aiokafka import AIOKafkaConsumer
    except Exception as exc:
        _state["last_error"] = f"aiokafka import failed: {exc}"
        return

    while True:
        consumer = None
        try:
            consumer = AIOKafkaConsumer(
                bootstrap_servers=KAFKA_BOOTSTRAP,
                group_id=f"atc-cc-cdc-{uuid.uuid4().hex[:8]}",
                auto_offset_reset="latest",
                enable_auto_commit=False,
                client_id="atc-command-center-cdc",
            )
            consumer.subscribe(pattern=TOPIC_PATTERN)
            await consumer.start()
            _state["connected"] = True
            _state["last_error"] = None
            from main import publish_event

            async for msg in consumer:
                entry = _parse(msg.topic, msg.value)
                if not entry:
                    continue
                _ring.append(entry)
                _record_metrics(entry)
                _state["consumed"] += 1
                _state["last_ts"] = entry["ts"]
                try:
                    _state["topics"] = sorted(consumer.subscription() or [])
                except Exception:
                    pass
                try:
                    await publish_event({"type": "cdc_change", "entry": entry})
                except Exception:
                    pass
        except Exception as exc:
            _state["connected"] = False
            _state["last_error"] = str(exc)
            await asyncio.sleep(10)  # backoff then reconnect
        finally:
            if consumer is not None:
                try:
                    await consumer.stop()
                except Exception:
                    pass


def snapshot(minutes: int = 15) -> dict[str, Any]:
    """Lightweight CDC snapshot for other modules (Data Flow graph)."""
    agg = _aggregate_window(minutes)
    return {
        "connected": _state["connected"],
        "consumed": _state["consumed"],
        "buffered": len(_ring),
        "window_total": agg["total"],
        "by_source": agg["by_source"],
    }


# ── Endpoints ────────────────────────────────────────────────────────────────
@router.get("")
async def list_changes(
    limit: int = Query(200, le=1000),
    source: str | None = None,
    op: str | None = None,
    table: str | None = None,
    minutes: int | None = Query(None, le=1440),
) -> JSONResponse:
    items = list(_ring)
    if minutes is not None:
        cutoff = datetime.now(timezone.utc).timestamp() - minutes * 60
        filtered_by_time: list[dict[str, Any]] = []
        for c in items:
            try:
                if datetime.fromisoformat(c["ts"].replace("Z", "+00:00")).timestamp() >= cutoff:
                    filtered_by_time.append(c)
            except Exception:
                continue
        items = filtered_by_time
    if source:
        items = [c for c in items if c["source"] == source]
    if op:
        items = [c for c in items if c["op"] == op]
    if table:
        items = [c for c in items if c["table"] == table]
    items = list(reversed(items))[:limit]
    return JSONResponse({
        "ok": True,
        "changes": items,
        "buffered": len(_ring),
        "buffer_cap": RING_SIZE,
        "connected": _state["connected"],
        "consumed": _state["consumed"],
        "last_error": _state["last_error"],
        "last_ts": _state["last_ts"],
    })


@router.get("/stats")
async def change_stats(minutes: int = Query(15, le=1440)) -> JSONResponse:
    agg = _aggregate_window(minutes)
    return JSONResponse({
        "ok": True,
        "window_minutes": minutes,
        **agg,
        "connected": _state["connected"],
        "consumed": _state["consumed"],
        "buffered": len(_ring),
        "buffer_cap": RING_SIZE,
        "last_ts": _state["last_ts"],
    })


@router.get("/status")
async def changes_status() -> JSONResponse:
    return JSONResponse({"ok": True, "buffered": len(_ring), "buffer_cap": RING_SIZE, **_state})