leocrm/app/core/trigger_dispatcher.py

"""Trigger dispatcher — routes events to matching automation definitions.

This module provides the **generic** bridge between the EventBus and the
automation execution engine.  It replaces the previous hard-coded event
handler stubs (``on_contact_created`` etc.) with a single wildcard
subscriber that queries the database for matching
``AutomationDefinition`` rows and dispatches them through the common
``run_automation`` execution core.

Two event categories are supported:

1. **Domain events** (durable, via Outbox → Worker → EventBus)
   - trigger_type = ``"event"``
   - trigger_config = ``{"event_name": "contact.created"}``
   - Any event published to the EventBus (either directly or via the
     outbox processor) can trigger an automation.

2. **UI events** (ephemeral, via WebSocket → EventBus only)
   - trigger_type = ``"ui"``
   - trigger_config = ``{"event_name": "ui.contact_selected"}``
   - UI events are **never** written to the outbox.  They flow directly
     from the frontend WebSocket handler through the EventBus to this
     dispatcher.

All four trigger types (event, ui, schedule, manual) converge on the
same ``run_automation`` execution engine, ensuring consistent condition
evaluation, action execution, and run logging.
"""

from __future__ import annotations

import logging
from typing import TYPE_CHECKING, Any

from sqlalchemy import select

if TYPE_CHECKING:
    from app.core.event_bus import EventBus

logger = logging.getLogger(__name__)

# Prefix that distinguishes ephemeral UI events from domain events.
# UI events must NEVER be enqueued into the transactional outbox.
_UI_EVENT_PREFIX = "ui."


class TriggerDispatcher:
    """Generic event-to-automation dispatcher.

    Subscribes to the ``*`` wildcard on the EventBus so that **every**
    event — domain or UI — is evaluated for matching automations.

    The dispatcher is stateless after construction; it queries the
    database on each event to find matching ``AutomationDefinition``
    rows.  This is intentionally generic: no hard-coded event list is
    maintained, and any registered outbox event can trigger an
    automation.
    """

    def __init__(self, event_bus: EventBus) -> None:
        self._event_bus = event_bus
        self._handler: Any = None

    def register(self) -> None:
        """Subscribe the wildcard handler on the event bus."""
        self._handler = self._on_event
        self._event_bus.subscribe("*", self._handler)
        logger.info("TriggerDispatcher registered — listening to all events")

    def unregister(self) -> None:
        """Unsubscribe from the event bus (idempotent)."""
        if self._handler is not None:
            self._event_bus.unsubscribe("*", self._handler)
            self._handler = None

    async def _on_event(self, payload: dict[str, Any]) -> None:
        """Wildcard handler invoked for every EventBus event.

        Determines the event name from the payload envelope, classifies
        it as domain or UI, queries matching automation definitions, and
        dispatches each through ``run_automation``.
        """
        event_name: str = payload.get("event_name", "")
        if not event_name:
            logger.debug("TriggerDispatcher: event without event_name, skipping")
            return

        is_ui_event = event_name.startswith(_UI_EVENT_PREFIX)
        trigger_type = "ui" if is_ui_event else "event"

        logger.debug(
            "TriggerDispatcher: evaluating event '%s' (trigger_type=%s)",
            event_name,
            trigger_type,
        )

        try:
            await self._dispatch_matching_automations(
                event_name=event_name,
                trigger_type=trigger_type,
                payload=payload,
            )
            # F-PROACTIVE: Also check for matching agent definitions on context/UI events
            if is_ui_event or event_name.startswith("context."):
                await self._dispatch_matching_agents(
                    event_name=event_name,
                    trigger_type=trigger_type,
                    payload=payload,
                )
        except Exception:
            logger.exception(
                "TriggerDispatcher: error dispatching event '%s'", event_name
            )

    async def _dispatch_matching_automations(
        self,
        event_name: str,
        trigger_type: str,
        payload: dict[str, Any],
    ) -> None:
        """Query DB for active automations matching *event_name* and dispatch."""
        from app.core.db import get_session_factory
        from app.plugins.builtins.contracts import get_contract
        automation_contract = get_contract("automation")
        AutomationDefinition = automation_contract.Automation  # noqa: N806
        if automation_contract is None:
            logger.debug("Automation plugin not available — trigger skipped")
            return

        factory = get_session_factory()
        tenant_id = payload.get("tenant_id")

        async with factory() as db:
            query = (
                select(AutomationDefinition)
                .where(AutomationDefinition.is_active.is_(True))
                .where(AutomationDefinition.trigger_type == trigger_type)
            )
            if tenant_id is not None:
                query = query.where(
                    AutomationDefinition.tenant_id == tenant_id
                )

            result = await db.execute(query)
            automations = list(result.scalars().all())

        if not automations:
            logger.debug(
                "TriggerDispatcher: no active automations for event '%s' (type=%s)",
                event_name,
                trigger_type,
            )
            return

        for automation in automations:
            config = automation.trigger_config or {}
            configured_event = config.get("event_name", "")
            if configured_event != event_name:
                continue

            logger.info(
                "TriggerDispatcher: dispatching automation '%s' (%s) for event '%s'",
                automation.name,
                automation.id,
                event_name,
            )

            await self._enqueue_automation(
                automation_id=str(automation.id),
                trigger_type=trigger_type,
                trigger_data=payload,
            )

    async def _dispatch_matching_agents(
        self,
        event_name: str,
        trigger_type: str,
        payload: dict[str, Any],
    ) -> None:
        """F-PROACTIVE: Query DB for active agents matching *event_name* and dispatch.

        Checks AgentDefinition.trigger_config for matching context/UI events.
        If match found, creates an AgentRun and dispatches via run_agent.
        """
        from app.core.db import get_session_factory
        from app.plugins.builtins.contracts import get_contract

        automation_contract = get_contract("automation")
        if automation_contract is None:
            return

        AgentDefinition = automation_contract.AgentDefinition  # noqa: N806
        if AgentDefinition is None:
            return

        factory = get_session_factory()
        tenant_id = payload.get("tenant_id")

        async with factory() as db:
            query = (
                select(AgentDefinition)
                .where(AgentDefinition.is_active.is_(True))
                .where(AgentDefinition.mode == "proactive")
            )
            if tenant_id is not None:
                query = query.where(AgentDefinition.tenant_id == tenant_id)

            result = await db.execute(query)
            agents = list(result.scalars().all())

        if not agents:
            return

        for agent in agents:
            config = agent.trigger_config or {}
            configured_event = config.get("event_name", "")
            if configured_event != event_name:
                continue

            logger.info(
                "TriggerDispatcher: dispatching agent '%s' (%s) for event '%s'",
                agent.name,
                agent.id,
                event_name,
            )

            try:
                await automation_contract.run_agent(
                    ctx={},
                    agent_id=str(agent.id),
                    trigger_type=trigger_type,
                    trigger_data=payload,
                )
            except Exception:
                logger.exception(
                    "TriggerDispatcher: run_agent failed for agent_id=%s",
                    agent.id,
                )

    async def _enqueue_automation(
        self,
        automation_id: str,
        trigger_type: str,
        trigger_data: dict[str, Any],
    ) -> None:
        """Dispatch automation through the common execution core.

        Uses ``run_automation`` directly (in-process) for low latency.
        For production workloads with back-pressure, the caller may
        alternatively enqueue via ``enqueue_job``.
        """
        try:
            from app.plugins.builtins.contracts import get_contract
            automation_contract = get_contract("automation")
            if automation_contract is not None:
                await automation_contract.run_automation(
                    ctx={},
                    automation_id=automation_id,
                    trigger_type=trigger_type,
                    trigger_data=trigger_data,
                )
        except Exception:
            logger.exception(
                "TriggerDispatcher: run_automation failed for automation_id=%s",
                automation_id,
            )


# ── Module-level helpers ─────────────────────────────────────────────────────

_dispatcher: TriggerDispatcher | None = None


def get_trigger_dispatcher() -> TriggerDispatcher | None:
    """Return the singleton dispatcher, or ``None`` if not registered."""
    return _dispatcher


def register_trigger_dispatcher(event_bus: EventBus) -> TriggerDispatcher:
    """Create and register the trigger dispatcher on *event_bus*.

    Safe to call multiple times — subsequent calls are no-ops.
    """
    global _dispatcher
    if _dispatcher is not None:
        logger.debug("TriggerDispatcher already registered")
        return _dispatcher
    _dispatcher = TriggerDispatcher(event_bus)
    _dispatcher.register()
    return _dispatcher


def unregister_trigger_dispatcher() -> None:
    """Unregister and discard the singleton dispatcher."""
    global _dispatcher
    if _dispatcher is not None:
        _dispatcher.unregister()
        _dispatcher = None


def is_ui_event(event_name: str) -> bool:
    """Return ``True`` if *event_name* is an ephemeral UI event.

    UI events must never be written to the transactional outbox.
    """
    return event_name.startswith(_UI_EVENT_PREFIX)