orchestrata_ultimate.postman.subscription_manager

Reference › orchestrata_ultimate › postman › subscription_manager

The subscription store for topic fan-out, SPEC-002 story FR-006 (REQ-002-FR6-001 to -012) and story US-005. SubscriptionManager implements ISubscriptionManager from interfaces and records which agent listens to which topic pattern. Postman owns one and its publish_topic uses match_subscribers, get_subscriptions and the module-level evaluate_filters to decide who receives a published payload. Topic definitions are in topic_registry.

subscribe(agent_id, topic_pattern, ttl_seconds, filters) checks that the pattern works with fnmatch (otherwise InvalidTopicPatternError), computes an optional expiry, and stores a TopicSubscription under the agent. unsubscribe removes one pattern and unsubscribe_all removes all of an agent's subscriptions, returning the count. get_subscribers lists the agents for an exact topic. match_subscribers(topic) returns each agent once whose unexpired subscription pattern matches the topic, wildcards included. get_subscriptions(agent_id) returns an agent's subscriptions, expired ones included. evaluate_filters(payload, filters) is a standalone function that tests a payload against a subscription's key-value predicates; an empty filter set accepts every payload.

At a glance

Class diagram

Dependency diagram

API

Classes

SubscriptionManager

Bases: ISubscriptionManager · Decorators: traces(spec='002', story='FR-006', reqs=['REQ-002-FR6-001', 'REQ-002-FR6-004', 'REQ-002-FR6-006', 'REQ-002-FR6-007', 'R… · Line 76

Concrete implementation of ISubscriptionManager.

Functions

Source

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

"""
Subscription Manager implementation.

Implements ISubscriptionManager interface for managing topic subscriptions.
"""

import fnmatch
import logging
from datetime import UTC, datetime, timedelta
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.postman.exceptions import (
    InvalidTopicPatternError,
    SubscriptionNotFoundError,
)
from orchestrata_ultimate.postman.interfaces import ISubscriptionManager
from orchestrata_ultimate.postman.types import TopicSubscription

logger = logging.getLogger(__name__)


@traces(spec="002", story="US-005", reqs=["REQ-002-US-005-001", "REQ-002-US-005-006"])
def evaluate_filters(payload: dict[str, Any], filters: dict[str, Any]) -> bool:
    """Evaluate whether *payload* satisfies a subscription's *filters*.

    Postman-side twin of ``gateway/pubsub.py::evaluate_filters`` (SPEC-014
    FR-009) — identical semantics, duplicated here rather than imported so
    ``postman`` does not depend on ``gateway`` (layering; see
    ``architecture_checker.py``). Keep the two in sync if either changes.

    An empty (or falsy) ``filters`` mapping matches every payload —
    subscriptions created without filters keep receiving everything on
    their topic, unchanged. Multiple filter keys combine with AND
    semantics: every key must match for the payload to pass. Two keys get
    richer semantics: ``subject`` is matched by fnmatch-style glob against
    ``payload["subject"]``, and ``volk_id`` matches if *either*
    ``payload["to"]`` or ``payload["from"]`` equals the filter value (CXP
    packets carry both). Every other key is matched by plain equality
    against the top-level payload field of the same name.

    Args:
        payload: The publish payload being considered for delivery.
        filters: The subscription's key-value filter predicates.

    Returns:
        True if payload matches every filter key (or filters is empty).
    """
    if not filters:
        return True
    for key, expected in filters.items():
        if key == "volk_id":
            if payload.get("to") != expected and payload.get("from") != expected:
                return False
        elif key == "subject":
            if not fnmatch.fnmatch(str(payload.get("subject", "")), str(expected)):
                return False
        elif payload.get(key) != expected:
            return False
    return True


@traces(
    spec="002",
    story="FR-006",
    reqs=[
        "REQ-002-FR6-001",
        "REQ-002-FR6-004",
        "REQ-002-FR6-006",
        "REQ-002-FR6-007",
        "REQ-002-FR6-008",
        "REQ-002-FR6-011",
        "REQ-002-FR6-012",
    ],
)
class SubscriptionManager(ISubscriptionManager):
    """Concrete implementation of ISubscriptionManager."""

    @traces(spec="002", story="FR-006", reqs=["REQ-002-FR6-001"])
    def __init__(self) -> None:
        # agent_id -> list of TopicSubscription
        self._subscriptions: dict[str, list[TopicSubscription]] = {}

    @traces(
        spec="002",
        story="FR-006",
        reqs=["REQ-002-FR6-006", "REQ-002-US-005-001", "REQ-002-US-005-006"],
    )
    async def subscribe(
        self,
        agent_id: str,
        topic_pattern: str,
        ttl_seconds: int | None = None,
        filters: dict[str, Any] | None = None,
    ) -> TopicSubscription:
        """Subscribe an agent to a topic pattern.

        Args:
            agent_id: Subscribing agent ID.
            topic_pattern: Topic pattern (may contain wildcards).
            ttl_seconds: Optional TTL for the subscription.
            filters: Optional key-value predicates evaluated (via
                ``evaluate_filters``) against a publish payload before
                delivery; omitted/empty means every payload on the topic
                is delivered.

        Returns:
            The created TopicSubscription.

        Raises:
            InvalidTopicPatternError: If the topic_pattern is invalid.
        """
        # Validate pattern syntax
        try:
            fnmatch.fnmatch("dummy", topic_pattern)
        except TypeError as e:
            raise InvalidTopicPatternError(
                f"Invalid topic pattern '{topic_pattern}': {e}",
                error_code="INVALID_PATTERN",
                context={"topic_pattern": topic_pattern},
            ) from e

        # Compute expiration
        if ttl_seconds is not None:
            ttl_expires_at = datetime.now(UTC) + timedelta(seconds=ttl_seconds)
        else:
            ttl_expires_at = None

        subscription = TopicSubscription(
            agent_id=agent_id,
            topic_pattern=topic_pattern,
            ttl_expires_at=ttl_expires_at,
            filters=filters or {},
            # created_at is set automatically by default_factory
        )

        self._subscriptions.setdefault(agent_id, []).append(subscription)
        return subscription

    @traces(spec="002", story="FR-006", reqs=["REQ-002-FR6-007"])
    async def unsubscribe(self, agent_id: str, topic_pattern: str) -> bool:
        """Unsubscribe an agent from a specific topic pattern.

        Args:
            agent_id: Agent ID.
            topic_pattern: Exact topic pattern used during subscription.

        Returns:
            True if the subscription was found and removed.

        Raises:
            SubscriptionNotFoundError: If no matching subscription exists.
        """
        if agent_id not in self._subscriptions:
            raise SubscriptionNotFoundError(
                f"No subscriptions found for agent '{agent_id}'.",
                error_code="AGENT_NOT_FOUND",
                context={"agent_id": agent_id, "topic_pattern": topic_pattern},
            )

        subs = self._subscriptions[agent_id]
        # Find subscription with exact pattern match
        for i, sub in enumerate(subs):
            if sub.topic_pattern == topic_pattern:
                # Remove it
                del subs[i]
                # Clean up empty list
                if not subs:
                    del self._subscriptions[agent_id]
                return True

        raise SubscriptionNotFoundError(
            f"Subscription for agent '{agent_id}' to pattern '{topic_pattern}' not found.",
            error_code="SUBSCRIPTION_NOT_FOUND",
            context={"agent_id": agent_id, "topic_pattern": topic_pattern},
        )

    @traces(spec="002", story="FR-006", reqs=["REQ-002-FR6-007"])
    async def unsubscribe_all(self, agent_id: str) -> int:
        """Unsubscribe an agent from all topics.

        Args:
            agent_id: Agent ID.

        Returns:
            Number of subscriptions removed.
        """
        if agent_id not in self._subscriptions:
            return 0

        count = len(self._subscriptions[agent_id])
        del self._subscriptions[agent_id]
        return count

    @traces(spec="002", story="FR-006", reqs=["REQ-002-FR6-012"])
    async def get_subscribers(self, topic: str) -> list[str]:
        """Get all subscribers for an exact topic.

        Only returns subscribers whose subscription pattern exactly equals the
        given topic and whose subscription is not expired.

        Args:
            topic: Exact topic name (no wildcard expansion).

        Returns:
            List of subscriber agent IDs.
        """
        subscribers: list[str] = []
        seen = set()

        for agent_id, subs in self._subscriptions.items():
            for sub in subs:
                if sub.topic_pattern == topic and not sub.is_expired():
                    if agent_id not in seen:
                        seen.add(agent_id)
                        subscribers.append(agent_id)
                    break  # only one subscription per agent needed for exact match
        return subscribers

    @traces(spec="002", story="US-002", reqs=["REQ-002-US-002-009"])
    async def get_subscriptions(self, agent_id: str) -> list[TopicSubscription]:
        """Get all subscriptions for an agent, including expired ones.

        Args:
            agent_id: Agent ID.

        Returns:
            List of TopicSubscription objects for the agent.
        """
        return self._subscriptions.get(agent_id, []).copy()

    @traces(spec="002", story="FR-006", reqs=["REQ-002-FR6-011"])
    async def match_subscribers(self, topic: str) -> list[str]:
        """Find all subscribers matching the topic (including wildcards).

        Returns agents whose subscription pattern matches the given topic,
        taking wildcards into account, and whose subscription is not expired.

        Args:
            topic: Topic to match against subscription patterns.

        Returns:
            List of subscriber agent IDs.
        """
        subscribers: list[str] = []
        seen = set()

        for agent_id, subs in self._subscriptions.items():
            for sub in subs:
                if sub.is_expired():
                    continue
                if fnmatch.fnmatch(topic, sub.topic_pattern):
                    if agent_id not in seen:
                        seen.add(agent_id)
                        subscribers.append(agent_id)
                    break  # each agent only once
        return subscribers