webmail-client/backend/app/events.py

# INTERFACE VERSION: 1
"""EventBus — Redis pub/sub fan-out + in-process durable consumers (SPEC §5.3, §8.3).

``publish`` fans out to the Redis channel ``evt:{event}`` (the WS hub relays these
to the owning user) and invokes any in-process subscribers. Payloads always
include ``user_id`` (SPEC §5.3). Only the event names in SPEC §8.3 are allowed.

NOTE(spec): the frozen contract also enqueues a durable arq job ``on_{event}`` for
out-of-process consumers (plugins). In M0 the only durable consumers (search index,
thread denormalization) run inline in the indexer per FR-30, so cross-process arq
fan-out is deferred to the plugin milestone (M3). The signatures here are frozen.
"""

from __future__ import annotations

import json
import logging
from collections.abc import Awaitable, Callable
from typing import Protocol

from app.redis_client import get_redis

logger = logging.getLogger(__name__)

Handler = Callable[[dict[str, object]], Awaitable[None]]

# The complete, allowed event catalog (SPEC §8.3). Plugin events (plugin.{name}.*)
# are validated separately by the plugin loader (M3).
ALLOWED_EVENTS: frozenset[str] = frozenset(
    {
        "message.new",
        "message.changed",
        "message.deleted",
        "message.sent",
        "flags.changed",
        "thread.updated",
        "sync.progress",
        "account.state_changed",
        "draft.saved",
        "outbox.state_changed",
        # Sync v2 §3: published on login and every WS connect; drives tier
        # promotion in the sync workers (§5). Payload: {user_id}.
        "user.active",
        # The user's sessions were revoked by an administrator (blocked or
        # deleted): open tabs drop to the sign-in screen. Payload: {user_id}.
        "user.signed_out",
    }
)


def channel_for(event: str) -> str:
    """Return the Redis pub/sub channel name for an event."""
    return f"evt:{event}"


class EventBusProtocol(Protocol):
    async def publish(self, event: str, payload: dict[str, object]) -> None: ...
    def subscribe(self, event: str, handler: Handler) -> None: ...


class EventBus:
    """Concrete EventBus backed by Redis pub/sub with in-process subscribers."""

    def __init__(self) -> None:
        self._handlers: dict[str, list[Handler]] = {}

    def subscribe(self, event: str, handler: Handler) -> None:
        """Register an in-process durable consumer for ``event``."""
        self._handlers.setdefault(event, []).append(handler)

    async def publish(self, event: str, payload: dict[str, object]) -> None:
        """Fan out ``event`` to Redis subscribers and in-process handlers."""
        if event not in ALLOWED_EVENTS and not event.startswith("plugin."):
            raise ValueError(f"unknown event {event!r} (see SPEC §8.3)")
        if "user_id" not in payload:
            raise ValueError("event payload must include user_id (SPEC §5.3)")

        message = json.dumps({"event": event, "payload": payload}, default=str)
        await get_redis().publish(channel_for(event), message)

        for handler in self._handlers.get(event, ()):
            try:
                await handler(payload)
            except Exception:  # a failing consumer must not break the publisher
                logger.exception("event handler failed for %s", event)


_bus: EventBus | None = None


def get_event_bus() -> EventBus:
    """Return the process-wide EventBus singleton."""
    global _bus
    if _bus is None:
        _bus = EventBus()
    return _bus