leocrm/app/core/webhook_dispatcher.py

"""Webhook dispatcher — subscribes to the event bus and dispatches to matching webhooks."""

from __future__ import annotations

import asyncio
import logging
import uuid
from typing import Any

from sqlalchemy import cast, select
from sqlalchemy.dialects.postgresql import JSONB

from app.core.db import get_session_factory
from app.core.event_bus import EventBus, get_event_bus
from app.models.webhook import Webhook
from app.services.webhook_service import send_webhook

logger = logging.getLogger(__name__)


async def _dispatch_event(payload: dict[str, Any]) -> None:
    """Handle an event from the event bus: find matching webhooks and dispatch.

    The payload is expected to contain:
        - event_name: str
        - tenant_id: str (UUID)
        - data: dict (the actual event data)
    """
    event_name = payload.get("event_name", "")
    tenant_id_str = payload.get("tenant_id", "")
    data = payload.get("data", {})

    if not event_name or not tenant_id_str:
        logger.warning("webhook_dispatcher: missing event_name or tenant_id in payload")
        return

    try:
        tenant_id = uuid.UUID(tenant_id_str)
    except (ValueError, TypeError):
        logger.warning(f"webhook_dispatcher: invalid tenant_id: {tenant_id_str}")
        return

    # Find active webhooks for this tenant that subscribe to this event
    session_factory = get_session_factory()
    async with session_factory() as db:
        # Set tenant context for RLS
        from app.core.db import set_tenant_context
        await set_tenant_context(db, tenant_id)
        stmt = select(Webhook).where(
            Webhook.tenant_id == tenant_id,
            Webhook.is_active == True,  # noqa: E712
            # Webhook.events is a JSONB array column (NOT a relationship):
            # events @> '["<event_name>"]' — JSONB containment instead of
            # the invalid relationship .any() call that crashed every event
            # with "Neither 'AnnotatedColumn' nor 'Comparator' object has an
            # attribute 'any'" (158 failed outbox events in production).
            cast(Webhook.events, JSONB).contains([event_name]),
        )
        result = await db.execute(stmt)
        webhooks = list(result.scalars().all())

    if not webhooks:
        return

    # Dispatch to all matching webhooks concurrently
    tasks = []
    for webhook in webhooks:
        tasks.append(_dispatch_single(webhook, event_name, data))

    if tasks:
        await asyncio.gather(*tasks, return_exceptions=True)


async def _dispatch_single(
    webhook: Webhook,
    event_name: str,
    data: dict[str, Any],
) -> None:
    """Send a webhook and log the result. Raises on failure (P1.5 fix).

    Previously errors were swallowed, causing the outbox to mark events
    as 'published' even when webhook delivery failed.
    """
    try:
        result = await send_webhook(webhook, event_name, data)
        if result["success"]:
            logger.info(
                f"Webhook {webhook.id} sent successfully to {webhook.url} "
                f"for event {event_name} (status={result['status_code']})"
            )
        else:
            logger.warning(
                f"Webhook {webhook.id} failed for {webhook.url} "
                f"event {event_name}: {result.get('error')}"
            )
            raise RuntimeError(f"Webhook {webhook.id} failed: {result.get('error')}")
    except Exception as exc:
        logger.error(
            f"Webhook {webhook.id} dispatch error for {webhook.url}: {exc}"
        )
        raise  # Re-raise so outbox can retry (P1.5 fix)


def register_webhook_event_handlers(event_bus: EventBus | None = None) -> None:
    """Register webhook dispatcher on the global event bus.

    Subscribes to all events via the wildcard '*' handler and filters
    by event name internally. Should be called during application startup.
    """
    bus = event_bus or get_event_bus()
    bus.subscribe("*", _dispatch_event)
    logger.info("Webhook dispatcher registered on event bus (wildcard '*' handler)")