"""Process-wide event sink + SSE subscription bus for the kanban subsystem.

Two integration paths share this module:

  1. ``set_kanban_event_sink`` / ``notify_kanban_event`` — process-wide
     push callback (mirrors :func:`utils.task_plan.set_task_plan_event_sink`)
     so storage and dispatcher can announce events without importing
     ``mugi_backend.HeadlessDelegate``. The backend wires the sink at
     startup with a callback that pushes a ``system_event`` SSE frame to
     subscribed clients.

  2. ``subscribe`` / ``unsubscribe`` — thread-safe asyncio.Queue fan-out
     used by the dedicated ``/kanban/stream`` SSE endpoint. Subscribers
     register with their loop; ``notify_kanban_event`` schedules the put
     onto each subscriber's loop via ``call_soon_threadsafe`` so emitters
     remain pure-thread (the dispatcher is not asyncio).

Event payloads are **ciphertext-free** by contract — they carry ids,
status transitions, assignee, tenant, run id, attempt, and timestamps only.
SSE consumers fetch the decrypted body via the auth-gated REST endpoint.
"""

from __future__ import annotations

import asyncio
import logging
import threading
from typing import Any, Callable

_log = logging.getLogger(__name__)

_EventSink = Callable[[dict, str | None], None]
_event_sink: _EventSink | None = None

# Registered SSE subscribers. We hold (loop, queue) tuples and use
# call_soon_threadsafe to enqueue, so notify_kanban_event remains pure
# threading-safe and never blocks on a slow client.
_subs_lock = threading.Lock()
_subscribers: list[tuple[asyncio.AbstractEventLoop, asyncio.Queue]] = []
_MAX_SUBSCRIBER_QUEUE = 500


def set_kanban_event_sink(sink: _EventSink | None) -> None:
    """Register (or clear) the process-wide push callback for kanban events.

    The sink is called as ``sink(event_dict, thread_id_or_none)``. It must
    not raise; exceptions are swallowed and logged at debug level so a
    failing sink cannot break a storage write.
    """
    global _event_sink
    _event_sink = sink


def notify_kanban_event(payload: dict, *, thread_id: str | None = None) -> None:
    """Emit a kanban event through the registered sink AND any /kanban/stream
    subscribers. No-op when neither is wired. Always safe to call from any
    thread.
    """
    if not isinstance(payload, dict):
        return
    snapshot = dict(payload)
    sink = _event_sink
    if sink is not None:
        try:
            sink(dict(snapshot), thread_id)
        except Exception:
            _log.debug("kanban event sink raised", exc_info=True)
    with _subs_lock:
        if not _subscribers:
            return
        subs = list(_subscribers)
    dead: list[tuple[asyncio.AbstractEventLoop, asyncio.Queue]] = []
    for loop, queue in subs:
        try:
            loop.call_soon_threadsafe(_safe_put_nowait, queue, snapshot)
        except RuntimeError:
            dead.append((loop, queue))
        except Exception:
            _log.debug("notify_kanban_event subscriber dispatch failed", exc_info=True)
    if dead:
        with _subs_lock:
            for entry in dead:
                if entry in _subscribers:
                    _subscribers.remove(entry)


def _safe_put_nowait(queue: asyncio.Queue, payload: dict) -> None:
    try:
        dropped = False
        if queue.qsize() >= _MAX_SUBSCRIBER_QUEUE:
            try:
                queue.get_nowait()
                dropped = True
            except Exception:
                pass
        if dropped:
            _log.info(
                "kanban SSE subscriber queue full; dropped oldest event (queue cap=%d)",
                _MAX_SUBSCRIBER_QUEUE,
            )
        queue.put_nowait(payload)
    except Exception:
        _log.debug("kanban subscriber queue put failed", exc_info=True)


def subscribe() -> asyncio.Queue:
    """Register an SSE subscriber and return its asyncio.Queue. Caller must
    own the running asyncio loop and call :func:`unsubscribe` when done.
    """
    loop = asyncio.get_running_loop()
    queue: asyncio.Queue = asyncio.Queue()
    with _subs_lock:
        _subscribers.append((loop, queue))
    return queue


def unsubscribe(queue: asyncio.Queue) -> None:
    with _subs_lock:
        for entry in list(_subscribers):
            if entry[1] is queue:
                _subscribers.remove(entry)


def subscriber_count() -> int:
    """Test / health helper."""
    with _subs_lock:
        return len(_subscribers)


def get_kanban_event_sink() -> _EventSink | None:
    """Test/inspection helper. Returns the currently registered sink."""
    return _event_sink
