"""
Global activity event bus for app-wide background visibility.

The bus is intentionally thread-safe and lightweight:
- emit(event) can be called from any thread.
- subscribe() returns an asyncio.Queue for SSE fan-out.
- snapshot() returns active tasks + recent history for reconnect hydration.
"""
from __future__ import annotations

import asyncio
import json
import logging
import threading
import time
from typing import Any

_log = logging.getLogger(__name__)
from run_context import get_run_context, sanitize_run_id

_MAX_HISTORY = 1000
_MAX_SUMMARY_CHARS = 500
_MAX_TAGS = 12
_MAX_META_JSON_CHARS = 4000

# Optional process-global emitter callback so modules that do not hold a direct
# bus reference can still publish activity.
_activity_emitter = None
_activity_emitter_lock = threading.Lock()
_EARLY_EVENT_BUFFER_MAX = 64
_early_event_buffer: list[dict[str, Any]] = []
_early_buffer_lock = threading.Lock()


def set_activity_emitter(callback) -> None:
    """Set process-global activity emitter callback (one dict arg)."""
    global _activity_emitter
    pending: list[dict[str, Any]] = []
    with _activity_emitter_lock:
        _activity_emitter = callback
        if callback is None:
            with _early_buffer_lock:
                _early_event_buffer.clear()
            return
        with _early_buffer_lock:
            pending = list(_early_event_buffer)
            _early_event_buffer.clear()
    for evt in pending:
        try:
            callback(evt)
        except Exception:
            _log.debug("activity emitter callback failed (buffer flush)", exc_info=True)


def emit_activity_event(event: dict) -> None:
    """Emit an activity event through the process-global emitter if configured."""
    payload = dict(event or {})
    with _activity_emitter_lock:
        cb = _activity_emitter
        if cb is None:
            with _early_buffer_lock:
                if len(_early_event_buffer) >= _EARLY_EVENT_BUFFER_MAX:
                    _early_event_buffer.pop(0)
                _early_event_buffer.append(payload)
            return
    try:
        cb(payload)
    except Exception:
        _log.debug("activity emitter callback failed", exc_info=True)


def early_activity_buffer_size() -> int:
    """Test/health helper: events buffered before emitter registration."""
    with _early_buffer_lock:
        return len(_early_event_buffer)


def _sanitize_tags(values: Any) -> list[str]:
    if not isinstance(values, list):
        return []
    out: list[str] = []
    seen: set[str] = set()
    for item in values:
        if len(out) >= _MAX_TAGS:
            break
        tag = str(item or "").strip()
        if not tag or tag in seen:
            continue
        seen.add(tag)
        out.append(tag[:40])
    return out


def _sanitize_meta(value: Any) -> dict[str, Any]:
    if not isinstance(value, dict):
        return {}
    out: dict[str, Any] = {}
    for key, meta_val in value.items():
        k = str(key or "").strip()
        if not k:
            continue
        out[k[:80]] = meta_val
    try:
        raw = json.dumps(out, default=str)
    except Exception:
        return {"_truncated": True, "preview": str(out)[:_MAX_META_JSON_CHARS]}
    if len(raw) <= _MAX_META_JSON_CHARS:
        return out
    return {"_truncated": True, "preview": raw[:_MAX_META_JSON_CHARS]}


def normalize_activity_payload(event: dict[str, Any], *, tags: list[str] | None = None) -> dict[str, Any]:
    payload: dict[str, Any] = dict(event or {})
    ctx = get_run_context()
    if "run_id" not in payload and ctx.get("run_id"):
        payload["run_id"] = ctx.get("run_id")
    if "parent_run_id" not in payload and ctx.get("parent_run_id"):
        payload["parent_run_id"] = ctx.get("parent_run_id")
    if "thread_id" not in payload and ctx.get("thread_id"):
        payload["thread_id"] = ctx.get("thread_id")
    if "run_id" in payload:
        payload["run_id"] = sanitize_run_id(str(payload.get("run_id") or ""))
        if not payload["run_id"]:
            payload.pop("run_id", None)
    if "parent_run_id" in payload:
        payload["parent_run_id"] = sanitize_run_id(str(payload.get("parent_run_id") or ""))
        if not payload["parent_run_id"]:
            payload.pop("parent_run_id", None)
    summary = str(payload.get("summary") or "")
    payload["summary"] = summary[:_MAX_SUMMARY_CHARS]
    merged_tags = _sanitize_tags(payload.get("tags"))
    if tags:
        merged_tags = _sanitize_tags(merged_tags + list(tags))
    if merged_tags:
        payload["tags"] = merged_tags
    else:
        payload.pop("tags", None)
    if "meta" in payload:
        payload["meta"] = _sanitize_meta(payload.get("meta"))
    return payload


def emit_activity_enriched(event: dict, tags: list[str] | None = None) -> None:
    emit_activity_event(normalize_activity_payload(dict(event or {}), tags=tags))


# Optional process-global system event emitter so any in-process module can
# push transient system events to the chat UI via SSE.
_system_event_emitter = None
_system_event_lock = threading.Lock()


def set_system_event_emitter(callback) -> None:
    """Set process-global system event emitter.

    Callback signature: (event_dict: dict, thread_id: str | None) -> None
    """
    global _system_event_emitter
    with _system_event_lock:
        _system_event_emitter = callback


def emit_system_event(text: str, source: str = "system", thread_id: str = "") -> None:
    """Emit a system event to the chat UI via SSE. Safe to call from any thread."""
    with _system_event_lock:
        cb = _system_event_emitter
    if cb is None:
        return
    try:
        cb(
            {"type": "system_event", "text": text[:500], "source": source[:50]},
            thread_id or None,
        )
    except Exception:
        _log.debug("system event emitter callback failed", exc_info=True)


class ActivityBus:
    """Thread-safe fan-out bus with active-task tracking."""

    def __init__(self) -> None:
        self._lock = threading.Lock()
        self._subscribers: set[asyncio.Queue] = set()
        self._history: list[dict[str, Any]] = []
        self._active: dict[str, dict[str, Any]] = {}
        self._seq = 0

    def _event_key(self, event: dict[str, Any]) -> str:
        source = str(event.get("source") or "unknown")
        job_id = str(event.get("job_id") or "").strip()
        if job_id:
            return f"{source}:{job_id}"
        thread_id = str(event.get("thread_id") or "").strip()
        if thread_id:
            return f"{source}:thread:{thread_id}"
        return f"{source}:global"

    def emit(self, event: dict[str, Any]) -> dict[str, Any]:
        payload: dict[str, Any] = normalize_activity_payload(dict(event or {}))
        payload.setdefault("schema", "mugi.activity.v1")
        payload.setdefault("schema_version", 1)
        payload.setdefault("source", "unknown")
        payload.setdefault("state", "info")
        payload.setdefault("summary", "")
        payload.setdefault("ts_ms", int(time.time() * 1000))
        with self._lock:
            self._seq += 1
            payload["seq"] = self._seq

            key = self._event_key(payload)
            state = str(payload.get("state") or "info").lower()
            if state in {"started", "progress", "info", "heartbeat"}:
                self._active[key] = payload
            elif state in {"completed", "error", "cancelled", "stopped"}:
                self._active.pop(key, None)

            self._history.append(payload)
            if len(self._history) > _MAX_HISTORY:
                self._history = self._history[-_MAX_HISTORY:]

            for q in list(self._subscribers):
                try:
                    q.put_nowait(payload)
                except asyncio.QueueFull:
                    try:
                        q.get_nowait()
                        q.put_nowait(payload)
                        _log.info(
                            "activity bus subscriber queue full; dropped oldest event (source=%s)",
                            payload.get("source"),
                        )
                    except Exception:
                        _log.debug(
                            "activity bus subscriber queue full for %s (event retained in history)",
                            self._event_key(payload),
                        )
        return payload

    def subscribe(self) -> asyncio.Queue:
        q: asyncio.Queue = asyncio.Queue(maxsize=4096)
        with self._lock:
            self._subscribers.add(q)
        return q

    def unsubscribe(self, q: asyncio.Queue) -> None:
        with self._lock:
            self._subscribers.discard(q)

    def snapshot(self, *, recent_limit: int = 200) -> dict[str, Any]:
        with self._lock:
            recent = list(self._history[-max(0, int(recent_limit)):])
            active = list(self._active.values())
            return {
                "active": active,
                "recent": recent,
                "seq": self._seq,
                "generated_at_ms": int(time.time() * 1000),
            }
