diff --git a/backend/.env.example b/backend/.env.example index 098c87f..7ea7425 100644 --- a/backend/.env.example +++ b/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= diff --git a/backend/pyproject.toml b/backend/pyproject.toml index a2d10a5..0b7bc41 100644 --- a/backend/pyproject.toml +++ b/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] diff --git a/backend/src/api/app.py b/backend/src/api/app.py index 4757664..7d34589 100644 --- a/backend/src/api/app.py +++ b/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) diff --git a/backend/src/api/routers/webhooks.py b/backend/src/api/routers/webhooks.py new file mode 100644 index 0000000..a07ff32 --- /dev/null +++ b/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) diff --git a/backend/src/api/webhooks.py b/backend/src/api/webhooks.py new file mode 100644 index 0000000..fe36757 --- /dev/null +++ b/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() diff --git a/backend/src/utils/env.py b/backend/src/utils/env.py index f945eb6..ccbdd86 100644 --- a/backend/src/utils/env.py +++ b/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): diff --git a/backend/src/utils/read/webhooks.py b/backend/src/utils/read/webhooks.py new file mode 100644 index 0000000..aeaf35c --- /dev/null +++ b/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'" + ) diff --git a/backend/uv.lock b/backend/uv.lock index deff984..8357cc5 100644 --- a/backend/uv.lock +++ b/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" },