feat(webhooks): add webhook support for external integrations

Add WebhookConfig model, CRUD API endpoints, HMAC-SHA256 signed delivery,
and Celery-based async dispatch with retry/backoff for document events
(document.uploaded, document.processed, document.failed).

Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>

app/api/__init__.py

@@ -23,6 +23,7 @@ from app.api.url_upload import router as url_upload_router
 
 # Import all the individual routers
 from app.api.user import router as user_router
+from app.api.webhooks import router as webhooks_router
 
 # Set up logging
 logger = logging.getLogger(__name__)
@@ -46,3 +47,4 @@ router.include_router(url_upload_router)
 router.include_router(search_router)
 router.include_router(queue_router)
 router.include_router(saved_searches_router)
+router.include_router(webhooks_router)

app/api/webhooks.py

@@ -0,0 +1,206 @@
+"""API endpoints for managing webhook configurations.
+
+Provides CRUD operations for webhook configs that notify external systems
+when document events occur (``document.uploaded``, ``document.processed``,
+``document.failed``).
+"""
+
+import json
+import logging
+from typing import Annotated, Any
+
+from fastapi import APIRouter, Depends, HTTPException, Request, status
+from pydantic import BaseModel, Field
+from sqlalchemy.orm import Session
+
+from app.database import get_db
+from app.models import WebhookConfig
+from app.utils.webhook import VALID_EVENTS
+
+logger = logging.getLogger(__name__)
+router = APIRouter(prefix="/webhooks", tags=["webhooks"])
+
+DbSession = Annotated[Session, Depends(get_db)]
+
+
+# ---------------------------------------------------------------------------
+# Auth helper (reuse the pattern from settings API)
+# ---------------------------------------------------------------------------
+
+
+def _require_admin(request: Request) -> dict:
+    """Ensure the caller is an admin. Raises 403 otherwise."""
+    user = request.session.get("user")
+    if not user or not user.get("is_admin"):
+        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Admin access required")
+    return user
+
+
+AdminUser = Annotated[dict, Depends(_require_admin)]
+
+
+# ---------------------------------------------------------------------------
+# Pydantic schemas
+# ---------------------------------------------------------------------------
+
+
+class WebhookCreate(BaseModel):
+    """Schema for creating a new webhook configuration."""
+
+    url: str = Field(..., min_length=1, max_length=2048, description="Target URL for webhook delivery")
+    secret: str | None = Field(default=None, max_length=512, description="Shared secret for HMAC-SHA256 signatures")
+    events: list[str] = Field(..., min_length=1, description="List of events to subscribe to")
+    is_active: bool = Field(default=True, description="Whether the webhook is active")
+    description: str | None = Field(default=None, max_length=500, description="Optional human-readable description")
+
+
+class WebhookUpdate(BaseModel):
+    """Schema for updating an existing webhook configuration."""
+
+    url: str | None = Field(default=None, min_length=1, max_length=2048)
+    secret: str | None = Field(default=None, max_length=512)
+    events: list[str] | None = Field(default=None, min_length=1)
+    is_active: bool | None = None
+    description: str | None = Field(default=None, max_length=500)
+
+
+class WebhookResponse(BaseModel):
+    """Schema returned to clients (secret is never exposed)."""
+
+    id: int
+    url: str
+    events: list[str]
+    is_active: bool
+    description: str | None
+    has_secret: bool
+
+    model_config = {"from_attributes": True}
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _validate_events(events: list[str]) -> None:
+    """Raise 422 if any event name is not recognised."""
+    invalid = set(events) - VALID_EVENTS
+    if invalid:
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+            detail=f"Invalid event(s): {', '.join(sorted(invalid))}. Valid events: {', '.join(sorted(VALID_EVENTS))}",
+        )
+
+
+def _to_response(cfg: WebhookConfig) -> dict[str, Any]:
+    """Convert a DB model instance to a response dict."""
+    try:
+        events = json.loads(cfg.events)
+    except (json.JSONDecodeError, TypeError):
+        events = []
+    return {
+        "id": cfg.id,
+        "url": cfg.url,
+        "events": events,
+        "is_active": cfg.is_active,
+        "description": cfg.description,
+        "has_secret": cfg.secret is not None and len(cfg.secret) > 0,
+    }
+
+
+# ---------------------------------------------------------------------------
+# Endpoints
+# ---------------------------------------------------------------------------
+
+
+@router.get("/", summary="List all webhook configurations")
+def list_webhooks(db: DbSession, _admin: AdminUser) -> list[dict[str, Any]]:
+    """Return all webhook configurations. Secrets are never included."""
+    configs = db.query(WebhookConfig).order_by(WebhookConfig.id).all()
+    return [_to_response(c) for c in configs]
+
+
+@router.get("/{webhook_id}", summary="Get a single webhook configuration")
+def get_webhook(webhook_id: int, db: DbSession, _admin: AdminUser) -> dict[str, Any]:
+    """Return a single webhook configuration by ID."""
+    cfg = db.query(WebhookConfig).filter(WebhookConfig.id == webhook_id).first()
+    if not cfg:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Webhook not found")
+    return _to_response(cfg)
+
+
+@router.post("/", status_code=status.HTTP_201_CREATED, summary="Create a webhook configuration")
+def create_webhook(body: WebhookCreate, db: DbSession, _admin: AdminUser) -> dict[str, Any]:
+    """Create a new webhook configuration."""
+    _validate_events(body.events)
+
+    cfg = WebhookConfig(
+        url=body.url,
+        secret=body.secret,
+        events=json.dumps(sorted(body.events)),
+        is_active=body.is_active,
+        description=body.description,
+    )
+    try:
+        db.add(cfg)
+        db.commit()
+        db.refresh(cfg)
+    except Exception:
+        db.rollback()
+        raise
+
+    logger.info("Webhook %d created for events %s", cfg.id, body.events)
+    return _to_response(cfg)
+
+
+@router.put("/{webhook_id}", summary="Update a webhook configuration")
+def update_webhook(webhook_id: int, body: WebhookUpdate, db: DbSession, _admin: AdminUser) -> dict[str, Any]:
+    """Update an existing webhook configuration. Only supplied fields are changed."""
+    cfg = db.query(WebhookConfig).filter(WebhookConfig.id == webhook_id).first()
+    if not cfg:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Webhook not found")
+
+    if body.url is not None:
+        cfg.url = body.url
+    if body.secret is not None:
+        cfg.secret = body.secret
+    if body.events is not None:
+        _validate_events(body.events)
+        cfg.events = json.dumps(sorted(body.events))
+    if body.is_active is not None:
+        cfg.is_active = body.is_active
+    if body.description is not None:
+        cfg.description = body.description
+
+    try:
+        db.commit()
+        db.refresh(cfg)
+    except Exception:
+        db.rollback()
+        raise
+
+    logger.info("Webhook %d updated", cfg.id)
+    return _to_response(cfg)
+
+
+@router.delete("/{webhook_id}", status_code=status.HTTP_204_NO_CONTENT, summary="Delete a webhook configuration")
+def delete_webhook(webhook_id: int, db: DbSession, _admin: AdminUser) -> None:
+    """Delete a webhook configuration."""
+    cfg = db.query(WebhookConfig).filter(WebhookConfig.id == webhook_id).first()
+    if not cfg:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Webhook not found")
+
+    try:
+        db.delete(cfg)
+        db.commit()
+    except Exception:
+        db.rollback()
+        raise
+
+    logger.info("Webhook %d deleted", cfg.id)
+
+
+@router.get("/events/", summary="List valid webhook event types")
+def list_events(_admin: AdminUser) -> list[str]:
+    """Return the list of valid event types that can be subscribed to."""
+    return sorted(VALID_EVENTS)

app/config.py

@@ -274,6 +274,12 @@ class Settings(BaseSettings):
         description="Send notifications when files are successfully processed",
     )
 
+    # Webhook settings
+    webhook_enabled: bool = Field(
+        default=True,
+        description="Enable webhook delivery for document events",
+    )
+
     # File upload size limits (for security - see SECURITY_AUDIT.md)
     max_upload_size: int = Field(
         default=1073741824,  # 1GB in bytes (1024 * 1024 * 1024)

app/models.py

@@ -143,3 +143,18 @@ class SavedSearch(Base):
     updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now())
 
     __table_args__ = (UniqueConstraint("user_id", "name", name="unique_user_search_name"),)
+
+
+class WebhookConfig(Base):
+    """Webhook configuration for notifying external systems of document events."""
+
+    __tablename__ = "webhook_configs"
+
+    id = Column(Integer, primary_key=True, index=True)
+    url = Column(String, nullable=False)  # Target URL for webhook delivery
+    secret = Column(String, nullable=True)  # Shared secret for HMAC-SHA256 signature
+    events = Column(Text, nullable=False)  # JSON list of subscribed events
+    is_active = Column(Boolean, default=True, nullable=False)  # Whether the webhook is active
+    description = Column(String, nullable=True)  # Optional human-readable description
+    created_at = Column(DateTime(timezone=True), server_default=func.now())
+    updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now())

app/tasks/webhook_tasks.py

@@ -0,0 +1,39 @@
+"""Celery task for asynchronous webhook delivery with retry and backoff.
+
+Uses :class:`~app.tasks.retry_config.BaseTaskWithRetry` so failed deliveries
+are automatically retried with exponential backoff (default: 60 s, 300 s,
+900 s) and ±20 % jitter.
+"""
+
+import logging
+from typing import Any
+
+from app.celery_app import celery
+from app.tasks.retry_config import BaseTaskWithRetry
+from app.utils.webhook import deliver_webhook
+
+logger = logging.getLogger(__name__)
+
+
+@celery.task(base=BaseTaskWithRetry, bind=True, name="webhook.deliver")
+def deliver_webhook_task(self, url: str, payload: dict[str, Any], secret: str | None = None) -> dict[str, Any]:
+    """Deliver a webhook payload to *url* with automatic retries.
+
+    Args:
+        url: Target webhook URL.
+        payload: The full webhook payload envelope.
+        secret: Optional shared secret for HMAC-SHA256 signing.
+
+    Returns:
+        A dict with ``status`` and ``url`` on success.
+
+    Raises:
+        RuntimeError: Re-raised to trigger Celery retry on delivery failure.
+    """
+    logger.info("Delivering webhook to %s (attempt %d/%d)", url, self.request.retries + 1, self.max_retries + 1)
+
+    success = deliver_webhook(url, payload, secret)
+    if success:
+        return {"status": "delivered", "url": url}
+
+    raise RuntimeError(f"Webhook delivery to {url} failed")

app/utils/webhook.py

@@ -0,0 +1,165 @@
+"""Webhook delivery utility for notifying external systems of document events.
+
+Provides functions to dispatch webhook payloads with HMAC-SHA256 signatures
+and to query active webhook configurations from the database.
+
+Supported events:
+- ``document.uploaded``  – a new document has been ingested
+- ``document.processed`` – a document finished processing successfully
+- ``document.failed``    – document processing failed
+"""
+
+import hashlib
+import hmac
+import json
+import logging
+import time
+from typing import Any
+
+import requests
+
+from app.database import SessionLocal
+from app.models import WebhookConfig
+
+logger = logging.getLogger(__name__)
+
+#: Events recognised by the webhook subsystem.
+VALID_EVENTS: frozenset[str] = frozenset(
+    {
+        "document.uploaded",
+        "document.processed",
+        "document.failed",
+    }
+)
+
+#: Timeout (seconds) for outgoing webhook HTTP requests.
+WEBHOOK_TIMEOUT = 10
+
+
+def compute_signature(payload_bytes: bytes, secret: str) -> str:
+    """Compute an HMAC-SHA256 hex-digest for *payload_bytes* using *secret*.
+
+    Args:
+        payload_bytes: The raw JSON body to sign.
+        secret: The shared secret string.
+
+    Returns:
+        ``sha256=<hex-digest>`` signature string.
+    """
+    mac = hmac.new(secret.encode("utf-8"), payload_bytes, hashlib.sha256)
+    return f"sha256={mac.hexdigest()}"
+
+
+def deliver_webhook(url: str, payload: dict[str, Any], secret: str | None = None) -> bool:
+    """Send a single webhook POST request.
+
+    Args:
+        url: Target URL.
+        payload: JSON-serialisable dictionary.
+        secret: If provided, an ``X-Webhook-Signature`` header is included.
+
+    Returns:
+        ``True`` when the remote server responds with a 2xx status.
+    """
+    body = json.dumps(payload, default=str, sort_keys=True)
+    body_bytes = body.encode("utf-8")
+
+    headers: dict[str, str] = {"Content-Type": "application/json"}
+    if secret:
+        headers["X-Webhook-Signature"] = compute_signature(body_bytes, secret)
+
+    try:
+        resp = requests.post(url, data=body_bytes, headers=headers, timeout=WEBHOOK_TIMEOUT)
+        if resp.ok:
+            logger.info("Webhook delivered to %s (status %d)", url, resp.status_code)
+            return True
+        logger.warning("Webhook to %s returned status %d", url, resp.status_code)
+        return False
+    except requests.RequestException as exc:
+        logger.error("Webhook delivery to %s failed: %s", url, exc)
+        return False
+
+
+def build_payload(event: str, data: dict[str, Any]) -> dict[str, Any]:
+    """Build a standardised webhook payload envelope.
+
+    Args:
+        event: The event name (e.g. ``document.uploaded``).
+        data: Event-specific data.
+
+    Returns:
+        Dictionary with ``event``, ``timestamp``, and ``data`` keys.
+    """
+    return {
+        "event": event,
+        "timestamp": time.time(),
+        "data": data,
+    }
+
+
+def get_active_webhooks_for_event(event: str) -> list[dict[str, Any]]:
+    """Return all active webhook configs subscribed to *event*.
+
+    Queries the database directly so this helper can be called from both the
+    API layer and Celery tasks.
+
+    Args:
+        event: The event name to filter on.
+
+    Returns:
+        A list of dicts with ``id``, ``url``, ``secret``, and ``events`` keys.
+    """
+    db = SessionLocal()
+    try:
+        configs = db.query(WebhookConfig).filter(WebhookConfig.is_active.is_(True)).all()
+        result: list[dict[str, Any]] = []
+        for cfg in configs:
+            try:
+                subscribed = json.loads(cfg.events)
+            except (json.JSONDecodeError, TypeError):
+                subscribed = []
+            if event in subscribed:
+                result.append(
+                    {
+                        "id": cfg.id,
+                        "url": cfg.url,
+                        "secret": cfg.secret,
+                        "events": subscribed,
+                    }
+                )
+        return result
+    finally:
+        db.close()
+
+
+def dispatch_webhook_event(event: str, data: dict[str, Any]) -> None:
+    """Fan-out a webhook event to all matching active configurations.
+
+    This is the main entry-point used by application code to trigger webhooks.
+    It delegates to :func:`deliver_webhook_task` (Celery) for each matching
+    webhook so delivery happens asynchronously with automatic retries.
+
+    Args:
+        event: Event name (must be in :data:`VALID_EVENTS`).
+        data: Event-specific payload data.
+    """
+    if event not in VALID_EVENTS:
+        logger.warning("Ignoring unknown webhook event: %s", event)
+        return
+
+    webhooks = get_active_webhooks_for_event(event)
+    if not webhooks:
+        logger.debug("No active webhooks for event %s", event)
+        return
+
+    payload = build_payload(event, data)
+
+    # Import here to avoid circular dependency with celery_app
+    from app.tasks.webhook_tasks import deliver_webhook_task
+
+    for wh in webhooks:
+        try:
+            deliver_webhook_task.delay(wh["url"], payload, wh["secret"])
+            logger.debug("Queued webhook delivery to %s for event %s", wh["url"], event)
+        except Exception as exc:
+            logger.error("Failed to queue webhook to %s: %s", wh["url"], exc)