orchestrata_ultimate.kernel.event_bus

Reference › orchestrata_ultimate › kernel › event_bus

The system event bus of the kernel. EventBus delivers a SystemEvent to handlers in the same process and, in the same call, publishes it to Redis so other processes see it and so it can be replayed later. kernel owns one and creates it lazily; the gateway server also imports it. The module sits next to the curated kernel narrative.

emit calls _call_in_process_handlers first and then _publish_to_redis. subscribe registers a handler against a topic pattern with an optional subscriber id, a time-to-live and a filter function, and returns an EventSubscription; unsubscribe removes it by id. _find_matching_handlers skips expired subscriptions, matches the event type to the pattern with fnmatch, applies the filter, and handlers run under a per-handler timeout (5 seconds by default). _publish_to_redis publishes the serialised event to a channel named from the prefix orchestrata:events and the event type, and appends the fields to a Redis stream with an approximate length cap (100000 entries by default). replay reads the stream with an optional time window and count, filters by topic pattern and calls the handler for each match, returning the number delivered.

Redis host and port come from REDIS_HOST and REDIS_PORT at first use when not passed in, with no hardcoded default, and the client comes from the shared core.redis_factory. A Redis error during publish is logged and swallowed, so in-process delivery still happens; a Redis error during replay is logged and re-raised.

At a glance

Class diagram

Dependency diagram

API

Classes

EventBus

Line 35

Dual-layer EventBus with in-process handlers and Redis pub/sub.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/kernel/event_bus.py

# mypy: ignore-errors
"""Dual-layer (in-process + Redis) EventBus implementation.

Spec: 001-kernel (addendum) — EventBus
"""

import asyncio
import fnmatch
import json
import logging
import os
from collections.abc import Awaitable, Callable
from datetime import UTC, datetime, timedelta
from typing import TYPE_CHECKING, Any, cast
from uuid import uuid4

import redis.asyncio as redis
from redis.exceptions import RedisError

if TYPE_CHECKING:
    from redis.typing import EncodableT, FieldT

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.protocols import EventSubscription, SystemEvent

logger = logging.getLogger(__name__)

# Redis stream that persists every emitted SystemEvent (XADD'd by
# _publish_to_redis). Exposed as a module constant so consumers outside this
# module (e.g. the gateway's kernel-events forwarder) never duplicate the
# literal string.
EVENTS_STREAM_KEY = "orchestrata:stream:events"


class EventBus:
    """Dual-layer EventBus with in-process handlers and Redis pub/sub.

    Implements IEventBus protocol. Events are delivered to in-process
    subscribers via asyncio.gather, then published to Redis channels
    and streams for cross-process distribution and persistence.
    """

    def __init__(
        self,
        redis_host: str | None = None,
        redis_port: int | None = None,
        redis_password: str | None = None,
        redis_db: int = 0,
        stream_key: str = EVENTS_STREAM_KEY,
        channel_prefix: str = "orchestrata:events",
        handler_timeout: float = 5.0,
        stream_maxlen: int = 100000,
        redis_client: redis.Redis | None = None,
    ) -> None:
        """Initialize EventBus with Redis connection parameters.

        Args:
            redis_host: Redis host; ``None`` resolves ``REDIS_HOST`` from the
                environment at first use — no hardcoded fallback (infra
                endpoints come from ``.env``, never silent defaults).
            redis_port: Redis port; ``None`` resolves ``REDIS_PORT`` likewise.
            redis_client: Pre-built async client injected directly (the
                production default still lazily builds one from
                ``redis_host``/``redis_port`` on first use when this is
                omitted). Exposed so tests can point the EventBus at a
                fail-fast client without paying redis-py's default
                connection-retry backoff.
        """
        self._redis_host = redis_host
        self._redis_port = redis_port
        self._redis_password = redis_password
        self._redis_db = redis_db
        self._stream_key = stream_key
        self._channel_prefix = channel_prefix
        self._handler_timeout = handler_timeout
        self._stream_maxlen = stream_maxlen

        self._redis: redis.Redis | None = redis_client
        self._subscriptions: dict[str, dict] = {}
        self._lock = asyncio.Lock()

    @traces(spec="001", story="US-010", reqs=["REQ-001-US-010-012"])
    async def _get_redis(self) -> redis.Redis:
        """Lazy Redis client initialization."""
        if self._redis is None:
            from orchestrata_ultimate.core.redis_factory import create_async_redis

            # Resolve infra endpoints at call time, no hardcoded fallback:
            # REDIS_HOST/REDIS_PORT come from .env (validate_env.py preflights
            # them), and a silent localhost default would point at nothing.
            # os.environ.get(...) (not the bare subscript) + an explicit
            # raise keeps the "fail loudly when unset" intent without a
            # KeyError whose message doesn't say which var or why.
            if self._redis_host is not None:
                host = self._redis_host
            else:
                env_host = os.environ.get("REDIS_HOST")
                if env_host is None:
                    raise RuntimeError(
                        "REDIS_HOST is not set (.env) and no explicit host "
                        "was passed to EventBus"
                    )
                host = env_host
            if self._redis_port is not None:
                port = self._redis_port
            else:
                env_port = os.environ.get("REDIS_PORT")
                if env_port is None:
                    raise RuntimeError(
                        "REDIS_PORT is not set (.env) and no explicit port "
                        "was passed to EventBus"
                    )
                port = int(env_port)
            self._redis = create_async_redis(
                host=host,
                port=port,
                password=self._redis_password,
                db=self._redis_db,
            )
        return self._redis

    async def close(self) -> None:
        """Close the Redis connection."""
        if self._redis is not None:
            await self._redis.aclose()
            self._redis = None

    async def emit(self, event: SystemEvent) -> None:
        """Emit an event to in-process handlers and Redis.

        Delivers to matching in-process subscribers first, then publishes
        to Redis for cross-process distribution and persistence.
        """
        await self._call_in_process_handlers(event)
        await self._publish_to_redis(event)

    @traces(spec="015", story="US-011", reqs=["REQ-015-149"])
    async def subscribe(
        self,
        topic_pattern: str,
        handler: Callable[[SystemEvent], Awaitable[None]],
        *,
        subscriber_id: str | None = None,
        ttl_seconds: int | None = None,
        filter_fn: Callable[[SystemEvent], bool] | None = None,
    ) -> EventSubscription:
        """Subscribe to events matching a topic pattern."""
        if subscriber_id is None:
            subscriber_id = str(uuid4())

        subscription_id = str(uuid4())
        created_at = datetime.now(UTC)

        expires_at = None
        if ttl_seconds is not None:
            expires_at = created_at + timedelta(seconds=ttl_seconds)

        async with self._lock:
            self._subscriptions[subscription_id] = {
                "subscription_id": subscription_id,
                "topic_pattern": topic_pattern,
                "subscriber_id": subscriber_id,
                "handler": handler,
                "filter_fn": filter_fn,
                "created_at": created_at,
                "expires_at": expires_at,
                "event_count": 0,
            }

        return EventSubscription(
            id=subscription_id,
            topic_pattern=topic_pattern,
            subscriber_id=subscriber_id,
            created_at=created_at,
            expires_at=expires_at,
            event_count=0,
        )

    async def unsubscribe(self, subscription_id: str) -> bool:
        """Unsubscribe by subscription ID."""
        async with self._lock:
            if subscription_id in self._subscriptions:
                del self._subscriptions[subscription_id]
                return True
            return False

    async def replay(
        self,
        topic_pattern: str,
        handler: Callable[[SystemEvent], Awaitable[None]],
        *,
        since: datetime | None = None,
        until: datetime | None = None,
        count: int | None = None,
    ) -> int:
        """Replay historical events matching a topic pattern from Redis stream."""
        try:
            redis_client = await self._get_redis()

            start = "-" if since is None else str(int(since.timestamp() * 1000))
            end = "+" if until is None else str(int(until.timestamp() * 1000))

            entries = await redis_client.xrange(
                self._stream_key, min=start, max=end, count=count
            )

            delivered = 0
            for _entry_id, data in entries:
                event = self._deserialize_stream_entry(data)
                if fnmatch.fnmatch(event.type, topic_pattern):
                    await handler(event)
                    delivered += 1

            return delivered
        except RedisError:
            logger.error("Redis replay failed", exc_info=True)
            raise

    def subscriber_count(self, topic_pattern: str | None = None) -> int:
        """Count subscribers matching the pattern, or all if None."""
        count = 0
        current_time = datetime.now(UTC)
        for sub_data in self._subscriptions.values():
            expires_at = sub_data["expires_at"]
            if expires_at is not None and current_time >= expires_at:
                continue
            if topic_pattern is None:
                count += 1
            elif fnmatch.fnmatch(sub_data["topic_pattern"], topic_pattern):
                count += 1
        return count

    # ------------------------------------------------------------------
    # Internal methods
    # ------------------------------------------------------------------

    @traces(spec="015", story="US-011", reqs=["REQ-015-155"])
    async def _call_in_process_handlers(self, event: SystemEvent) -> None:
        """Call matching in-process handlers with timeout."""
        matching = await self._find_matching_handlers(event)
        if not matching:
            return

        tasks = [sub_data["handler"](event) for sub_data in matching]
        try:
            results = await asyncio.wait_for(
                asyncio.gather(*tasks, return_exceptions=True),
                timeout=self._handler_timeout,
            )
            for i, result in enumerate(results):
                if isinstance(result, Exception):
                    logger.warning(
                        "Subscriber %s failed: %s",
                        matching[i]["subscriber_id"],
                        result,
                    )
                else:
                    async with self._lock:
                        sub_id = matching[i]["subscription_id"]
                        if sub_id in self._subscriptions:
                            self._subscriptions[sub_id]["event_count"] += 1
        except TimeoutError:
            logger.warning(
                "Timeout (%ss) delivering event %s to %d handlers",
                self._handler_timeout,
                event.type,
                len(matching),
            )

    @traces(spec="015", story="US-011", reqs=["REQ-015-147"])
    async def _find_matching_handlers(self, event: SystemEvent) -> list[dict]:
        """Find all subscription handlers that match the event topic."""
        matching: list[dict] = []
        current_time = datetime.now(UTC)

        async with self._lock:
            for sub_data in self._subscriptions.values():
                expires_at = sub_data["expires_at"]
                if expires_at is not None and current_time >= expires_at:
                    continue

                if not fnmatch.fnmatch(event.type, sub_data["topic_pattern"]):
                    continue

                filter_fn = sub_data["filter_fn"]
                if filter_fn is not None and not filter_fn(event):
                    continue

                matching.append(sub_data)

        return matching

    async def _publish_to_redis(self, event: SystemEvent) -> None:
        """Publish event to Redis channel and stream."""
        try:
            redis_client = await self._get_redis()

            channel = f"{self._channel_prefix}:{event.type}"
            event_json = self._serialize_event(event)
            await redis_client.publish(channel, event_json)

            stream_data = {
                "id": event.id,
                "type": event.type,
                "source": event.source,
                "timestamp": event.timestamp.isoformat(),
                "payload": json.dumps(event.payload),
                "correlation_id": event.correlation_id or "",
                "severity": event.severity,
            }
            # `dict[str, str]` isn't assignable to `Dict[FieldT, EncodableT]`
            # despite `str` being a member of both unions: `Dict`'s type
            # parameters are invariant, so pyright requires an exact type
            # match rather than accepting a narrower concrete dict
            # (precedent: gateway/ring_buffer.py).
            await redis_client.xadd(
                self._stream_key,
                cast("dict[FieldT, EncodableT]", stream_data),
                maxlen=self._stream_maxlen,
                approximate=True,
            )
        except RedisError:
            logger.error("Redis publish failed for event %s", event.type, exc_info=True)

    def _serialize_event(self, event: SystemEvent) -> str:
        """Serialize SystemEvent to JSON string."""
        return json.dumps(event.to_dict())

    def _deserialize_event(self, data: str) -> SystemEvent:
        """Deserialize JSON string to SystemEvent."""
        return SystemEvent.from_dict(json.loads(data))

    def _deserialize_stream_entry(self, data: dict[str, Any]) -> SystemEvent:
        """Deserialize a Redis stream entry to SystemEvent."""
        try:
            payload = json.loads(data.get("payload", "{}"))
        except json.JSONDecodeError, TypeError:
            payload = {}

        try:
            timestamp = datetime.fromisoformat(data["timestamp"])
        except ValueError, KeyError:
            timestamp = datetime.now(UTC)

        return SystemEvent(
            id=data.get("id", str(uuid4())),
            type=data.get("type", ""),
            source=data.get("source", ""),
            timestamp=timestamp,
            payload=payload,
            correlation_id=data.get("correlation_id") or None,
            severity=data.get("severity", "info"),
        )