api/event_bus.py

"""
backend/api/event_bus.py — Event Bus pub/sub (Fase 1 ADR-S26-S30)

Responsabilità: TRASPORTARE eventi tra componenti in modo asincrono ed efimero.
NON persiste — per la persistenza usa event_store.py.

Architettura:
  Publisher → Event Bus → [Subscriber1, Subscriber2, ...]
                      ↓
              Redis fanout (Upstash REST) per multi-process pub/sub

Topics predefiniti (espandibili via publish):
  task.created | task.completed | task.failed
  tool.started  | tool.finished
  memory.updated
  response.generated
  session.started | session.ended
  workflow.started | workflow.step | workflow.completed | workflow.failed

Invarianti ADR:
  S27: ogni evento pubblicato ha correlation_id tracciabile
  S30: nessuna dipendenza da provider LLM specifico

Endpoints:
  POST /api/events/publish         — pubblica evento (auth: MACHINE)
  GET  /api/events/stream/{topic}  — SSE stream (auth: MACHINE)
  GET  /api/events/bus/status      — diagnostica (auth: MACHINE)
"""
import asyncio, json, time, uuid, logging, os
from typing import AsyncIterator
from fastapi import APIRouter, Depends, Request
from fastapi.responses import StreamingResponse
from pydantic import BaseModel, Field
from .auth_guard import require_role, AuthRole
from .state import safe_json_dumps

_logger = logging.getLogger("api.event_bus")

router = APIRouter(
    prefix="/api/events",
    tags=["event-bus"],
    dependencies=[Depends(require_role(AuthRole.MACHINE))],
)

# ── In-memory subscriber registry ──────────────────────────────────────────────
# topic → set of asyncio.Queue (one per active SSE subscriber)
_subscribers: dict[str, set[asyncio.Queue]] = {}
_subscribers_lock = asyncio.Lock()

# ── Predefined topics (open set — publishers can add arbitrary topics) ─────────
BUILTIN_TOPICS = {
    "task.created", "task.completed", "task.failed",
    "tool.started",  "tool.finished",
    "memory.updated",
    "response.generated",
    "session.started", "session.ended",
    "workflow.started", "workflow.step", "workflow.completed", "workflow.failed",
}

# ── Max queue depth per subscriber (prevents memory leak on slow clients) ──────
_MAX_QUEUE_DEPTH = int(os.getenv("EVENT_BUS_QUEUE_DEPTH", "256"))


# ── Pydantic models ────────────────────────────────────────────────────────────

class BusEvent(BaseModel):
    topic:          str         = Field(..., description="Topic dell'evento (es. task.created)")
    payload:        dict        = Field(default_factory=dict)
    correlation_id: str | None  = Field(None, description="ID tracciabilità cross-componente (S27)")
    session_id:     str | None  = None
    source:         str | None  = None  # componente mittente (es. 'planner', 'executor')


class PublishResponse(BaseModel):
    event_id:       str
    topic:          str
    delivered_to:   int         # numero di subscriber locali notificati
    redis_fanout:   bool        # True se fanout Redis riuscito


# ── Redis fanout (fire-and-forget, non bloccante) ───────────────────────────────

async def _redis_fanout(topic: str, event_payload: dict) -> bool:
    """Pubblica l'evento su Redis LIST per fanout multi-process. Non lancia mai."""
    try:
        import httpx
        redis_url   = os.getenv("UPSTASH_REDIS_REST_URL",   "")
        redis_token = os.getenv("UPSTASH_REDIS_REST_TOKEN", "")
        if not redis_url or not redis_token:
            return False
        key  = f"eb:{topic}"
        data = json.dumps(event_payload)
        async with httpx.AsyncClient(timeout=1.5) as c:
            await c.post(
                redis_url,
                json=["LPUSH", key, data],
                headers={"Authorization": f"Bearer {redis_token}"},
            )
            # TTL 60s: un evento non consumato entro 60s viene scartato
            await c.post(
                redis_url,
                json=["EXPIRE", key, 60],
                headers={"Authorization": f"Bearer {redis_token}"},
            )
        return True
    except Exception as exc:
        _logger.debug("[event_bus] redis fanout skip: %s", exc)
        return False


# ── Core publish ───────────────────────────────────────────────────────────────

async def publish(
    topic:          str,
    payload:        dict,
    correlation_id: str | None = None,
    session_id:     str | None = None,
    source:         str | None = None,
) -> dict:
    """
    Pubblica un evento sul bus. Chiamabile internamente da qualsiasi modulo.
    Restituisce il dizionario evento con event_id assegnato.
    """
    event = {
        "event_id":       str(uuid.uuid4()),
        "topic":          topic,
        "payload":        payload,
        "correlation_id": correlation_id or str(uuid.uuid4()),
        "session_id":     session_id,
        "source":         source,
        "timestamp":      time.time(),
    }

    delivered = 0
    async with _subscribers_lock:
        queues = _subscribers.get(topic, set()).copy()

    for q in queues:
        try:
            q.put_nowait(event)
            delivered += 1
        except asyncio.QueueFull:
            _logger.warning("[event_bus] subscriber queue full on topic=%s — drop event", topic)

    # Redis fanout asincrono (non attende)
    asyncio.create_task(_redis_fanout(topic, event))

    _logger.debug("[event_bus] published topic=%s event_id=%s delivered_local=%d",
                  topic, event["event_id"][:8], delivered)
    return event


# ── Endpoints ─────────────────────────────────────────────────────────────────

@router.post("/publish", response_model=PublishResponse, summary="Pubblica evento sul bus")
async def publish_event(evt: BusEvent) -> PublishResponse:
    """
    Pubblica un evento su un topic. I subscriber SSE attivi ricevono l'evento
    immediatamente. Redis fanout notifica i processi remoti.
    """
    result = await publish(
        topic=evt.topic,
        payload=evt.payload,
        correlation_id=evt.correlation_id,
        session_id=evt.session_id,
        source=evt.source,
    )
    async with _subscribers_lock:
        n = len(_subscribers.get(evt.topic, set()))
    return PublishResponse(
        event_id=result["event_id"],
        topic=evt.topic,
        delivered_to=n,
        redis_fanout=True,  # ottimistico — fanout è fire-and-forget
    )


@router.get("/stream/{topic}", summary="SSE stream eventi per topic")
async def stream_events(topic: str, request: Request):
    """
    Server-Sent Events stream per un topic specifico.
    Il client rimane connesso e riceve ogni evento pubblicato su quel topic.
    La connessione si chiude quando il client disconnette.
    """
    q: asyncio.Queue = asyncio.Queue(maxsize=_MAX_QUEUE_DEPTH)

    async with _subscribers_lock:
        if topic not in _subscribers:
            _subscribers[topic] = set()
        _subscribers[topic].add(q)

    _logger.info("[event_bus] SSE subscribe topic=%s (total=%d)",
                 topic, len(_subscribers.get(topic, set())))

    async def generator() -> AsyncIterator[str]:
        try:
            yield f"data: {json.dumps({'type': 'connected', 'topic': topic})}\n\n"
            while True:
                if await request.is_disconnected():
                    break
                try:
                    event = await asyncio.wait_for(q.get(), timeout=15.0)
                    yield f"data: {safe_json_dumps(event)}\n\n"
                except asyncio.TimeoutError:
                    # heartbeat keepalive
                    yield f": keepalive {int(time.time())}\n\n"
        finally:
            async with _subscribers_lock:
                _subscribers.get(topic, set()).discard(q)
            _logger.info("[event_bus] SSE unsubscribe topic=%s", topic)

    return StreamingResponse(
        generator(),
        media_type="text/event-stream",
        headers={
            "Cache-Control":     "no-cache",
            "X-Accel-Buffering": "no",
            "Connection":        "keep-alive",
        },
    )


@router.get("/bus/status", summary="Diagnostica Event Bus")
async def bus_status():
    """Restituisce il numero di subscriber attivi per topic."""
    async with _subscribers_lock:
        status = {t: len(qs) for t, qs in _subscribers.items()}
    total = sum(status.values())
    return {
        "status":     "ok",
        "component":  "event_bus",
        "topics":     status,
        "total_subscribers": total,
        "builtin_topics": sorted(BUILTIN_TOPICS),
    }