feat(cdc): live Debezium CDC consumer + /api/changes for the Changes dashboard

Add api/cdc_consumer.py: aiokafka background consumer subscribes to the CDC
topics (postgres_sales/mysql_hr/mongodb_supplychain + cassandra/neo4j), parses
Debezium before/after envelopes, keeps a ring buffer and publishes each change
live as type=cdc_change. Endpoints /api/changes, /api/changes/stats, /status.

api/Dockerfile

@@ -4,7 +4,7 @@ WORKDIR /app
 RUN apt-get update && apt-get install -y --no-install-recommends curl && rm -rf /var/lib/apt/lists/*
 COPY requirements.txt .
 RUN pip install --no-cache-dir -r requirements.txt
-COPY main.py lab_context.py agent_terminal.py workload.py node_registry.py node_ops.py topology_views.py supervisor.py approval_service.py db.py dockhand_envs.py presentation.py database_inventory.py presentation_upload.py presentation_static.py storage_s3.py elasticsearch_api.py sql_console.py hdfs_api.py ssh_terminal.py pipeline_ops.py hadoop_analytics.py agent_ops.py hive_bench_seed.json .
+COPY main.py lab_context.py agent_terminal.py workload.py node_registry.py node_ops.py topology_views.py supervisor.py approval_service.py db.py dockhand_envs.py presentation.py database_inventory.py presentation_upload.py presentation_static.py storage_s3.py elasticsearch_api.py sql_console.py hdfs_api.py ssh_terminal.py pipeline_ops.py hadoop_analytics.py agent_ops.py cdc_consumer.py hive_bench_seed.json .
 RUN mkdir -p /data
 ENV DATABASE_URL=sqlite:////data/atc-agents.db
 EXPOSE 3201

api/cdc_consumer.py

@@ -0,0 +1,246 @@
+"""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 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", "1000"))
+
+# 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)
+_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,
+    }
+
+
+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)
+                _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
+
+
+# ── Endpoints ────────────────────────────────────────────────────────────────
+@router.get("")
+async def list_changes(
+    limit: int = Query(100, le=500),
+    source: str | None = None,
+    op: str | None = None,
+    table: str | None = None,
+) -> JSONResponse:
+    items = list(_ring)
+    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),
+        "connected": _state["connected"],
+        "consumed": _state["consumed"],
+        "last_error": _state["last_error"],
+    })
+
+
+@router.get("/stats")
+async def change_stats(minutes: int = Query(15, le=240)) -> JSONResponse:
+    now = datetime.now(timezone.utc)
+    cutoff = now.timestamp() - minutes * 60
+    by_source: dict[str, int] = {}
+    by_op: dict[str, int] = {}
+    by_table: dict[str, int] = {}
+    buckets: dict[str, int] = {}
+    total = 0
+    for c in _ring:
+        try:
+            t = datetime.fromisoformat(c["ts"]).timestamp()
+        except Exception:
+            continue
+        if t < cutoff:
+            continue
+        total += 1
+        by_source[c["source"]] = by_source.get(c["source"], 0) + 1
+        by_op[c["op"]] = by_op.get(c["op"], 0) + 1
+        by_table[c["table"]] = by_table.get(c["table"], 0) + 1
+        bucket = datetime.fromtimestamp(t, timezone.utc).strftime("%H:%M")
+        buckets[bucket] = buckets.get(bucket, 0) + 1
+    return JSONResponse({
+        "ok": True,
+        "window_minutes": minutes,
+        "total": total,
+        "by_source": by_source,
+        "by_op": by_op,
+        "by_table": by_table,
+        "buckets": [{"t": k, "n": v} for k, v in sorted(buckets.items())],
+        "connected": _state["connected"],
+        "consumed": _state["consumed"],
+    })
+
+
+@router.get("/status")
+async def changes_status() -> JSONResponse:
+    return JSONResponse({"ok": True, "buffered": len(_ring), **_state})

api/main.py

@@ -43,6 +43,7 @@ from hadoop_analytics import router as hadoop_router
 from elasticsearch_api import router as elasticsearch_router
 from sql_console import router as sql_router
 from agent_ops import router as agent_ops_router, agent_dml_loop
+from cdc_consumer import router as cdc_router, cdc_consumer_loop
 from ssh_terminal import ssh_session
 from node_registry import NODE_IDS, NODE_AGENT, NODE_REGISTRY, is_node_id
 from node_ops import build_node_detail, probe_node, run_node_probe_task
@@ -713,10 +714,12 @@ async def lifespan(app: FastAPI):
         await terminal_log(nid, f"{meta['label']} shell ready — click node to connect", level="info", phase="boot")
     task = asyncio.create_task(heartbeat_loop())
     dml_task = asyncio.create_task(agent_dml_loop())
+    cdc_task = asyncio.create_task(cdc_consumer_loop())
     add_feed("infra-sentinel", "ATC Command Center API online", "info")
     yield
     task.cancel()
     dml_task.cancel()
+    cdc_task.cancel()
     if redis_client:
         await redis_client.close()
 
@@ -729,6 +732,7 @@ app.include_router(hadoop_router)
 app.include_router(elasticsearch_router)
 app.include_router(sql_router)
 app.include_router(agent_ops_router)
+app.include_router(cdc_router)
 app.add_middleware(
     CORSMiddleware,
     allow_origins=["*"],

api/requirements.txt

@@ -15,3 +15,4 @@ neo4j==5.26.0
 python-pptx==1.0.2
 boto3==1.35.99
 paramiko==3.5.0
+aiokafka==0.12.0