api_webhooks.py

"""
DocuLens — Webhook management and delivery.

Allows users to register webhook URLs that receive POST notifications
when extraction or validation completes.
"""

import os
import json
import hmac
import hashlib
import logging
import time
from typing import Optional
from uuid import uuid4

import requests
from fastapi import APIRouter, HTTPException, Depends
from pydantic import BaseModel, HttpUrl

from middleware import check_rate_limit

logger = logging.getLogger(__name__)

webhooks_router = APIRouter()

# ---------------------------------------------------------------------------
# In-memory webhook store (Supabase-backed when available)
# ---------------------------------------------------------------------------

_webhooks: dict[str, dict] = {}  # keyed by webhook_id


def _get_webhooks_from_db() -> list[dict]:
    """Load webhooks from Supabase if available."""
    try:
        from db.supabase import get_client
        client = get_client()
        if not client:
            return []
        resp = client.table("webhooks").select("*").eq("is_active", True).execute()
        return resp.data or []
    except Exception as e:
        logger.debug(f"Could not load webhooks from DB: {e}")
        return []


def _save_webhook_to_db(webhook: dict) -> bool:
    """Persist webhook to Supabase."""
    try:
        from db.supabase import get_client
        client = get_client()
        if not client:
            return False
        client.table("webhooks").insert(webhook).execute()
        return True
    except Exception as e:
        logger.warning(f"Failed to save webhook to DB: {e}")
        return False


def _delete_webhook_from_db(webhook_id: str) -> bool:
    """Remove webhook from Supabase."""
    try:
        from db.supabase import get_client
        client = get_client()
        if not client:
            return False
        client.table("webhooks").delete().eq("id", webhook_id).execute()
        return True
    except Exception as e:
        logger.warning(f"Failed to delete webhook from DB: {e}")
        return False


# ---------------------------------------------------------------------------
# Models
# ---------------------------------------------------------------------------

class WebhookCreate(BaseModel):
    url: HttpUrl
    events: list[str] = ["extraction.completed", "validation.completed"]
    secret: Optional[str] = None  # optional signing secret


class WebhookResponse(BaseModel):
    id: str
    url: str
    events: list[str]
    is_active: bool
    created_at: str


# ---------------------------------------------------------------------------
# Endpoints
# ---------------------------------------------------------------------------

@webhooks_router.post("/api/v1/webhooks")
async def register_webhook(
    body: WebhookCreate,
    api_key: str = Depends(check_rate_limit),
):
    """Register a webhook URL to receive extraction/validation result notifications."""
    valid_events = {"extraction.completed", "validation.completed", "batch.completed"}
    for event in body.events:
        if event not in valid_events:
            raise HTTPException(
                status_code=400,
                detail=f"Invalid event '{event}'. Valid events: {sorted(valid_events)}",
            )

    webhook_id = str(uuid4())
    webhook = {
        "id": webhook_id,
        "url": str(body.url),
        "events": body.events,
        "secret": body.secret or "",
        "is_active": True,
        "created_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
    }

    _save_webhook_to_db(webhook)
    _webhooks[webhook_id] = webhook

    return {
        "id": webhook_id,
        "url": webhook["url"],
        "events": webhook["events"],
        "is_active": True,
        "created_at": webhook["created_at"],
    }


@webhooks_router.get("/api/v1/webhooks")
async def list_webhooks(api_key: str = Depends(check_rate_limit)):
    """List all registered webhooks."""
    # Merge in-memory and DB webhooks
    db_hooks = _get_webhooks_from_db()
    all_hooks = {h["id"]: h for h in db_hooks}
    all_hooks.update(_webhooks)

    return {
        "webhooks": [
            {
                "id": h["id"],
                "url": h["url"],
                "events": h.get("events", []),
                "is_active": h.get("is_active", True),
                "created_at": h.get("created_at", ""),
            }
            for h in all_hooks.values()
            if h.get("is_active", True)
        ]
    }


@webhooks_router.delete("/api/v1/webhooks/{webhook_id}")
async def delete_webhook(webhook_id: str, api_key: str = Depends(check_rate_limit)):
    """Deactivate a webhook."""
    if webhook_id in _webhooks:
        _webhooks[webhook_id]["is_active"] = False
    _delete_webhook_from_db(webhook_id)
    return {"deleted": True}


# ---------------------------------------------------------------------------
# Delivery — called internally after extraction/validation completes
# ---------------------------------------------------------------------------

WEBHOOK_TIMEOUT = int(os.environ.get("WEBHOOK_TIMEOUT", "10"))  # seconds


def _sign_payload(payload_bytes: bytes, secret: str) -> str:
    """Create HMAC-SHA256 signature for webhook payload."""
    return hmac.new(secret.encode(), payload_bytes, hashlib.sha256).hexdigest()


def deliver_webhook(event: str, data: dict):
    """
    Fire webhooks for a given event. Called internally after
    extraction or validation completes. Non-blocking best-effort.
    """
    # Gather all active webhooks (in-memory + DB)
    all_hooks = list(_webhooks.values())
    try:
        all_hooks.extend(_get_webhooks_from_db())
    except Exception:
        pass

    # Deduplicate by ID
    seen = set()
    unique_hooks = []
    for h in all_hooks:
        if h["id"] not in seen and h.get("is_active", True):
            seen.add(h["id"])
            unique_hooks.append(h)

    for hook in unique_hooks:
        if event not in hook.get("events", []):
            continue

        payload = {
            "event": event,
            "timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
            "data": data,
        }
        payload_bytes = json.dumps(payload, default=str).encode()

        headers = {"Content-Type": "application/json"}
        if hook.get("secret"):
            headers["X-Webhook-Signature"] = _sign_payload(payload_bytes, hook["secret"])

        try:
            resp = requests.post(
                hook["url"],
                data=payload_bytes,
                headers=headers,
                timeout=WEBHOOK_TIMEOUT,
            )
            logger.info(
                "Webhook delivered: event=%s url=%s status=%d",
                event, hook["url"], resp.status_code,
            )
        except Exception as e:
            logger.warning("Webhook delivery failed: url=%s error=%s", hook["url"], e)