Phase 4: Webhooks, Backup/Restore UI, Onboarding/Tutorial

- Webhooks Backend: model, schema, service (HMAC-SHA256, httpx), routes, event bus dispatcher, migration 0042 - Webhooks Frontend: SettingsWebhooksPage (CRUD, test button, event multi-select), API client - Backup/Restore Backend: model, schema, service (pg_dump/pg_restore), routes (admin-only), migration 0043 - Backup/Restore Frontend: SettingsBackupPage (create, list, restore dialog with RESTORE confirmation, auto-refresh) - Onboarding: OnboardingTour (8 steps, custom CSS overlay), WelcomeDialog, onboardingStore (zustand + localStorage) - Onboarding integrated into AppShell - Routes: /settings/webhooks, /settings/backup registered - Settings nav: Webhooks, Backup & Restore entries added - Migration conflict fixed: 0042_webhooks → 0043_backups chain
2026-07-26 03:17:40 +02:00
parent 10dcc8ae90
commit 79ece0fe2e
23 changed files with 3094 additions and 0 deletions
@@ -0,0 +1,98 @@
+"""Webhook dispatcher — subscribes to the event bus and dispatches to matching webhooks."""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+import uuid
+from typing import Any
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import async_sessionmaker
+
+from app.core.db import get_engine, get_session_factory
+from app.core.event_bus import EventBus, get_event_bus
+from app.models.webhook import Webhook
+from app.services.webhook_service import send_webhook
+
+logger = logging.getLogger(__name__)
+
+
+async def _dispatch_event(payload: dict[str, Any]) -> None:
+    """Handle an event from the event bus: find matching webhooks and dispatch.
+
+    The payload is expected to contain:
+        - event_name: str
+        - tenant_id: str (UUID)
+        - data: dict (the actual event data)
+    """
+    event_name = payload.get("event_name", "")
+    tenant_id_str = payload.get("tenant_id", "")
+    data = payload.get("data", {})
+
+    if not event_name or not tenant_id_str:
+        logger.warning("webhook_dispatcher: missing event_name or tenant_id in payload")
+        return
+
+    try:
+        tenant_id = uuid.UUID(tenant_id_str)
+    except (ValueError, TypeError):
+        logger.warning(f"webhook_dispatcher: invalid tenant_id: {tenant_id_str}")
+        return
+
+    # Find active webhooks for this tenant that subscribe to this event
+    session_factory = get_session_factory()
+    async with session_factory() as db:
+        stmt = select(Webhook).where(
+            Webhook.tenant_id == tenant_id,
+            Webhook.is_active == True,  # noqa: E712
+            Webhook.events.any(event_name),
+        )
+        result = await db.execute(stmt)
+        webhooks = list(result.scalars().all())
+
+    if not webhooks:
+        return
+
+    # Dispatch to all matching webhooks concurrently
+    tasks = []
+    for webhook in webhooks:
+        tasks.append(_dispatch_single(webhook, event_name, data))
+
+    if tasks:
+        await asyncio.gather(*tasks, return_exceptions=True)
+
+
+async def _dispatch_single(
+    webhook: Webhook,
+    event_name: str,
+    data: dict[str, Any],
+) -> None:
+    """Send a webhook and log the result."""
+    try:
+        result = await send_webhook(webhook, event_name, data)
+        if result["success"]:
+            logger.info(
+                f"Webhook {webhook.id} sent successfully to {webhook.url} "
+                f"for event {event_name} (status={result['status_code']})"
+            )
+        else:
+            logger.warning(
+                f"Webhook {webhook.id} failed for {webhook.url} "
+                f"event {event_name}: {result.get('error')}"
+            )
+    except Exception as exc:
+        logger.error(
+            f"Webhook {webhook.id} dispatch error for {webhook.url}: {exc}"
+        )
+
+
+def register_webhook_event_handlers(event_bus: EventBus | None = None) -> None:
+    """Register webhook dispatcher on the global event bus.
+
+    Subscribes to all events via the wildcard '*' handler and filters
+    by event name internally. Should be called during application startup.
+    """
+    bus = event_bus or get_event_bus()
+    bus.subscribe("*", _dispatch_event)
+    logger.info("Webhook dispatcher registered on event bus (wildcard '*' handler)")
@@ -52,6 +52,8 @@ from app.routes import (
    custom_field_definitions,
    custom_fields,
    saved_filters,
+    webhooks,
+    backups,
 )


@@ -331,9 +333,11 @@ def create_app() -> FastAPI:
    app.include_router(addresses.router)
    app.include_router(bank_accounts.router)
    app.include_router(audit.router)
+    app.include_router(backups.router)
    app.include_router(custom_field_definitions.router)
    app.include_router(custom_fields.router)
    app.include_router(saved_filters.router)
+    app.include_router(webhooks.router)

    # ── Register plugin routes for all built-in plugins ──
    # Routes are registered here (before app start); activation status
@@ -21,7 +21,9 @@ from app.models.system_settings import SystemSettings
 from app.models.tax import TaxRate
 from app.models.tenant import Tenant
 from app.models.user import User, UserTenant
+from app.models.backup import Backup
 from app.models.custom_field_definition import CustomFieldDefinition
+from app.models.webhook import Webhook
 from app.models.workflow import Workflow, WorkflowInstance, WorkflowStepHistory

 __all__ = [
@@ -55,7 +57,9 @@ __all__ = [
    "PluginMigration",
    "AIConversation",
    "AIMessage",
+    "Backup",
    "CustomFieldDefinition",
+    "Webhook",
    "Workflow",
    "WorkflowInstance",
    "WorkflowStepHistory",
@@ -0,0 +1,46 @@
+"""Backup model — tracks database backup records per tenant."""
+
+from __future__ import annotations
+
+import uuid
+from datetime import datetime
+
+from sqlalchemy import BigInteger, DateTime, String, Text
+from sqlalchemy.dialects.postgresql import UUID as PGUUID
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.core.db import Base, TenantMixin
+
+
+class Backup(Base, TenantMixin):
+    """Database backup record scoped to a tenant.
+
+    Tracks pg_dump backups with status, file location, and error details.
+    """
+
+    __tablename__ = "backups"
+
+    id: Mapped[uuid.UUID] = mapped_column(
+        PGUUID(as_uuid=True), primary_key=True, default=uuid.uuid4
+    )
+
+    # ── Identity ──
+    filename: Mapped[str] = mapped_column(String(255), nullable=False)
+    size_bytes: Mapped[int | None] = mapped_column(BigInteger, nullable=True, default=None)
+
+    # ── Status ──
+    status: Mapped[str] = mapped_column(
+        String(20), nullable=False, default="pending"
+    )  # pending, completed, failed
+    error_message: Mapped[str | None] = mapped_column(Text, nullable=True, default=None)
+
+    # ── Audit ──
+    created_by: Mapped[uuid.UUID | None] = mapped_column(
+        PGUUID(as_uuid=True), nullable=True
+    )
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, default=datetime.utcnow
+    )
+    completed_at: Mapped[datetime | None] = mapped_column(
+        DateTime(timezone=True), nullable=True, default=None
+    )
@@ -0,0 +1,43 @@
+"""Webhook model — outgoing webhook subscriptions for event-driven notifications."""
+
+from __future__ import annotations
+
+import uuid
+
+from sqlalchemy import Boolean, Integer, JSON, String
+from sqlalchemy.dialects.postgresql import UUID as PGUUID
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.core.db import Base, TenantMixin
+
+
+class Webhook(Base, TenantMixin):
+    """Outgoing webhook subscription.
+
+    Each webhook defines a target URL, a list of events to subscribe to,
+    and delivery settings (retry count, timeout, HMAC secret).
+    """
+
+    __tablename__ = "webhooks"
+
+    id: Mapped[uuid.UUID] = mapped_column(
+        PGUUID(as_uuid=True), primary_key=True, default=uuid.uuid4
+    )
+
+    # ── Target ──
+    url: Mapped[str] = mapped_column(String(500), nullable=False)
+    events: Mapped[list[str]] = mapped_column(JSON, nullable=False, default=list)
+    secret: Mapped[str | None] = mapped_column(String(255), nullable=True, default=None)
+
+    # ── Delivery Settings ──
+    is_active: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True)
+    retry_count: Mapped[int] = mapped_column(Integer, nullable=False, default=3)
+    timeout_seconds: Mapped[int] = mapped_column(Integer, nullable=False, default=30)
+
+    # ── Audit ──
+    created_by: Mapped[uuid.UUID | None] = mapped_column(
+        PGUUID(as_uuid=True), nullable=True
+    )
+    updated_by: Mapped[uuid.UUID | None] = mapped_column(
+        PGUUID(as_uuid=True), nullable=True
+    )
@@ -0,0 +1,104 @@
+"""API routes for Backup management."""
+
+from __future__ import annotations
+
+import uuid
+
+from fastapi import APIRouter, Depends, HTTPException
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.db import get_db
+from app.deps import get_current_user, require_permission
+from app.schemas.backup import BackupListResponse, BackupResponse
+from app.services import backup_service
+
+router = APIRouter(prefix="/api/v1/backups", tags=["backups"])
+
+
+@router.get(
+    "",
+    response_model=BackupListResponse,
+    dependencies=[Depends(require_permission("automation:admin"))],
+)
+async def list_backups(
+    db: AsyncSession = Depends(get_db),
+    current_user: dict = Depends(get_current_user),
+):
+    """List all backups for the current tenant."""
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+    backups = await backup_service.list_backups(db, tenant_id)
+    return BackupListResponse(
+        backups=[BackupResponse.model_validate(b) for b in backups],
+        total=len(backups),
+    )
+
+
+@router.post(
+    "",
+    response_model=BackupResponse,
+    status_code=201,
+    dependencies=[Depends(require_permission("automation:admin"))],
+)
+async def create_backup(
+    db: AsyncSession = Depends(get_db),
+    current_user: dict = Depends(get_current_user),
+):
+    """Create a new database backup."""
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+    user_id = uuid.UUID(current_user["user_id"])
+    backup = await backup_service.create_backup(db, tenant_id, user_id=user_id)
+    return backup
+
+
+@router.post(
+    "/{backup_id}/restore",
+    response_model=BackupResponse,
+    dependencies=[Depends(require_permission("automation:admin"))],
+)
+async def restore_backup(
+    backup_id: str,
+    db: AsyncSession = Depends(get_db),
+    current_user: dict = Depends(get_current_user),
+):
+    """Restore a database backup.
+
+    WARNING: This is a destructive operation. It drops and recreates the database.
+    """
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+    try:
+        b_id = uuid.UUID(backup_id)
+    except (ValueError, TypeError):
+        raise HTTPException(400, detail={"detail": "Invalid backup_id", "code": "invalid_id"}) from None
+
+    try:
+        backup = await backup_service.restore_backup(db, tenant_id, b_id)
+        return backup
+    except ValueError as exc:
+        raise HTTPException(400, detail={"detail": str(exc), "code": "invalid_state"}) from exc
+    except FileNotFoundError as exc:
+        raise HTTPException(404, detail={"detail": str(exc), "code": "file_not_found"}) from exc
+    except RuntimeError as exc:
+        raise HTTPException(500, detail={"detail": str(exc), "code": "restore_failed"}) from exc
+
+
+@router.delete(
+    "/{backup_id}",
+    status_code=204,
+    dependencies=[Depends(require_permission("automation:admin"))],
+)
+async def delete_backup(
+    backup_id: str,
+    db: AsyncSession = Depends(get_db),
+    current_user: dict = Depends(get_current_user),
+):
+    """Delete a backup record and its file."""
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+    try:
+        b_id = uuid.UUID(backup_id)
+    except (ValueError, TypeError):
+        raise HTTPException(400, detail={"detail": "Invalid backup_id", "code": "invalid_id"}) from None
+
+    deleted = await backup_service.delete_backup(db, tenant_id, b_id)
+    if not deleted:
+        raise HTTPException(404, detail={"detail": "Backup not found", "code": "not_found"})
+    return None
@@ -0,0 +1,161 @@
+"""API routes for Webhook CRUD and testing."""
+
+from __future__ import annotations
+
+import uuid
+
+from fastapi import APIRouter, Depends, HTTPException, Query
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.db import get_db
+from app.deps import get_current_user, require_permission
+from app.schemas.webhook import (
+    WebhookCreate,
+    WebhookResponse,
+    WebhookUpdate,
+)
+from app.services import webhook_service
+
+router = APIRouter(prefix="/api/v1/webhooks", tags=["webhooks"])
+
+
+@router.get(
+    "",
+    response_model=list[WebhookResponse],
+    dependencies=[Depends(require_permission("automation:read"))],
+)
+async def list_webhooks(
+    event: str | None = Query(None, description="Filter by event name"),
+    db: AsyncSession = Depends(get_db),
+    current_user: dict = Depends(get_current_user),
+):
+    """List all webhooks for the current tenant."""
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+    webhooks = await webhook_service.list_webhooks(db, tenant_id, event=event)
+    return webhooks
+
+
+@router.post(
+    "",
+    response_model=WebhookResponse,
+    status_code=201,
+    dependencies=[Depends(require_permission("automation:write"))],
+)
+async def create_webhook(
+    body: WebhookCreate,
+    db: AsyncSession = Depends(get_db),
+    current_user: dict = Depends(get_current_user),
+):
+    """Create a new webhook subscription."""
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+    user_id = uuid.UUID(current_user["user_id"])
+    webhook = await webhook_service.create_webhook(
+        db, tenant_id, user_id, body.model_dump()
+    )
+    return webhook
+
+
+@router.get(
+    "/{webhook_id}",
+    response_model=WebhookResponse,
+    dependencies=[Depends(require_permission("automation:read"))],
+)
+async def get_webhook(
+    webhook_id: str,
+    db: AsyncSession = Depends(get_db),
+    current_user: dict = Depends(get_current_user),
+):
+    """Get a single webhook by ID."""
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+    try:
+        wh_id = uuid.UUID(webhook_id)
+    except (ValueError, TypeError):
+        raise HTTPException(400, detail={"detail": "Invalid webhook_id", "code": "invalid_id"}) from None
+
+    webhook = await webhook_service.get_webhook(db, tenant_id, wh_id)
+    if webhook is None:
+        raise HTTPException(404, detail={"detail": "Webhook not found", "code": "not_found"})
+    return webhook
+
+
+@router.patch(
+    "/{webhook_id}",
+    response_model=WebhookResponse,
+    dependencies=[Depends(require_permission("automation:write"))],
+)
+async def update_webhook(
+    webhook_id: str,
+    body: WebhookUpdate,
+    db: AsyncSession = Depends(get_db),
+    current_user: dict = Depends(get_current_user),
+):
+    """Update an existing webhook subscription."""
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+    user_id = uuid.UUID(current_user["user_id"])
+    try:
+        wh_id = uuid.UUID(webhook_id)
+    except (ValueError, TypeError):
+        raise HTTPException(400, detail={"detail": "Invalid webhook_id", "code": "invalid_id"}) from None
+
+    update_data = {k: v for k, v in body.model_dump().items() if v is not None}
+    if not update_data:
+        raise HTTPException(400, detail={"detail": "No fields to update", "code": "no_updates"})
+
+    webhook = await webhook_service.update_webhook(
+        db, tenant_id, wh_id, update_data, user_id=user_id
+    )
+    if webhook is None:
+        raise HTTPException(404, detail={"detail": "Webhook not found", "code": "not_found"})
+    return webhook
+
+
+@router.delete(
+    "/{webhook_id}",
+    status_code=204,
+    dependencies=[Depends(require_permission("automation:write"))],
+)
+async def delete_webhook(
+    webhook_id: str,
+    db: AsyncSession = Depends(get_db),
+    current_user: dict = Depends(get_current_user),
+):
+    """Delete a webhook subscription."""
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+    try:
+        wh_id = uuid.UUID(webhook_id)
+    except (ValueError, TypeError):
+        raise HTTPException(400, detail={"detail": "Invalid webhook_id", "code": "invalid_id"}) from None
+
+    deleted = await webhook_service.delete_webhook(db, tenant_id, wh_id)
+    if not deleted:
+        raise HTTPException(404, detail={"detail": "Webhook not found", "code": "not_found"})
+    return None
+
+
+@router.post(
+    "/{webhook_id}/test",
+    dependencies=[Depends(require_permission("automation:write"))],
+)
+async def test_webhook(
+    webhook_id: str,
+    db: AsyncSession = Depends(get_db),
+    current_user: dict = Depends(get_current_user),
+):
+    """Send a test payload to a webhook to verify connectivity."""
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+    try:
+        wh_id = uuid.UUID(webhook_id)
+    except (ValueError, TypeError):
+        raise HTTPException(400, detail={"detail": "Invalid webhook_id", "code": "invalid_id"}) from None
+
+    webhook = await webhook_service.get_webhook(db, tenant_id, wh_id)
+    if webhook is None:
+        raise HTTPException(404, detail={"detail": "Webhook not found", "code": "not_found"})
+
+    test_payload = {
+        "type": "test",
+        "message": "This is a test webhook from LeoCRM",
+        "webhook_id": str(webhook.id),
+    }
+    result = await webhook_service.send_webhook(webhook, "webhook.test", test_payload)
+    return result
@@ -0,0 +1,31 @@
+"""Pydantic schemas for Backup CRUD."""
+
+from __future__ import annotations
+
+import uuid
+from datetime import datetime
+
+from pydantic import BaseModel, ConfigDict
+
+
+class BackupResponse(BaseModel):
+    """Schema for returning a backup record."""
+
+    model_config = ConfigDict(from_attributes=True)
+
+    id: uuid.UUID
+    tenant_id: uuid.UUID
+    filename: str
+    size_bytes: int | None = None
+    status: str  # pending, completed, failed
+    error_message: str | None = None
+    created_by: uuid.UUID | None = None
+    created_at: datetime
+    completed_at: datetime | None = None
+
+
+class BackupListResponse(BaseModel):
+    """Schema for listing backups."""
+
+    backups: list[BackupResponse]
+    total: int
@@ -0,0 +1,49 @@
+"""Pydantic schemas for Webhook CRUD."""
+
+from __future__ import annotations
+
+import uuid
+from datetime import datetime
+
+from pydantic import BaseModel, ConfigDict, Field
+
+
+class WebhookCreate(BaseModel):
+    """Schema for creating a new webhook."""
+
+    url: str = Field(..., min_length=1, max_length=500, description="Target URL for the webhook")
+    events: list[str] = Field(..., min_length=1, description="List of event names to subscribe to")
+    secret: str | None = Field(default=None, max_length=255, description="HMAC secret for payload signing")
+    is_active: bool = Field(default=True, description="Whether the webhook is active")
+    retry_count: int = Field(default=3, ge=0, le=10, description="Number of retry attempts on failure")
+    timeout_seconds: int = Field(default=30, ge=1, le=120, description="Request timeout in seconds")
+
+
+class WebhookUpdate(BaseModel):
+    """Schema for updating an existing webhook."""
+
+    url: str | None = Field(default=None, max_length=500)
+    events: list[str] | None = Field(default=None, min_length=1)
+    secret: str | None = Field(default=None, max_length=255)
+    is_active: bool | None = Field(default=None)
+    retry_count: int | None = Field(default=None, ge=0, le=10)
+    timeout_seconds: int | None = Field(default=None, ge=1, le=120)
+
+
+class WebhookResponse(BaseModel):
+    """Schema for returning a webhook."""
+
+    model_config = ConfigDict(from_attributes=True)
+
+    id: uuid.UUID
+    tenant_id: uuid.UUID
+    url: str
+    events: list[str]
+    secret: str | None = None
+    is_active: bool = True
+    retry_count: int = 3
+    timeout_seconds: int = 30
+    created_by: uuid.UUID | None = None
+    updated_by: uuid.UUID | None = None
+    created_at: datetime
+    updated_at: datetime
@@ -0,0 +1,268 @@
+"""Service for database backup and restore operations."""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+import os
+import uuid
+from datetime import datetime
+from pathlib import Path
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.models.backup import Backup
+
+logger = logging.getLogger(__name__)
+
+BACKUP_DIR = Path("/tmp/leocrm-backups")
+
+
+def _ensure_backup_dir() -> None:
+    """Ensure the backup directory exists."""
+    BACKUP_DIR.mkdir(parents=True, exist_ok=True)
+
+
+def _get_database_url() -> str:
+    """Get DATABASE_URL from environment."""
+    url = os.environ.get("DATABASE_URL")
+    if not url:
+        raise RuntimeError("DATABASE_URL environment variable is not set")
+    return url
+
+
+def _parse_pg_url(url: str) -> dict[str, str]:
+    """Parse a PostgreSQL connection URL into components for pg_dump/pg_restore.
+
+    Handles formats:
+        postgresql://user:pass@host:port/dbname
+        postgresql+asyncpg://user:pass@host:port/dbname
+    """
+    # Remove async driver prefix if present
+    if "+asyncpg" in url:
+        url = url.replace("+asyncpg", "")
+    if "+psycopg2" in url:
+        url = url.replace("+psycopg2", "")
+
+    # Parse the URL manually to avoid dependency on urllib parsing quirks
+    # Format: postgresql://user:pass@host:port/dbname
+    rest = url.split("://", 1)[1] if "://" in url else url
+    user_info, rest = rest.split("@", 1) if "@" in rest else ("", rest)
+    user = ""
+    password = ""
+    if ":" in user_info:
+        user, password = user_info.split(":", 1)
+    else:
+        user = user_info
+
+    host_port, dbname = rest.split("/", 1) if "/" in rest else (rest, "")
+    host = host_port
+    port = "5432"
+    if ":" in host_port:
+        host, port = host_port.split(":", 1)
+
+    return {
+        "host": host,
+        "port": port,
+        "user": user,
+        "password": password,
+        "dbname": dbname,
+    }
+
+
+async def list_backups(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+) -> list[Backup]:
+    """List all backups for a tenant, ordered by creation date descending."""
+    stmt = (
+        select(Backup)
+        .where(Backup.tenant_id == tenant_id)
+        .order_by(Backup.created_at.desc())
+    )
+    result = await db.execute(stmt)
+    return list(result.scalars().all())
+
+
+async def create_backup(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    user_id: uuid.UUID | None = None,
+) -> Backup:
+    """Create a database backup using pg_dump.
+
+    Creates a backup record, runs pg_dump to a file, then updates the record
+    with the file size and status.
+    """
+    _ensure_backup_dir()
+
+    timestamp = datetime.utcnow().strftime("%Y%m%d_%H%M%S")
+    filename = f"leocrm_backup_{tenant_id}_{timestamp}.dump"
+    filepath = BACKUP_DIR / filename
+
+    # Create initial pending record
+    backup = Backup(
+        tenant_id=tenant_id,
+        filename=filename,
+        status="pending",
+        created_by=user_id,
+    )
+    db.add(backup)
+    await db.flush()
+    await db.refresh(backup)
+
+    try:
+        database_url = _get_database_url()
+        pg = _parse_pg_url(database_url)
+
+        # Build pg_dump command
+        env = os.environ.copy()
+        if pg["password"]:
+            env["PGPASSWORD"] = pg["password"]
+
+        cmd = [
+            "pg_dump",
+            "--host", pg["host"],
+            "--port", pg["port"],
+            "--username", pg["user"],
+            "--format", "custom",
+            "--file", str(filepath),
+            pg["dbname"],
+        ]
+
+        logger.info("Running pg_dump: %s to %s", cmd, filepath)
+
+        process = await asyncio.create_subprocess_exec(
+            *cmd,
+            env=env,
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE,
+        )
+        stdout, stderr = await process.communicate()
+
+        if process.returncode != 0:
+            error_msg = stderr.decode() if stderr else "pg_dump failed with unknown error"
+            logger.error("pg_dump failed: %s", error_msg)
+            backup.status = "failed"
+            backup.error_message = error_msg
+            await db.flush()
+            await db.refresh(backup)
+            return backup
+
+        # Get file size
+        size_bytes = filepath.stat().st_size if filepath.exists() else 0
+
+        # Update backup record
+        backup.status = "completed"
+        backup.size_bytes = size_bytes
+        backup.completed_at = datetime.utcnow()
+        await db.flush()
+        await db.refresh(backup)
+
+        logger.info("Backup completed: %s (%d bytes)", filename, size_bytes)
+        return backup
+
+    except Exception as exc:
+        logger.exception("Backup creation failed")
+        backup.status = "failed"
+        backup.error_message = str(exc)
+        await db.flush()
+        await db.refresh(backup)
+        return backup
+
+
+async def restore_backup(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    backup_id: uuid.UUID,
+) -> Backup:
+    """Restore a database backup using pg_restore.
+
+    WARNING: This is a destructive operation. It drops and recreates the database.
+    """
+    stmt = select(Backup).where(
+        Backup.id == backup_id,
+        Backup.tenant_id == tenant_id,
+    )
+    result = await db.execute(stmt)
+    backup = result.scalar_one_or_none()
+
+    if backup is None:
+        raise ValueError("Backup not found")
+
+    if backup.status != "completed":
+        raise ValueError(f"Backup status is '{backup.status}', cannot restore")
+
+    filepath = BACKUP_DIR / backup.filename
+    if not filepath.exists():
+        raise FileNotFoundError(f"Backup file not found: {filepath}")
+
+    try:
+        database_url = _get_database_url()
+        pg = _parse_pg_url(database_url)
+
+        env = os.environ.copy()
+        if pg["password"]:
+            env["PGPASSWORD"] = pg["password"]
+
+        cmd = [
+            "pg_restore",
+            "--host", pg["host"],
+            "--port", pg["port"],
+            "--username", pg["user"],
+            "--dbname", pg["dbname"],
+            "--clean",
+            "--if-exists",
+            "--no-owner",
+            "--no-acl",
+            str(filepath),
+        ]
+
+        logger.info("Running pg_restore: %s", cmd)
+
+        process = await asyncio.create_subprocess_exec(
+            *cmd,
+            env=env,
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE,
+        )
+        stdout, stderr = await process.communicate()
+
+        if process.returncode != 0:
+            error_msg = stderr.decode() if stderr else "pg_restore failed with unknown error"
+            logger.error("pg_restore failed: %s", error_msg)
+            raise RuntimeError(f"Restore failed: {error_msg}")
+
+        logger.info("Restore completed from backup: %s", backup.filename)
+        return backup
+
+    except Exception as exc:
+        logger.exception("Restore failed")
+        raise
+
+
+async def delete_backup(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    backup_id: uuid.UUID,
+) -> bool:
+    """Delete a backup record and its file."""
+    stmt = select(Backup).where(
+        Backup.id == backup_id,
+        Backup.tenant_id == tenant_id,
+    )
+    result = await db.execute(stmt)
+    backup = result.scalar_one_or_none()
+
+    if backup is None:
+        return False
+
+    # Delete the file if it exists
+    filepath = BACKUP_DIR / backup.filename
+    if filepath.exists():
+        filepath.unlink()
+
+    await db.delete(backup)
+    await db.flush()
+    return True
@@ -0,0 +1,183 @@
+"""CRUD service for Webhook subscriptions and delivery."""
+
+from __future__ import annotations
+
+import hashlib
+import hmac
+import json
+import logging
+import uuid
+from typing import Any
+
+import httpx
+from sqlalchemy import select, delete
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.models.webhook import Webhook
+
+logger = logging.getLogger(__name__)
+
+
+async def list_webhooks(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    event: str | None = None,
+) -> list[Webhook]:
+    """List webhooks for a tenant, optionally filtered by event."""
+    stmt = select(Webhook).where(
+        Webhook.tenant_id == tenant_id,
+    )
+    if event:
+        stmt = stmt.where(Webhook.events.any(event))
+    stmt = stmt.order_by(Webhook.created_at.desc())
+    result = await db.execute(stmt)
+    return list(result.scalars().all())
+
+
+async def get_webhook(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    webhook_id: uuid.UUID,
+) -> Webhook | None:
+    """Get a single webhook by ID."""
+    stmt = select(Webhook).where(
+        Webhook.id == webhook_id,
+        Webhook.tenant_id == tenant_id,
+    )
+    result = await db.execute(stmt)
+    return result.scalar_one_or_none()
+
+
+async def create_webhook(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    user_id: uuid.UUID,
+    data: dict[str, Any],
+) -> Webhook:
+    """Create a new webhook subscription."""
+    webhook = Webhook(
+        tenant_id=tenant_id,
+        url=data["url"],
+        events=data["events"],
+        secret=data.get("secret"),
+        is_active=data.get("is_active", True),
+        retry_count=data.get("retry_count", 3),
+        timeout_seconds=data.get("timeout_seconds", 30),
+        created_by=user_id,
+        updated_by=user_id,
+    )
+    db.add(webhook)
+    await db.flush()
+    await db.refresh(webhook)
+    return webhook
+
+
+async def update_webhook(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    webhook_id: uuid.UUID,
+    data: dict[str, Any],
+    user_id: uuid.UUID | None = None,
+) -> Webhook | None:
+    """Update an existing webhook subscription."""
+    webhook = await get_webhook(db, tenant_id, webhook_id)
+    if webhook is None:
+        return None
+
+    update_fields = ["url", "events", "secret", "is_active", "retry_count", "timeout_seconds"]
+    for field in update_fields:
+        if field in data:
+            setattr(webhook, field, data[field])
+
+    if user_id is not None:
+        webhook.updated_by = user_id
+
+    await db.flush()
+    await db.refresh(webhook)
+    return webhook
+
+
+async def delete_webhook(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    webhook_id: uuid.UUID,
+) -> bool:
+    """Delete a webhook subscription."""
+    stmt = select(Webhook).where(
+        Webhook.id == webhook_id,
+        Webhook.tenant_id == tenant_id,
+    )
+    result = await db.execute(stmt)
+    webhook = result.scalar_one_or_none()
+    if webhook is None:
+        return False
+    await db.delete(webhook)
+    await db.flush()
+    return True
+
+
+def _sign_payload(payload: bytes, secret: str) -> str:
+    """Create HMAC-SHA256 signature for a payload."""
+    return hmac.new(
+        secret.encode("utf-8"),
+        payload,
+        hashlib.sha256,
+    ).hexdigest()
+
+
+async def send_webhook(
+    webhook: Webhook,
+    event_name: str,
+    payload: dict[str, Any],
+) -> dict[str, Any]:
+    """Send a webhook HTTP POST with HMAC signature.
+
+    Returns a dict with 'success' (bool), 'status_code' (int | None),
+    and 'error' (str | None).
+    """
+    body = {
+        "event": event_name,
+        "payload": payload,
+        "timestamp": __import__("datetime").datetime.utcnow().isoformat() + "Z",
+    }
+    body_bytes = json.dumps(body, default=str).encode("utf-8")
+
+    headers = {
+        "Content-Type": "application/json",
+        "User-Agent": "LeoCRM-Webhook/1.0",
+    }
+
+    if webhook.secret:
+        signature = _sign_payload(body_bytes, webhook.secret)
+        headers["X-Webhook-Signature"] = f"sha256={signature}"
+
+    try:
+        async with httpx.AsyncClient(timeout=webhook.timeout_seconds) as client:
+            response = await client.post(
+                webhook.url,
+                content=body_bytes,
+                headers=headers,
+            )
+        return {
+            "success": response.is_success,
+            "status_code": response.status_code,
+            "error": None,
+        }
+    except httpx.TimeoutException:
+        return {"success": False, "status_code": None, "error": "timeout"}
+    except httpx.RequestError as exc:
+        return {"success": False, "status_code": None, "error": str(exc)}
+
+
+async def retry_webhook(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    webhook_id: uuid.UUID,
+    event_name: str,
+    payload: dict[str, Any],
+) -> dict[str, Any]:
+    """Retry sending a webhook with the given event and payload."""
+    webhook = await get_webhook(db, tenant_id, webhook_id)
+    if webhook is None:
+        return {"success": False, "status_code": None, "error": "webhook_not_found"}
+    return await send_webhook(webhook, event_name, payload)