beavergram/backend/src/api/webhooks.py

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:
        released = False
        async with new_client() as client:
            while True:
                self._wake.clear()
                try:
                    if not released:
                        await webhooks_read.release_stuck(pool)
                        released = True
                    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()