dify/api/services/trigger/subscription_binding_service.py

"""Publish plugin subscription bindings and expose detached subscribers to workers."""

from collections.abc import Iterable, Mapping, Sequence
from dataclasses import dataclass
from typing import Any, Protocol


@dataclass(frozen=True, slots=True)
class TriggerSubscriptionBinding:
    node_id: str
    provider_id: str
    event_name: str
    subscription_id: str


@dataclass(frozen=True, slots=True)
class TriggerSubscriber:
    id: str
    tenant_id: str
    app_id: str
    node_id: str


class TriggerSubscriptionBindingQuery(Protocol):
    def get_subscriber_triggers(
        self, *, tenant_id: str, subscription_id: str, event_name: str
    ) -> tuple[TriggerSubscriber, ...]: ...


class TriggerSubscriptionBindingWriter(Protocol):
    def __call__(self, *, tenant_id: str, app_id: str, bindings: Sequence[TriggerSubscriptionBinding]) -> None: ...


class TriggerSubscriptionBindingService:
    MAX_PLUGIN_TRIGGER_NODES_PER_WORKFLOW = 5

    def __init__(self, *, sync_bindings: TriggerSubscriptionBindingWriter) -> None:
        self._sync_bindings = sync_bindings

    def sync(self, *, tenant_id: str, app_id: str, nodes: Iterable[tuple[str, Mapping[str, Any]]]) -> None:
        """Sync subscribed plugin nodes using the publisher's transaction-bound writer."""
        bindings = [
            TriggerSubscriptionBinding(
                node_id=node_id,
                provider_id=node.get("provider_id", ""),
                event_name=node.get("event_name", ""),
                subscription_id=subscription_id,
            )
            for node_id, node in nodes
            if (subscription_id := node.get("subscription_id", ""))
        ]
        if len(bindings) > self.MAX_PLUGIN_TRIGGER_NODES_PER_WORKFLOW:
            raise ValueError(
                "Workflow exceeds maximum plugin trigger node limit. "
                f"Found {len(bindings)} plugin trigger nodes, "
                f"maximum allowed is {self.MAX_PLUGIN_TRIGGER_NODES_PER_WORKFLOW}"
            )
        self._sync_bindings(tenant_id=tenant_id, app_id=app_id, bindings=bindings)