"""Chat and plugin stream event buses for the SwiftUI WebView."""
from __future__ import annotations

import asyncio
import json
import threading


class ChatEventBus:
    """Thread-safe fan-out bus for JS snippets keyed by thread_id."""

    def __init__(self):
        self._lock = threading.Lock()
        self._subscribers: dict[str, set[asyncio.Queue]] = {}

    def emit(self, thread_id: str, js_code: str) -> None:
        with self._lock:
            for q in self._subscribers.get(thread_id, set()):
                try:
                    q.put_nowait(js_code)
                except asyncio.QueueFull:
                    pass

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

    def unsubscribe(self, thread_id: str, q: asyncio.Queue) -> None:
        with self._lock:
            subs = self._subscribers.get(thread_id)
            if subs:
                subs.discard(q)


class PluginStreamBus:
    """Thread-safe fan-out bus for plugin stream JSON events keyed by thread_id."""

    def __init__(self):
        self._lock = threading.Lock()
        self._subscribers: dict[str, set[asyncio.Queue]] = {}
        self._seq_by_thread: dict[str, int] = {}

    def emit(self, thread_id: str, event: dict) -> dict:
        with self._lock:
            seq = int(self._seq_by_thread.get(thread_id, 0)) + 1
            self._seq_by_thread[thread_id] = seq
            payload = dict(event or {})
            payload["thread_id"] = thread_id
            payload["seq"] = seq
            encoded = json.dumps(payload)
            # Fan out to the specific thread stream and an optional global stream.
            targets = set(self._subscribers.get(thread_id, set()))
            targets.update(self._subscribers.get("__all__", set()))
            for q in targets:
                try:
                    q.put_nowait(encoded)
                except asyncio.QueueFull:
                    pass
            return payload

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

    def unsubscribe(self, thread_id: str, q: asyncio.Queue) -> None:
        with self._lock:
            subs = self._subscribers.get(thread_id)
            if subs:
                subs.discard(q)
