feat(backend): webhooks deliver signed alerts from an outbox with retries and a test send

backend/.env.example

@@ -25,6 +25,7 @@ BOTS__BACKFILL_PER_CHAT=0
 
 API__HOST=0.0.0.0
 API__PORT=8080
+API__PUBLIC_URL=
 
 AUTH__TOKEN=
 

backend/pyproject.toml

@@ -20,6 +20,7 @@ dependencies = [
     "dishka>=1.4.0",
     "greenlet>=3.1.0",
     "fastmcp>=3.3",
+    "httpx2>=2.13.1",
 ]
 
 [build-system]

backend/src/api/app.py

@@ -38,7 +38,9 @@ from api.routers import (
     storage,
     stories,
     watches,
+    webhooks,
 )
+from api.webhooks import outbox
 from dependencies.container import container
 from utils.cache import DAY_HEADERS, IMMUTABLE_HEADERS, NO_STORE_HEADERS
 from utils.env import env
@@ -58,11 +60,14 @@ mcp_app = mcp.http_app(path="/")
 @asynccontextmanager
 async def lifespan(app_: Starlette) -> AsyncGenerator[None]:
     pool = await container.get(asyncpg.Pool)
+    hub.alert_listeners.append(outbox.wake)
     await hub.start(pool)
     purger = asyncio.create_task(
         purge_loop(pool, await container.get(ContentAddressedStorage))
     )
+    deliverer = asyncio.create_task(outbox.run(pool))
     yield
+    deliverer.cancel()
     purger.cancel()
     await hub.stop()
     await app_.state.dishka_container.close()
@@ -103,6 +108,7 @@ app.include_router(peers.router)
 app.include_router(discover.router)
 app.include_router(annotations.router)
 app.include_router(watches.router)
+app.include_router(webhooks.router)
 app.include_router(shares.router)
 app.include_router(storage.router)
 app.include_router(files.router)

backend/src/api/routers/webhooks.py

@@ -0,0 +1,118 @@
+from typing import Annotated
+from urllib.parse import urlsplit
+
+import asyncpg
+from dishka.integrations.fastapi import DishkaRoute, FromDishka
+from fastapi import APIRouter, HTTPException, Query
+from pydantic import BaseModel
+
+from api.webhooks import send_test
+from utils.read import webhooks
+from utils.read.models import DeliveryView, WebhookView
+
+router = APIRouter(prefix="/api/webhooks", tags=["webhooks"], route_class=DishkaRoute)
+
+MAX_URL = 2048
+
+
+class WebhookUpdate(BaseModel):
+    url: str
+    watch_ids: list[int] | None = None
+    enabled: bool = True
+
+
+class WebhookCreate(WebhookUpdate):
+    account_id: int
+
+
+def _url(value: str) -> str:
+    url = value.strip()
+    parts = urlsplit(url)
+    if (
+        len(url) > MAX_URL
+        or parts.scheme not in {"http", "https"}
+        or not parts.hostname
+    ):
+        raise HTTPException(
+            status_code=422, detail="url must be an http or https address"
+        )
+    return url
+
+
+def _watch_ids(ids: list[int] | None) -> list[int] | None:
+    return sorted(set(ids)) if ids else None
+
+
+async def _webhook(pool: asyncpg.Pool, webhook_id: int) -> WebhookView:
+    webhook = await webhooks.get_webhook(pool, webhook_id)
+    if webhook is None:
+        raise HTTPException(status_code=404, detail="webhook not found")
+    return webhook
+
+
+@router.get("")
+async def list_webhooks(
+    pool: FromDishka[asyncpg.Pool], account_id: Annotated[int, Query()]
+) -> list[WebhookView]:
+    return await webhooks.list_webhooks(pool, account_id)
+
+
+@router.post("", status_code=201)
+async def create_webhook(
+    pool: FromDishka[asyncpg.Pool], body: WebhookCreate
+) -> WebhookView:
+    return await webhooks.create_webhook(
+        pool,
+        body.account_id,
+        _url(body.url),
+        _watch_ids(body.watch_ids),
+        enabled=body.enabled,
+    )
+
+
+@router.put("/{webhook_id}")
+async def update_webhook(
+    pool: FromDishka[asyncpg.Pool], webhook_id: int, body: WebhookUpdate
+) -> WebhookView:
+    webhook = await webhooks.update_webhook(
+        pool,
+        webhook_id,
+        _url(body.url),
+        _watch_ids(body.watch_ids),
+        enabled=body.enabled,
+    )
+    if webhook is None:
+        raise HTTPException(status_code=404, detail="webhook not found")
+    return webhook
+
+
+@router.delete("/{webhook_id}", status_code=204)
+async def delete_webhook(pool: FromDishka[asyncpg.Pool], webhook_id: int) -> None:
+    if not await webhooks.delete_webhook(pool, webhook_id):
+        raise HTTPException(status_code=404, detail="webhook not found")
+
+
+@router.post("/{webhook_id}/secret")
+async def rotate_secret(pool: FromDishka[asyncpg.Pool], webhook_id: int) -> WebhookView:
+    webhook = await webhooks.rotate_secret(pool, webhook_id)
+    if webhook is None:
+        raise HTTPException(status_code=404, detail="webhook not found")
+    return webhook
+
+
+@router.post("/{webhook_id}/test")
+async def test_webhook(pool: FromDishka[asyncpg.Pool], webhook_id: int) -> DeliveryView:
+    delivery = await send_test(pool, await _webhook(pool, webhook_id))
+    if delivery is None:
+        raise HTTPException(status_code=500, detail="test delivery was not recorded")
+    return delivery
+
+
+@router.get("/{webhook_id}/deliveries")
+async def list_deliveries(
+    pool: FromDishka[asyncpg.Pool],
+    webhook_id: int,
+    limit: Annotated[int, Query(ge=1, le=100)] = webhooks.DELIVERY_LOG,
+) -> list[DeliveryView]:
+    await _webhook(pool, webhook_id)
+    return await webhooks.list_deliveries(pool, webhook_id, limit)

backend/src/api/webhooks.py

@@ -0,0 +1,188 @@
+import asyncio
+import contextlib
+import hashlib
+import hmac
+import json
+import logging
+from dataclasses import dataclass
+from datetime import UTC, datetime, timedelta
+from typing import Any
+
+import asyncpg
+import httpx2
+
+from utils.env import env
+from utils.read import watches
+from utils.read import webhooks as webhooks_read
+from utils.read.models import AlertItem, DeliveryView, WebhookView
+
+logger = logging.getLogger(__name__)
+
+TIMEOUT_SECONDS = 10
+SCAN_SECONDS = 30
+BATCH = 20
+RETRY_DELAYS = (30, 90, 270, 810, 2430)
+ERROR_CHARS = 300
+USER_AGENT = "beavergram-webhooks"
+
+
+def sign(secret: str, body: bytes) -> str:
+    digest = hmac.new(secret.encode(), body, hashlib.sha256).hexdigest()
+    return f"sha256={digest}"
+
+
+def _encode(body: dict[str, Any]) -> str:
+    return json.dumps(body, ensure_ascii=False, separators=(",", ":"))
+
+
+def _with_url(body: dict[str, Any], path: str) -> dict[str, Any]:
+    if env.api.public_url:
+        body["url"] = env.api.public_url.rstrip("/") + path
+    return body
+
+
+def alert_body(item: AlertItem) -> dict[str, Any]:
+    data = item.model_dump(mode="json", exclude={"id", "seen"})
+    return _with_url({"type": "alert", "alert_id": item.id, **data}, item.path)
+
+
+def test_body(webhook: WebhookView) -> dict[str, Any]:
+    body = {
+        "type": "test",
+        "alert_id": None,
+        "account_id": webhook.account_id,
+        "ts": datetime.now(UTC).isoformat(),
+        "event": "test",
+        "watch": None,
+        "chat": None,
+        "message": None,
+        "peer": None,
+        "path": "/app",
+    }
+    return _with_url(body, "/app")
+
+
+@dataclass(frozen=True, slots=True)
+class Target:
+    delivery_id: int
+    event: str
+    url: str
+    secret: str
+
+
+async def post(
+    client: httpx2.AsyncClient, target: Target, body: str
+) -> tuple[int | None, str | None]:
+    raw = body.encode()
+    headers = {
+        "Content-Type": "application/json",
+        "User-Agent": USER_AGENT,
+        "X-Beavergram-Event": target.event,
+        "X-Beavergram-Delivery": str(target.delivery_id),
+        "X-Beavergram-Signature": sign(target.secret, raw),
+    }
+    try:
+        response = await client.post(target.url, content=raw, headers=headers)
+    except httpx2.HTTPError as exc:
+        return None, (str(exc) or type(exc).__name__)[:ERROR_CHARS]
+    if response.is_success:
+        return response.status_code, None
+    reason = response.text.strip()[:ERROR_CHARS] or response.reason_phrase
+    return response.status_code, reason or f"HTTP {response.status_code}"
+
+
+def new_client() -> httpx2.AsyncClient:
+    return httpx2.AsyncClient(timeout=TIMEOUT_SECONDS, follow_redirects=False)
+
+
+async def _body(pool: asyncpg.Pool, row: asyncpg.Record) -> str | None:
+    if row["body"] is not None:
+        return row["body"]
+    if row["alert_id"] is None:
+        return None
+    item = await watches.get_alert(pool, row["alert_id"])
+    if item is None:
+        return None
+    body = _encode(alert_body(item))
+    await webhooks_read.save_body(pool, row["id"], body)
+    return body
+
+
+async def _deliver(
+    pool: asyncpg.Pool, client: httpx2.AsyncClient, row: asyncpg.Record
+) -> None:
+    body = await _body(pool, row)
+    if body is None:
+        await webhooks_read.finish(
+            pool, row["id"], webhooks_read.STATUS_FAILED, None, "alert is gone"
+        )
+        return
+    target = Target(row["id"], row["event"], row["url"], row["secret"])
+    code, error = await post(client, target, body)
+    if error is None:
+        await webhooks_read.finish(
+            pool, row["id"], webhooks_read.STATUS_DELIVERED, code, None
+        )
+        return
+    attempt = row["attempts"]
+    if attempt > len(RETRY_DELAYS):
+        await webhooks_read.finish(
+            pool, row["id"], webhooks_read.STATUS_FAILED, code, error
+        )
+        return
+    retry_at = datetime.now(UTC) + timedelta(seconds=RETRY_DELAYS[attempt - 1])
+    await webhooks_read.finish(
+        pool, row["id"], webhooks_read.STATUS_PENDING, code, error, retry_at
+    )
+
+
+async def send_test(pool: asyncpg.Pool, webhook: WebhookView) -> DeliveryView | None:
+    body = _encode(test_body(webhook))
+    delivery_id = await webhooks_read.start_test(pool, webhook.id, body)
+    async with new_client() as client:
+        target = Target(delivery_id, "test", webhook.url, webhook.secret)
+        code, error = await post(client, target, body)
+    status = webhooks_read.STATUS_FAILED if error else webhooks_read.STATUS_DELIVERED
+    await webhooks_read.finish(pool, delivery_id, status, code, error)
+    return await webhooks_read.get_delivery(pool, delivery_id)
+
+
+class Outbox:
+    def __init__(self) -> None:
+        self._wake = asyncio.Event()
+
+    def wake(self) -> None:
+        self._wake.set()
+
+    async def run(self, pool: asyncpg.Pool) -> None:
+        await webhooks_read.release_stuck(pool)
+        async with new_client() as client:
+            while True:
+                self._wake.clear()
+                try:
+                    await self._drain(pool, client)
+                    delay = await self._delay(pool)
+                except Exception:
+                    logger.exception("Webhook delivery failed")
+                    delay = SCAN_SECONDS
+                with contextlib.suppress(TimeoutError):
+                    await asyncio.wait_for(self._wake.wait(), timeout=delay)
+
+    async def _drain(self, pool: asyncpg.Pool, client: httpx2.AsyncClient) -> None:
+        while rows := await webhooks_read.claim_due(pool, BATCH):
+            results = await asyncio.gather(
+                *(_deliver(pool, client, row) for row in rows), return_exceptions=True
+            )
+            for result in results:
+                if isinstance(result, Exception):
+                    logger.error("Webhook delivery crashed: %r", result)
+
+    async def _delay(self, pool: asyncpg.Pool) -> float:
+        due = await webhooks_read.next_due(pool)
+        if due is None:
+            return SCAN_SECONDS
+        wait = (due - datetime.now(UTC)).total_seconds()
+        return min(SCAN_SECONDS, max(wait, 1.0))
+
+
+outbox = Outbox()

backend/src/utils/env.py

@@ -41,6 +41,7 @@ class ApiSettings(BaseSettings):
     host: str = "0.0.0.0"  # noqa: S104
     port: int = 8080
     static_dir: str = "static"
+    public_url: str | None = None
 
 
 class AuthSettings(BaseSettings):

backend/src/utils/read/webhooks.py

@@ -0,0 +1,185 @@
+import secrets
+from datetime import datetime
+
+import asyncpg
+
+from utils.read.models import DeliveryView, WebhookView
+
+STATUS_PENDING = "pending"
+STATUS_SENDING = "sending"
+STATUS_DELIVERED = "delivered"
+STATUS_FAILED = "failed"
+DELIVERY_LOG = 20
+
+_COLS = "id, account_id, url, secret, enabled, watch_ids, created_at, updated_at"
+_DELIVERY_COLS = (
+    "id, webhook_id, alert_id, event, status, attempts, next_attempt_at, "
+    "response_code, last_error, created_at, finished_at"
+)
+
+_CLAIM = """
+UPDATE webhook_deliveries d SET status = 'sending', attempts = d.attempts + 1
+FROM webhooks w
+WHERE w.id = d.webhook_id AND d.id IN (
+    SELECT id FROM webhook_deliveries
+    WHERE status = 'pending' AND next_attempt_at <= now()
+    ORDER BY next_attempt_at LIMIT $1
+    FOR UPDATE SKIP LOCKED
+)
+RETURNING d.id, d.alert_id, d.event, d.attempts, d.body,
+          w.id AS webhook_id, w.account_id, w.url, w.secret
+"""
+
+_FINISH = """
+UPDATE webhook_deliveries SET
+    status = $2, response_code = $3, last_error = $4,
+    next_attempt_at = coalesce($5, next_attempt_at),
+    finished_at = CASE WHEN $2 = 'pending' THEN NULL ELSE now() END
+WHERE id = $1
+"""
+
+
+def new_secret() -> str:
+    return secrets.token_hex(32)
+
+
+async def list_webhooks(pool: asyncpg.Pool, account_id: int) -> list[WebhookView]:
+    rows = await pool.fetch(
+        f"SELECT {_COLS} FROM webhooks WHERE account_id = $1 ORDER BY id",  # noqa: S608
+        account_id,
+    )
+    return [WebhookView(**dict(row)) for row in rows]
+
+
+async def get_webhook(pool: asyncpg.Pool, webhook_id: int) -> WebhookView | None:
+    row = await pool.fetchrow(
+        f"SELECT {_COLS} FROM webhooks WHERE id = $1",  # noqa: S608
+        webhook_id,
+    )
+    return WebhookView(**dict(row)) if row else None
+
+
+async def create_webhook(
+    pool: asyncpg.Pool,
+    account_id: int,
+    url: str,
+    watch_ids: list[int] | None,
+    *,
+    enabled: bool,
+) -> WebhookView:
+    row = await pool.fetchrow(
+        "INSERT INTO webhooks (account_id, url, secret, enabled, watch_ids) "  # noqa: S608
+        f"VALUES ($1, $2, $3, $4, $5) RETURNING {_COLS}",
+        account_id,
+        url,
+        new_secret(),
+        enabled,
+        watch_ids,
+    )
+    return WebhookView(**dict(row))
+
+
+async def update_webhook(
+    pool: asyncpg.Pool,
+    webhook_id: int,
+    url: str,
+    watch_ids: list[int] | None,
+    *,
+    enabled: bool,
+) -> WebhookView | None:
+    async with pool.acquire() as conn, conn.transaction():
+        row = await conn.fetchrow(
+            "UPDATE webhooks SET url = $2, enabled = $3, watch_ids = $4, "  # noqa: S608
+            f"updated_at = now() WHERE id = $1 RETURNING {_COLS}",
+            webhook_id,
+            url,
+            enabled,
+            watch_ids,
+        )
+        if row is not None and not enabled:
+            await conn.execute(
+                "UPDATE webhook_deliveries SET status = 'failed', "
+                "last_error = 'endpoint disabled', finished_at = now() "
+                "WHERE webhook_id = $1 AND status = 'pending'",
+                webhook_id,
+            )
+    return WebhookView(**dict(row)) if row else None
+
+
+async def rotate_secret(pool: asyncpg.Pool, webhook_id: int) -> WebhookView | None:
+    row = await pool.fetchrow(
+        "UPDATE webhooks SET secret = $2, updated_at = now() "  # noqa: S608
+        f"WHERE id = $1 RETURNING {_COLS}",
+        webhook_id,
+        new_secret(),
+    )
+    return WebhookView(**dict(row)) if row else None
+
+
+async def delete_webhook(pool: asyncpg.Pool, webhook_id: int) -> bool:
+    result = await pool.execute("DELETE FROM webhooks WHERE id = $1", webhook_id)
+    return result.endswith("1")
+
+
+async def list_deliveries(
+    pool: asyncpg.Pool, webhook_id: int, limit: int = DELIVERY_LOG
+) -> list[DeliveryView]:
+    rows = await pool.fetch(
+        f"SELECT {_DELIVERY_COLS} FROM webhook_deliveries "  # noqa: S608
+        "WHERE webhook_id = $1 ORDER BY id DESC LIMIT $2",
+        webhook_id,
+        limit,
+    )
+    return [DeliveryView(**dict(row)) for row in rows]
+
+
+async def get_delivery(pool: asyncpg.Pool, delivery_id: int) -> DeliveryView | None:
+    row = await pool.fetchrow(
+        f"SELECT {_DELIVERY_COLS} FROM webhook_deliveries WHERE id = $1",  # noqa: S608
+        delivery_id,
+    )
+    return DeliveryView(**dict(row)) if row else None
+
+
+async def start_test(pool: asyncpg.Pool, webhook_id: int, body: str) -> int:
+    return await pool.fetchval(
+        "INSERT INTO webhook_deliveries (webhook_id, event, status, attempts, body) "
+        "VALUES ($1, 'test', 'sending', 1, $2) RETURNING id",
+        webhook_id,
+        body,
+    )
+
+
+async def release_stuck(pool: asyncpg.Pool) -> None:
+    await pool.execute(
+        "UPDATE webhook_deliveries SET status = 'pending' WHERE status = 'sending'"
+    )
+
+
+async def claim_due(pool: asyncpg.Pool, limit: int) -> list[asyncpg.Record]:
+    return await pool.fetch(_CLAIM, limit)
+
+
+async def save_body(pool: asyncpg.Pool, delivery_id: int, body: str) -> None:
+    await pool.execute(
+        "UPDATE webhook_deliveries SET body = $2 WHERE id = $1", delivery_id, body
+    )
+
+
+async def finish(  # noqa: PLR0913
+    pool: asyncpg.Pool,
+    delivery_id: int,
+    status: str,
+    response_code: int | None,
+    error: str | None,
+    next_attempt_at: datetime | None = None,
+) -> None:
+    await pool.execute(
+        _FINISH, delivery_id, status, response_code, error, next_attempt_at
+    )
+
+
+async def next_due(pool: asyncpg.Pool) -> datetime | None:
+    return await pool.fetchval(
+        "SELECT min(next_attempt_at) FROM webhook_deliveries WHERE status = 'pending'"
+    )

backend/uv.lock

@@ -149,6 +149,7 @@ dependencies = [
     { name = "fastapi" },
     { name = "fastmcp" },
     { name = "greenlet" },
+    { name = "httpx2" },
     { name = "kurigram" },
     { name = "pydantic-settings" },
     { name = "rich" },
@@ -166,6 +167,7 @@ requires-dist = [
     { name = "fastapi", specifier = ">=0.115.0" },
     { name = "fastmcp", specifier = ">=3.3" },
     { name = "greenlet", specifier = ">=3.1.0" },
+    { name = "httpx2", specifier = ">=2.13.1" },
     { name = "kurigram", specifier = ">=2.2.23" },
     { name = "pydantic-settings", specifier = ">=2.7.0" },
     { name = "rich", specifier = ">=13.9.0" },