"""Author RAG — Outbound Webhook Delivery System. Authors can configure webhook URLs to receive real-time events when their chatbot has activity. Used for integrations with: - Zapier, Make.com, n8n - CRM systems (HubSpot, Salesforce) - Custom dashboards Supported events: new_conversation → visitor started chatting buy_intent_detected → high-purchase-signal query detected link_clicked → visitor clicked a buy link low_budget_warning → token budget below 20% Routes: GET /api/admin/{slug}/webhooks → list configured webhooks POST /api/admin/{slug}/webhooks → register new endpoint DELETE /api/admin/{slug}/webhooks/{id} → remove endpoint POST /api/admin/{slug}/webhooks/{id}/test → send test payload """ import hashlib import hmac import json from datetime import datetime, timezone import httpx import structlog from fastapi import APIRouter, Depends, HTTPException from pydantic import BaseModel, Field, HttpUrl from typing import Literal from app.dependencies import get_current_author from app.config import get_settings logger = structlog.get_logger(__name__) cfg = get_settings() router = APIRouter() # In-memory store (Phase 3: migrate to DB table) _webhooks: dict[str, list[dict]] = {} # author_id → list of webhook configs SUPPORTED_EVENTS = [ "new_conversation", "buy_intent_detected", "link_clicked", "low_budget_warning", "subscription_expiring", ] class WebhookCreate(BaseModel): url: HttpUrl events: list[str] = Field(..., min_length=1, description="List of event types to subscribe to") description: str | None = Field(None, max_length=200) secret: str | None = Field(None, max_length=255, description="Used to sign payloads — verify on your end") @router.get("/{slug}/webhooks") async def list_webhooks(slug: str, author=Depends(get_current_author)): if author.id != slug: raise HTTPException(403, "Forbidden") hooks = _webhooks.get(slug, []) # Don't expose the secret in responses return {"webhooks": [{**h, "secret": "***" if h.get("secret") else None} for h in hooks]} @router.post("/{slug}/webhooks", status_code=201) async def register_webhook( slug: str, body: WebhookCreate, author=Depends(get_current_author), ): if author.id != slug: raise HTTPException(403, "Forbidden") # Validate event types invalid = set(body.events) - set(SUPPORTED_EVENTS) if invalid: raise HTTPException(400, f"Unsupported events: {', '.join(invalid)}. Valid: {', '.join(SUPPORTED_EVENTS)}") if len(_webhooks.get(slug, [])) >= 5: raise HTTPException(400, "Maximum 5 webhooks per author. Remove one to add another.") from app.models.base import generate_uuid hook = { "id": generate_uuid()[:8], "url": str(body.url), "events": body.events, "description": body.description, "secret": body.secret, "is_active": True, "created_at": datetime.now(timezone.utc).isoformat(), "last_triggered_at": None, "failure_count": 0, } _webhooks.setdefault(slug, []).append(hook) return {**hook, "secret": "***" if hook.get("secret") else None} @router.delete("/{slug}/webhooks/{webhook_id}", status_code=200) async def remove_webhook( slug: str, webhook_id: str, author=Depends(get_current_author), ): if author.id != slug: raise HTTPException(403, "Forbidden") hooks = _webhooks.get(slug, []) before = len(hooks) _webhooks[slug] = [h for h in hooks if h["id"] != webhook_id] if len(_webhooks[slug]) == before: raise HTTPException(404, "Webhook not found") return {"message": "Webhook removed"} @router.post("/{slug}/webhooks/{webhook_id}/test") async def test_webhook( slug: str, webhook_id: str, author=Depends(get_current_author), ): """Send a test payload to verify the endpoint is reachable.""" if author.id != slug: raise HTTPException(403, "Forbidden") hook = next((h for h in _webhooks.get(slug, []) if h["id"] == webhook_id), None) if not hook: raise HTTPException(404, "Webhook not found") test_payload = { "event": "test", "author_id": slug, "timestamp": datetime.now(timezone.utc).isoformat(), "data": {"message": "This is a test webhook from ChatBot AI"}, } result = await _deliver(hook, test_payload) return {"success": result["ok"], "status_code": result.get("status_code"), "error": result.get("error")} # ── Internal Delivery ───────────────────────────────────────────────────────── def _sign_payload(secret: str, payload_bytes: bytes) -> str: """HMAC-SHA256 signature for webhook verification.""" return "sha256=" + hmac.new(secret.encode(), payload_bytes, hashlib.sha256).hexdigest() async def _deliver(hook: dict, payload: dict) -> dict: """Attempt to deliver a webhook payload with a 5-second timeout.""" payload_bytes = json.dumps(payload).encode() headers = { "Content-Type": "application/json", "X-ChatBotAI-Event": payload.get("event", "unknown"), "X-ChatBotAI-Timestamp": datetime.now(timezone.utc).isoformat(), } if hook.get("secret"): headers["X-ChatBotAI-Signature"] = _sign_payload(hook["secret"], payload_bytes) try: async with httpx.AsyncClient(timeout=5.0) as client: resp = await client.post(hook["url"], content=payload_bytes, headers=headers) ok = resp.status_code < 400 hook["last_triggered_at"] = datetime.now(timezone.utc).isoformat() if not ok: hook["failure_count"] = hook.get("failure_count", 0) + 1 # Disable webhook after 10 consecutive failures if hook["failure_count"] >= 10: hook["is_active"] = False logger.warning("Webhook auto-disabled after 10 failures", url=hook["url"]) else: hook["failure_count"] = 0 return {"ok": ok, "status_code": resp.status_code} except Exception as e: hook["failure_count"] = hook.get("failure_count", 0) + 1 logger.error("Webhook delivery failed", url=hook["url"], error=str(e)) return {"ok": False, "error": str(e)} async def trigger_event(author_id: str, event_type: str, data: dict) -> None: """Called by the chat pipeline to fire webhooks for a specific event type.""" hooks = _webhooks.get(author_id, []) payload = { "event": event_type, "author_id": author_id, "timestamp": datetime.now(timezone.utc).isoformat(), "data": data, } for hook in hooks: if hook.get("is_active") and event_type in hook.get("events", []): await _deliver(hook, payload)