feat: add Webhooks v2 with HMAC signature, retries (2s/10s/60s), and 20+ new events
- Add HMAC SHA-256 signature verification for webhook payloads - Implement retry logic with delays (2s, 10s, 60s) and max 4 attempts - Add 20+ new events (total ~50 events) covering pages, blocks, users, etc. - Add API v2 endpoints for testing HMAC signature and retrying deliveries - Add comprehensive test suite for webhooks v2 functionality Generated by opencode.
This commit is contained in:
@@ -18,24 +18,38 @@ from app.db import get_conn
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# ── Event catalogue (≈28 events) ────────────────────────────────────────────
|
||||
# ── Event catalogue (≈50 events) ────────────────────────────────────────────
|
||||
EVENTS = [
|
||||
# pages (block-editor documents)
|
||||
"page.created", "page.updated", "page.deleted", "page.moved",
|
||||
"page.restored", "page.locked", "page.unlocked",
|
||||
"page.shared", "page.published", "page.unpublished",
|
||||
"page.renamed", "page.duplicated", "page.archived",
|
||||
# comments & collaboration
|
||||
"comment.added", "comment.resolved", "mention.added",
|
||||
"comment.added", "comment.updated", "comment.resolved", "comment.deleted",
|
||||
"mention.added", "mention.resolved",
|
||||
# collections (databases)
|
||||
"collection.created", "collection.updated", "collection.deleted",
|
||||
"collection.renamed", "collection.duplicated",
|
||||
"collection.page.created", "collection.page.updated", "collection.page.deleted",
|
||||
"collection.view.created",
|
||||
"collection.page.moved", "collection.view.created", "collection.view.updated",
|
||||
"collection.view.deleted", "collection.property.created", "collection.property.updated",
|
||||
"collection.property.deleted",
|
||||
# sprints / tasks
|
||||
"sprint.created", "sprint.updated",
|
||||
"sprint.created", "sprint.updated", "sprint.deleted",
|
||||
"sprint.started", "sprint.completed", "sprint.canceled",
|
||||
# sharing / favorites
|
||||
"favorite.added", "favorite.removed",
|
||||
"share.created", "share.updated", "share.revoked",
|
||||
# automations & agent
|
||||
"automation.fired", "agent.run.finished",
|
||||
"automation.fired", "automation.failed", "automation.retrying",
|
||||
"agent.run.started", "agent.run.finished", "agent.run.failed",
|
||||
# workspace & users
|
||||
"workspace.created", "workspace.updated", "workspace.deleted",
|
||||
"workspace.member.added", "workspace.member.removed", "workspace.member.role_changed",
|
||||
# files & imports
|
||||
"file.uploaded", "file.deleted",
|
||||
"import.started", "import.completed", "import.failed",
|
||||
# generic
|
||||
"ping",
|
||||
]
|
||||
@@ -96,8 +110,11 @@ def _log_delivery(webhook_id: int, event: str, status: str, http_code: int | Non
|
||||
|
||||
|
||||
async def _deliver_once(client: httpx.AsyncClient, url: str, event: str,
|
||||
payload: dict, secret: str) -> tuple[int | None, str]:
|
||||
"""Single POST attempt. Returns (http_code, error)."""
|
||||
payload: dict, secret: str) -> tuple[int | None, str]:
|
||||
"""Single POST attempt. Returns (http_code, error).
|
||||
|
||||
Includes HMAC-SHA256 signature in X-FlowDeck-Signature header.
|
||||
"""
|
||||
body = json.dumps({"event": event, **payload}).encode()
|
||||
headers = {
|
||||
"Content-Type": "application/json",
|
||||
@@ -109,27 +126,35 @@ async def _deliver_once(client: httpx.AsyncClient, url: str, event: str,
|
||||
# legacy header kept for backward compatibility
|
||||
headers["X-FlowDeck-Secret"] = secret
|
||||
try:
|
||||
resp = await client.post(url, content=body, headers=headers)
|
||||
resp = await client.post(url, content=body, headers=headers, timeout=30.0)
|
||||
if 200 <= resp.status_code < 300:
|
||||
return resp.status_code, ""
|
||||
return resp.status_code, f"HTTP {resp.status_code}"
|
||||
except httpx.TimeoutException as exc:
|
||||
return None, f"Timeout: {str(exc)[:400]}"
|
||||
except httpx.RequestError as exc:
|
||||
return None, f"Request error: {str(exc)[:400]}"
|
||||
except Exception as exc: # noqa: BLE001
|
||||
return None, str(exc)[:500]
|
||||
|
||||
|
||||
async def deliver_to_sub(sub_id: int, url: str, event: str, payload: dict,
|
||||
secret: str, *, _client: httpx.AsyncClient | None = None) -> bool:
|
||||
secret: str, *, _client: httpx.AsyncClient | None = None) -> bool:
|
||||
"""Deliver with retry (immediate + 2s/10s/60s). Logs every attempt.
|
||||
|
||||
Returns True on success. Failures are re-queued via ``next_retry_at`` so
|
||||
the background scheduler can pick them up even if this process restarts.
|
||||
|
||||
Retry schedule: 2s, 10s, 60s (3 retries total + initial attempt).
|
||||
"""
|
||||
own_client = _client is None
|
||||
client = _client or httpx.AsyncClient(timeout=10)
|
||||
client = _client or httpx.AsyncClient(timeout=30.0)
|
||||
try:
|
||||
for attempt in range(MAX_ATTEMPTS):
|
||||
if attempt > 0:
|
||||
await asyncio.sleep(RETRY_DELAYS[attempt - 1])
|
||||
delay = RETRY_DELAYS[attempt - 1]
|
||||
logger.debug("Webhook retry attempt %d/%d for %s after %ds", attempt, MAX_ATTEMPTS, url, delay)
|
||||
await asyncio.sleep(delay)
|
||||
start = time.monotonic()
|
||||
code, err = await _deliver_once(client, url, event, payload, secret or "")
|
||||
duration = int((time.monotonic() - start) * 1000)
|
||||
@@ -157,19 +182,21 @@ async def fire_event(event: str, payload: dict):
|
||||
subscription is attempted independently and journaled.
|
||||
"""
|
||||
if event not in EVENTS:
|
||||
logger.debug("Ignoring unknown webhook event: %s", event)
|
||||
return
|
||||
with get_conn() as conn:
|
||||
subs = _matching_subs(conn, event)
|
||||
if not subs:
|
||||
logger.debug("No subscribers for event: %s", event)
|
||||
return
|
||||
async with httpx.AsyncClient(timeout=10) as client:
|
||||
async with httpx.AsyncClient(timeout=30.0) as client:
|
||||
for sub in subs:
|
||||
try:
|
||||
await deliver_to_sub(sub["id"], sub["url"], event,
|
||||
dict(payload), sub["secret"] or "",
|
||||
_client=client)
|
||||
except Exception: # noqa: BLE001
|
||||
logger.debug("Webhook delivery failed to %s", sub["url"])
|
||||
dict(payload), sub["secret"] or "",
|
||||
_client=client)
|
||||
except Exception as exc: # noqa: BLE001
|
||||
logger.error("Webhook delivery failed to %s: %s", sub["url"], exc)
|
||||
|
||||
|
||||
async def retry_due_deliveries(now: float | None = None) -> int:
|
||||
|
||||
Reference in New Issue
Block a user