feat(automation): add Zapier and Make.com integration

Add REST hooks subscription endpoints, incoming action endpoints, and
Zapier-compatible flat payload format for automation platform integration.

- AutomationHook model for webhook subscriptions
- POST /api/automation/hooks/subscribe and DELETE /hooks/{id}
- GET /api/automation/triggers/sample/{event} for Zapier field mapping
- POST /api/automation/actions/upload for incoming document uploads
- Celery task with retry for async hook delivery
- Integration with existing webhook dispatch flow
- 30 passing tests covering all new functionality

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

app/api/__init__.py

@@ -8,6 +8,7 @@ from fastapi import APIRouter
 
 from app.api.admin_users import router as admin_users_router
 from app.api.api_tokens import router as api_tokens_router
+from app.api.automation import router as automation_router
 from app.api.azure import router as azure_router
 from app.api.backup import router as backup_router
 from app.api.billing import router as billing_router
@@ -82,3 +83,4 @@ router.include_router(imap_accounts_router)
 router.include_router(integrations_router)
 router.include_router(notifications_router)
 router.include_router(scheduled_jobs_router)
+router.include_router(automation_router)

app/api/automation.py

@@ -0,0 +1,306 @@
+"""API endpoints for Zapier / Make.com automation integration.
+
+Provides a REST hooks subscription interface for outgoing triggers and
+incoming action endpoints that external automation platforms can call.
+
+Outgoing triggers:
+    External platforms subscribe to DocuElevate events via
+    ``POST /api/automation/hooks/subscribe``.  When a subscribed event
+    fires, DocuElevate POSTs a flat Zapier-compatible JSON payload to the
+    registered ``target_url``.
+
+Incoming actions:
+    ``POST /api/automation/actions/upload`` allows automation platforms to
+    push documents into DocuElevate for processing.
+
+Authentication:
+    All endpoints require a valid API token via ``Authorization: Bearer``
+    header.
+"""
+
+import json
+import logging
+import os
+import tempfile
+from typing import Annotated, Any
+
+from fastapi import APIRouter, Depends, File, HTTPException, Request, UploadFile, status
+from pydantic import BaseModel, Field
+from sqlalchemy.orm import Session
+
+from app.config import settings
+from app.database import get_db
+from app.models import AutomationHook
+from app.utils.automation_hooks import SAMPLE_PAYLOADS
+from app.utils.webhook import VALID_EVENTS
+
+logger = logging.getLogger(__name__)
+router = APIRouter(prefix="/automation", tags=["automation"])
+
+DbSession = Annotated[Session, Depends(get_db)]
+
+
+# ---------------------------------------------------------------------------
+# Auth helper – require a valid API token (Bearer)
+# ---------------------------------------------------------------------------
+
+
+def _require_api_user(request: Request) -> dict:
+    """Ensure the caller is authenticated via session or API token.
+
+    Raises:
+        HTTPException: 401 if not authenticated, 403 if automation hooks are disabled.
+    """
+    if not settings.automation_hooks_enabled:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="Automation hooks are disabled",
+        )
+
+    # Check for API-token user first (set by auth middleware)
+    user = getattr(request.state, "api_token_user", None)
+    if user:
+        return user
+
+    # Fall back to session user
+    user = request.session.get("user")
+    if user:
+        return user
+
+    raise HTTPException(
+        status_code=status.HTTP_401_UNAUTHORIZED,
+        detail="Authentication required (Bearer token or session)",
+    )
+
+
+AuthUser = Annotated[dict, Depends(_require_api_user)]
+
+
+# ---------------------------------------------------------------------------
+# Pydantic schemas
+# ---------------------------------------------------------------------------
+
+
+class HookSubscribe(BaseModel):
+    """Schema for subscribing to automation hook events."""
+
+    target_url: str = Field(..., min_length=1, max_length=2048, description="URL to POST event payloads to")
+    events: list[str] = Field(..., min_length=1, description="Event types to subscribe to")
+    secret: str | None = Field(default=None, max_length=512, description="Optional HMAC-SHA256 signing secret")
+    hook_type: str = Field(
+        default="generic",
+        max_length=50,
+        description="Platform identifier (zapier, make, generic)",
+    )
+    description: str | None = Field(default=None, max_length=500, description="Optional human-readable label")
+
+
+class HookResponse(BaseModel):
+    """Schema returned when listing or creating hooks."""
+
+    id: int
+    target_url: str
+    events: list[str]
+    is_active: bool
+    hook_type: str
+    description: str | None
+    has_secret: bool
+
+    model_config = {"from_attributes": True}
+
+
+class ActionUploadResponse(BaseModel):
+    """Response after an automation action uploads a document."""
+
+    status: str
+    filename: str
+    task_id: str | None = None
+
+
+# ---------------------------------------------------------------------------
+# 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: {', '.join(sorted(VALID_EVENTS))}",
+        )
+
+
+def _hook_to_response(hook: AutomationHook) -> dict[str, Any]:
+    """Convert a DB model instance to a response dict."""
+    try:
+        events = json.loads(hook.events)
+    except (json.JSONDecodeError, TypeError):
+        events = []
+    return {
+        "id": hook.id,
+        "target_url": hook.target_url,
+        "events": events,
+        "is_active": hook.is_active,
+        "hook_type": hook.hook_type,
+        "description": hook.description,
+        "has_secret": hook.secret is not None and len(hook.secret) > 0,
+    }
+
+
+# ---------------------------------------------------------------------------
+# Outgoing triggers – REST hooks subscription endpoints
+# ---------------------------------------------------------------------------
+
+
+@router.post(
+    "/hooks/subscribe",
+    status_code=status.HTTP_201_CREATED,
+    summary="Subscribe to automation events (REST hooks)",
+)
+def subscribe_hook(body: HookSubscribe, db: DbSession, user: AuthUser) -> dict[str, Any]:
+    """Register a new automation hook subscription.
+
+    Zapier and Make.com call this endpoint to subscribe to DocuElevate
+    events.  When an event fires, a flat JSON payload is POSTed to
+    ``target_url``.
+    """
+    _validate_events(body.events)
+
+    hook = AutomationHook(
+        target_url=body.target_url,
+        secret=body.secret,
+        events=json.dumps(sorted(body.events)),
+        is_active=True,
+        hook_type=body.hook_type or "generic",
+        description=body.description,
+    )
+    try:
+        db.add(hook)
+        db.commit()
+        db.refresh(hook)
+    except Exception:
+        db.rollback()
+        raise
+
+    logger.info("Automation hook %d created (type=%s) for events %s", hook.id, hook.hook_type, body.events)
+    return _hook_to_response(hook)
+
+
+@router.delete(
+    "/hooks/{hook_id}",
+    status_code=status.HTTP_204_NO_CONTENT,
+    summary="Unsubscribe an automation hook",
+)
+def unsubscribe_hook(hook_id: int, db: DbSession, user: AuthUser) -> None:
+    """Remove an automation hook subscription.
+
+    Zapier calls this endpoint when a Zap is turned off or deleted.
+    """
+    hook = db.query(AutomationHook).filter(AutomationHook.id == hook_id).first()
+    if not hook:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Hook not found")
+
+    try:
+        db.delete(hook)
+        db.commit()
+    except Exception:
+        db.rollback()
+        raise
+
+    logger.info("Automation hook %d deleted", hook_id)
+
+
+@router.get("/hooks", summary="List automation hook subscriptions")
+def list_hooks(db: DbSession, user: AuthUser) -> list[dict[str, Any]]:
+    """Return all active automation hook subscriptions."""
+    hooks = db.query(AutomationHook).order_by(AutomationHook.id).all()
+    return [_hook_to_response(h) for h in hooks]
+
+
+# ---------------------------------------------------------------------------
+# Outgoing triggers – sample data for Zapier field mapping
+# ---------------------------------------------------------------------------
+
+
+@router.get("/triggers/sample/{event}", summary="Get sample trigger data")
+def get_trigger_sample(event: str, user: AuthUser) -> list[dict[str, Any]]:
+    """Return sample payload data for the given event type.
+
+    Zapier uses this during Zap setup to discover available fields and
+    provide a mapping interface.  The response is wrapped in an array
+    as Zapier expects.
+    """
+    if event not in VALID_EVENTS:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=f"Unknown event: {event}. Valid: {', '.join(sorted(VALID_EVENTS))}",
+        )
+
+    sample = SAMPLE_PAYLOADS.get(event, {"id": "evt_sample", "event": event, "timestamp": 0})
+    return [sample]
+
+
+# ---------------------------------------------------------------------------
+# Outgoing triggers – list valid events
+# ---------------------------------------------------------------------------
+
+
+@router.get("/events", summary="List valid automation event types")
+def list_events(user: AuthUser) -> list[str]:
+    """Return the list of valid event types that automation hooks can subscribe to."""
+    return sorted(VALID_EVENTS)
+
+
+# ---------------------------------------------------------------------------
+# Incoming actions – endpoints that Zapier / Make.com can call
+# ---------------------------------------------------------------------------
+
+
+@router.post("/actions/upload", summary="Upload a document (incoming action)")
+def action_upload(
+    request: Request,
+    db: DbSession,
+    user: AuthUser,
+    file: UploadFile = File(...),
+) -> dict[str, Any]:
+    """Accept a document upload from an automation platform.
+
+    This endpoint allows Zapier or Make.com to push a document into
+    DocuElevate for processing.  The file is saved to the work directory
+    and a background processing task is queued.
+    """
+    if not file.filename:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Filename is required")
+
+    owner_id = user.get("preferred_username") or user.get("email") or user.get("id", "automation")
+    workdir = settings.workdir or tempfile.gettempdir()
+    upload_dir = os.path.join(workdir, "uploads")
+    os.makedirs(upload_dir, exist_ok=True)
+
+    dest_path = os.path.join(upload_dir, file.filename)
+    try:
+        contents = file.file.read()
+        with open(dest_path, "wb") as f:
+            f.write(contents)
+    except Exception as exc:
+        logger.error("Failed to save uploaded file: %s", exc)
+        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Failed to save file")
+
+    # Queue background processing
+    task_id = None
+    try:
+        from app.tasks.process_document import process_document
+
+        result = process_document.delay(dest_path, owner_id)
+        task_id = result.id
+        logger.info("Automation upload queued: file=%s, task=%s, owner=%s", file.filename, task_id, owner_id)
+    except Exception as exc:
+        logger.warning("Could not queue processing task (Celery may be unavailable): %s", exc)
+
+    return {
+        "status": "accepted",
+        "filename": file.filename,
+        "task_id": task_id,
+    }

app/config.py

@@ -617,6 +617,12 @@ class Settings(BaseSettings):
         description="Enable webhook delivery for document events",
     )
 
+    # Automation hooks (Zapier / Make.com)
+    automation_hooks_enabled: bool = Field(
+        default=True,
+        description="Enable Zapier / Make.com automation hook subscriptions and delivery",
+    )
+
     # ── Backup / restore settings ──────────────────────────────────────────────
     backup_enabled: bool = Field(
         default=True,

app/models.py

@@ -176,6 +176,28 @@ class WebhookConfig(Base):
     updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now())
 
 
+class AutomationHook(Base):
+    """Zapier / Make.com compatible webhook subscription for automation triggers.
+
+    External automation platforms subscribe to DocuElevate events via the REST
+    hooks protocol.  When an event fires, DocuElevate POSTs a Zapier-compatible
+    flat JSON payload to ``target_url``.  The ``hook_type`` field records which
+    platform created the subscription (informational only).
+    """
+
+    __tablename__ = "automation_hooks"
+
+    id = Column(Integer, primary_key=True, index=True)
+    target_url = Column(String, nullable=False)  # URL to POST events to
+    secret = Column(String, nullable=True)  # Optional HMAC-SHA256 signing secret
+    events = Column(Text, nullable=False)  # JSON list of subscribed event names
+    is_active = Column(Boolean, default=True, nullable=False)
+    hook_type = Column(String(50), nullable=False, default="generic")  # zapier | make | generic
+    description = Column(String, nullable=True)  # Optional human-readable label
+    created_at = Column(DateTime(timezone=True), server_default=func.now())
+    updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now())
+
+
 class LocalUser(Base):
     """A locally-registered user authenticated by email and bcrypt password.
 

app/tasks/automation_tasks.py

@@ -0,0 +1,44 @@
+"""Celery task for asynchronous automation hook 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="automation.deliver_hook")
+def deliver_automation_hook_task(self, url: str, payload: dict[str, Any], secret: str | None = None) -> dict[str, Any]:
+    """Deliver an automation hook payload to *url* with automatic retries.
+
+    Args:
+        url: Target webhook URL (provided by Zapier / Make.com).
+        payload: The flat Zapier-compatible payload.
+        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 automation hook 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"Automation hook delivery to {url} failed")

app/utils/automation_hooks.py

@@ -0,0 +1,188 @@
+"""Automation hook utilities for Zapier / Make.com integration.
+
+Provides helpers to build Zapier-compatible flat payloads, query active
+automation hook subscriptions, and fan-out event delivery to all matching
+hooks via Celery tasks.
+
+The payload format is intentionally *flat* (no nested ``data`` key) so that
+Zapier and Make.com can map fields without JSONPath expressions.  An ``id``
+field is included for Zapier deduplication.
+"""
+
+import json
+import logging
+import time
+import uuid
+from typing import Any
+
+from app.config import settings
+from app.database import SessionLocal
+from app.models import AutomationHook
+from app.utils.webhook import VALID_EVENTS
+
+logger = logging.getLogger(__name__)
+
+
+# ---------------------------------------------------------------------------
+# Payload helpers
+# ---------------------------------------------------------------------------
+
+
+def build_zapier_payload(event: str, data: dict[str, Any]) -> dict[str, Any]:
+    """Build a flat, Zapier-compatible webhook payload.
+
+    Zapier works best with flat JSON objects that include an ``id`` field
+    for deduplication.  This function merges event metadata into the
+    top-level object alongside the event-specific *data*.
+
+    Args:
+        event: The event name (e.g. ``document.processed``).
+        data: Event-specific key/value pairs.
+
+    Returns:
+        A flat dictionary suitable for Zapier / Make.com consumption.
+    """
+    return {
+        "id": f"evt_{uuid.uuid4().hex[:16]}",
+        "event": event,
+        "timestamp": time.time(),
+        **data,
+    }
+
+
+# ---------------------------------------------------------------------------
+# Sample payloads (used by the /triggers/sample endpoint)
+# ---------------------------------------------------------------------------
+
+#: Example payloads that Zapier uses for field-mapping during Zap creation.
+SAMPLE_PAYLOADS: dict[str, dict[str, Any]] = {
+    "document.uploaded": {
+        "id": "evt_sample0001",
+        "event": "document.uploaded",
+        "timestamp": 1710000000.0,
+        "document_id": 42,
+        "filename": "invoice_2024.pdf",
+        "content_type": "application/pdf",
+        "size_bytes": 204800,
+        "owner_id": "user@example.com",
+    },
+    "document.processed": {
+        "id": "evt_sample0002",
+        "event": "document.processed",
+        "timestamp": 1710000060.0,
+        "document_id": 42,
+        "filename": "invoice_2024.pdf",
+        "status": "processed",
+        "title": "Invoice #1234",
+        "owner_id": "user@example.com",
+    },
+    "document.failed": {
+        "id": "evt_sample0003",
+        "event": "document.failed",
+        "timestamp": 1710000120.0,
+        "document_id": 42,
+        "filename": "corrupt.pdf",
+        "status": "failed",
+        "error": "Unable to extract text from document",
+        "owner_id": "user@example.com",
+    },
+    "user.signup": {
+        "id": "evt_sample0004",
+        "event": "user.signup",
+        "timestamp": 1710000180.0,
+        "user_id": "newuser@example.com",
+        "display_name": "Jane Doe",
+    },
+    "user.plan_changed": {
+        "id": "evt_sample0005",
+        "event": "user.plan_changed",
+        "timestamp": 1710000240.0,
+        "user_id": "user@example.com",
+        "old_tier": "free",
+        "new_tier": "pro",
+    },
+    "user.payment_issue": {
+        "id": "evt_sample0006",
+        "event": "user.payment_issue",
+        "timestamp": 1710000300.0,
+        "user_id": "user@example.com",
+        "issue": "Credit card declined",
+    },
+}
+
+
+# ---------------------------------------------------------------------------
+# Database queries
+# ---------------------------------------------------------------------------
+
+
+def get_active_hooks_for_event(event: str) -> list[dict[str, Any]]:
+    """Return all active automation hooks subscribed to *event*.
+
+    Args:
+        event: The event name to filter on.
+
+    Returns:
+        A list of dicts with ``id``, ``target_url``, ``secret``, and
+        ``events`` keys.
+    """
+    db = SessionLocal()
+    try:
+        hooks = db.query(AutomationHook).filter(AutomationHook.is_active.is_(True)).all()
+        result: list[dict[str, Any]] = []
+        for hook in hooks:
+            try:
+                subscribed = json.loads(hook.events)
+            except (json.JSONDecodeError, TypeError):
+                subscribed = []
+            if event in subscribed:
+                result.append(
+                    {
+                        "id": hook.id,
+                        "target_url": hook.target_url,
+                        "secret": hook.secret,
+                        "events": subscribed,
+                    }
+                )
+        return result
+    finally:
+        db.close()
+
+
+# ---------------------------------------------------------------------------
+# Dispatch
+# ---------------------------------------------------------------------------
+
+
+def dispatch_automation_hooks(event: str, data: dict[str, Any]) -> None:
+    """Fan-out an event to all matching active automation hooks.
+
+    Builds a Zapier-compatible flat payload and queues a Celery task for
+    each matching hook so delivery is asynchronous with automatic retries.
+
+    Args:
+        event: Event name (must be in :data:`VALID_EVENTS`).
+        data: Event-specific payload data.
+    """
+    if not settings.automation_hooks_enabled:
+        return
+
+    if event not in VALID_EVENTS:
+        logger.warning("Ignoring unknown automation hook event: %s", event)
+        return
+
+    hooks = get_active_hooks_for_event(event)
+    if not hooks:
+        logger.debug("No active automation hooks for event %s", event)
+        return
+
+    payload = build_zapier_payload(event, data)
+
+    from app.tasks.automation_tasks import deliver_automation_hook_task
+
+    for hook in hooks:
+        try:
+            deliver_automation_hook_task.delay(hook["target_url"], payload, hook["secret"])
+            logger.debug("Queued automation hook delivery to %s for event %s", hook["target_url"], event)
+        except Exception as exc:
+            logger.error("Failed to queue automation hook to %s: %s", hook["target_url"], exc)

app/utils/settings_service.py

@@ -1595,6 +1595,18 @@ SETTING_METADATA = {
         "required": False,
         "restart_required": False,
     },
+    "automation_hooks_enabled": {
+        "category": "Feature Flags",
+        "description": (
+            "Enable Zapier / Make.com automation hook subscriptions and delivery. "
+            "When enabled, external automation platforms can subscribe to DocuElevate events "
+            "via the REST hooks protocol. Default: True."
+        ),
+        "type": "boolean",
+        "sensitive": False,
+        "required": False,
+        "restart_required": False,
+    },
     # Backup / Restore
     "backup_enabled": {
         "category": "Backup",

app/utils/webhook.py

@@ -145,6 +145,8 @@ def dispatch_webhook_event(event: str, data: dict[str, Any]) -> None:
     It delegates to :func:`deliver_webhook_task` (Celery) for each matching
     webhook so delivery happens asynchronously with automatic retries.
 
+    Also dispatches to automation hooks (Zapier / Make.com) if enabled.
+
     Args:
         event: Event name (must be in :data:`VALID_EVENTS`).
         data: Event-specific payload data.
@@ -156,16 +158,23 @@ def dispatch_webhook_event(event: str, data: dict[str, Any]) -> None:
     webhooks = get_active_webhooks_for_event(event)
     if not webhooks:
         logger.debug("No active webhooks for event %s", event)
-        return
+    else:
+        payload = build_payload(event, data)
 
-    payload = build_payload(event, data)
+        # Import here to avoid circular dependency with celery_app
+        from app.tasks.webhook_tasks import deliver_webhook_task
 
-    # 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)
 
-    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)
+    # Also fan-out to Zapier / Make.com automation hooks
+    try:
+        from app.utils.automation_hooks import dispatch_automation_hooks
+
+        dispatch_automation_hooks(event, data)
+    except Exception as exc:
+        logger.error("Failed to dispatch automation hooks for event %s: %s", event, exc)