api/routes/webhooks/webhook_bridge.py

import asyncio
import logging
from datetime import datetime, timezone
from typing import Any, Dict, List, Optional
from pydantic import BaseModel, Field
from core.integration_registry import IntegrationRegistry
from core.circuit_breaker import circuit_breaker
from core.universal_communication_bridge import UniversalCommunicationBridge
from sqlalchemy.orm import Session

logger = logging.getLogger(__name__)

class UnifiedIncomingMessage(BaseModel):
    """Standardized incoming message from any communication platform"""
    platform: str
    sender_id: str
    recipient_id: str
    text: str
    timestamp: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
    thread_id: Optional[str] = None
    metadata: Dict[str, Any] = Field(default_factory=dict)
    raw_payload: Dict[str, Any] = Field(default_factory=dict)

class WebhookBridge:
    """
    Standardized bridge for incoming webhooks dispatching events 
    to the IntegrationRegistry and ChatOrchestrator.
    """
    
    def __init__(self):
        self._orchestrator = None

    def _get_orchestrator(self):
        """Lazy logic to handle circular dependencies."""
        if self._orchestrator is None:
            try:
                from integrations.chat_orchestrator import ChatOrchestrator
                self._orchestrator = ChatOrchestrator(workspace_id="default")
            except Exception as e:
                logger.error(f"Failed to initialize ChatOrchestrator: {e}")
        return self._orchestrator

    async def process_event(
        self, 
        platform: str, 
        tenant_id: str, 
        data: Dict[str, Any], 
        registry: IntegrationRegistry,
        db: Session
    ) -> Dict[str, Any]:
        """Process an incoming platform event via the UniversalCommunicationBridge."""
        logger.info(f"Webhook Bridge: Dispatching event from {platform} for tenant {tenant_id}")
        
        # 0. Circuit Breaker Check
        cb_key = f"{platform}:{tenant_id}"
        if not await circuit_breaker.is_enabled(cb_key):
            logger.warning(f"Webhook Bridge: Circuit breaker OPEN for {cb_key}. Ignoring event.")
            return {"status": "ignored", "reason": "circuit_breaker_open"}

        try:
            # 1. Use UniversalCommunicationBridge for standardized normalization
            ucb = UniversalCommunicationBridge(db)
            ucb_result = await ucb.receive_message(
                tenant_id=tenant_id,
                platform=platform,
                payload=data
            )
            
            if not ucb_result:
                return {"status": "ignored", "reason": "ucb_ignored_or_error"}
                
            # Handle standardized interactions (buttons, etc.)
            if ucb_result.get("type") == "interaction":
                return {
                    "status": "success",
                    "processed": True,
                    "type": "interaction",
                    "result": ucb_result.get("result")
                }

            # Handle standard text messages
            if ucb_result.get("type") != "message":
                return {"status": "ignored", "reason": "unsupported_ucb_type"}

            unified_msg = ucb_result["message"]
            text_content = unified_msg.content
            sender_id = unified_msg.sender_id

            # 2. Command Handling (e.g., /run)
            if text_content.startswith('/'):
                # Convert to local model for compatibility with _handle_command
                compat_msg = UnifiedIncomingMessage(
                    platform=platform,
                    sender_id=sender_id,
                    recipient_id=unified_msg.recipient_id or "bot",
                    text=text_content,
                    thread_id=unified_msg.thread_id,
                    metadata=unified_msg.metadata or {}
                )
                return await self._handle_command(compat_msg, tenant_id, registry)

            # 3. Chat Orchestrator Integration
            orchestrator = self._get_orchestrator()
            if not orchestrator:
                return {"status": "error", "message": "ChatOrchestrator unavailable"}

            session_id = f"{platform}_{sender_id}"
            response = await orchestrator.process_chat_message(
                message=text_content,
                session_id=session_id,
                user_id=f"ext_{sender_id}",
                context={
                    "platform": platform,
                    "tenant_id": tenant_id,
                    "sender_id": sender_id,
                    "recipient_id": unified_msg.recipient_id,
                    "thread_id": unified_msg.thread_id
                }
            )

            # 4. Auto-Response Dispatch (Optional based on response)
            if response and response.get("message"):
                # Use UCB for standard response
                ucb = UniversalCommunicationBridge(db)
                await ucb.send_message(
                    tenant_id=tenant_id,
                    platform=platform,
                    target_id=sender_id,
                    content=response["message"],
                    metadata={"thread_ts": unified_msg.thread_id}
                )

            return {
                "status": "success",
                "processed": True,
                "orchestrator_response": response
            }

        except Exception as e:
            logger.error(f"Webhook Bridge Error ({platform}): {e}")
            return {"status": "error", "message": str(e)}

    async def _handle_command(self, msg: UnifiedIncomingMessage, tenant_id: str, registry: IntegrationRegistry) -> Dict[str, Any]:
        """Handle platform commands (e.g., /run) via Registry."""
        parts = msg.text[1:].split(' ', 2)
        command = parts[0].lower()
        
        if command == "run" and len(parts) > 1:
            agent_name = parts[1]
            task_input = parts[2] if len(parts) > 2 else "Run Default"
            
            # Use registry to trigger agent task (simulated execution)
            # In production, we'd use core.agent_routes.execute_agent_task
            return {"status": "command_triggered", "command": "run", "agent": agent_name}
            
        return {"status": "command_ignored", "command": command}

# Global instance
webhook_bridge = WebhookBridge()