dify/api/services/trigger/provider_loader.py

"""
Load trigger plugin metadata with a context-scoped, tenant-isolated cache.
"""

import logging
from collections.abc import Mapping
from threading import Lock
from typing import Any

import contexts
from core.plugin.entities.plugin_daemon import CredentialType, PluginTriggerProviderEntity
from core.plugin.impl.exc import PluginDaemonError, PluginNotFoundError
from core.plugin.impl.trigger import PluginTriggerClient
from core.trigger.entities.entities import Subscription
from models.provider_ids import TriggerProviderID
from models.utils.trigger_endpoint import generate_trigger_plugin_icon_url
from services.trigger.plugin_provider import PluginTriggerProvider

logger = logging.getLogger(__name__)


class PluginTriggerProviderLoader:
    """Share provider discovery between gateways without retaining request state on the loader."""

    def __init__(self, *, client: PluginTriggerClient) -> None:
        self._client = client

    def get_trigger_plugin_icon(self, tenant_id: str, provider_id: str) -> str:
        """
        Get the icon of a trigger plugin
        """
        manager = self._client
        provider: PluginTriggerProviderEntity = manager.fetch_trigger_provider(
            tenant_id=tenant_id, provider_id=TriggerProviderID(provider_id)
        )
        return generate_trigger_plugin_icon_url(tenant_id, provider.declaration.identity.icon)

    def list_plugin_trigger_providers(self, tenant_id: str) -> list[PluginTriggerProvider]:
        """
        List all plugin trigger providers for a tenant

        :param tenant_id: Tenant ID
        :return: List of trigger provider controllers
        """
        manager = self._client
        provider_entities = manager.fetch_trigger_providers(tenant_id)

        controllers: list[PluginTriggerProvider] = []
        for provider in provider_entities:
            try:
                controller = PluginTriggerProvider(
                    entity=provider.declaration,
                    plugin_id=provider.plugin_id,
                    plugin_unique_identifier=provider.plugin_unique_identifier,
                    provider_id=TriggerProviderID(provider.provider),
                    tenant_id=tenant_id,
                )
                controllers.append(controller)
            except Exception:
                logger.exception("Failed to load trigger provider %s", provider.plugin_id)

        return controllers

    def get_trigger_provider(self, tenant_id: str, provider_id: TriggerProviderID) -> PluginTriggerProvider:
        """
        Get a specific plugin trigger provider

        :param tenant_id: Tenant ID
        :param provider_id: Provider ID
        :return: Provider runtime for this workspace
        """
        # check if context is set
        try:
            contexts.plugin_trigger_providers.get()
        except LookupError:
            contexts.plugin_trigger_providers.set({})
            contexts.plugin_trigger_providers_lock.set(Lock())

        plugin_trigger_providers = contexts.plugin_trigger_providers.get()
        cache_key = (tenant_id, str(provider_id))
        if cache_key in plugin_trigger_providers:
            return plugin_trigger_providers[cache_key]

        with contexts.plugin_trigger_providers_lock.get():
            # double check
            plugin_trigger_providers = contexts.plugin_trigger_providers.get()
            if cache_key in plugin_trigger_providers:
                return plugin_trigger_providers[cache_key]

            try:
                manager = self._client
                provider = manager.fetch_trigger_provider(tenant_id, provider_id)

                if not provider:
                    raise ValueError(f"Trigger provider {provider_id} not found")

                controller = PluginTriggerProvider(
                    entity=provider.declaration,
                    plugin_id=provider.plugin_id,
                    plugin_unique_identifier=provider.plugin_unique_identifier,
                    provider_id=provider_id,
                    tenant_id=tenant_id,
                )
                plugin_trigger_providers[cache_key] = controller
                return controller
            except PluginNotFoundError as e:
                raise ValueError(f"Trigger provider {provider_id} not found") from e
            except PluginDaemonError as e:
                raise e
            except Exception as e:
                logger.exception("Failed to load trigger provider")
                raise e

    def subscribe_trigger(
        self,
        tenant_id: str,
        user_id: str,
        provider_id: TriggerProviderID,
        endpoint: str,
        parameters: Mapping[str, Any],
        credentials: Mapping[str, object],
        credential_type: CredentialType,
    ) -> Subscription:
        """
        Subscribe to a trigger (e.g., register webhook)

        :param tenant_id: Tenant ID
        :param user_id: User ID
        :param provider_id: Provider ID
        :param endpoint: Subscription endpoint
        :param parameters: Subscription parameters
        :param credentials: Provider credentials
        :param credential_type: Credential type
        :return: Subscription result
        """
        provider: PluginTriggerProvider = self.get_trigger_provider(tenant_id=tenant_id, provider_id=provider_id)
        return provider.subscribe_trigger(
            user_id=user_id,
            endpoint=endpoint,
            parameters=parameters,
            credentials=credentials,
            credential_type=credential_type,
        )