From 486f38365f131cb5936568ea960f3b8553166a80 Mon Sep 17 00:00:00 2001 From: RhenCloud Date: Sat, 15 Aug 2026 14:58:48 +0800 Subject: [PATCH] feat(queue): async delivery via Cloudflare Queues with retry backoff and DLQ --- AGENTS.md | 13 +- README.md | 8 +- README.zh.md | 8 +- docs/guide/deployment.md | 19 ++- docs/guide/storage.md | 8 +- docs/zh/guide/deployment.md | 19 ++- docs/zh/guide/storage.md | 8 +- server/lib/core/dispatch.ts | 17 ++- server/lib/queue/consumer.ts | 90 +++++++++++++ server/lib/queue/delivery.ts | 133 +++++++++++++++++++ server/lib/types.ts | 1 + server/lib/webhook.ts | 19 +++ server/plugins/queue-consumer.ts | 8 ++ tests/delivery.test.ts | 211 +++++++++++++++++++++++++++++++ tests/queue-consumer.test.ts | 139 ++++++++++++++++++++ wrangler.jsonc | 23 ++++ 16 files changed, 702 insertions(+), 22 deletions(-) create mode 100644 server/lib/queue/consumer.ts create mode 100644 server/lib/queue/delivery.ts create mode 100644 server/plugins/queue-consumer.ts create mode 100644 tests/delivery.test.ts create mode 100644 tests/queue-consumer.test.ts diff --git a/AGENTS.md b/AGENTS.md index c80b89a..38b784a 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -24,6 +24,7 @@ Core pipeline: GitHub Webhook → Worker (verify + filter + format) → Discord - Audit log: every admin operation (logins, group/route/member/invite changes) recorded in D1 `audit_logs`; pruned by the scheduled trigger after `AUDIT_RETENTION_DAYS` (default 90) - Group webhook log channel: optional `Group.logTarget` (Discord channel/thread or Telegram chat/topic) receives one summary message per webhook the group's routes dispatched (event, repo, delivery id, per route×target ✅/❌ outcome, green/red color); best-effort, not recorded in `send_logs` - Per-group forge branding: optional `Group.forgeSources` (a list of `{ host, type: "github" | "gitea", name? }` the group defines itself) labels each message's footer with the first entry whose type matches the event's provider and whose host matches the repository URL's hostname (GitHub matches `github.com`, so two Gitea instances can be `git1.example.com`/`git2.example.com`); the label is the entry's optional `name` (fallback: host); links are derived from the repo URL and footer icons use raster PNGs Discord renders (GitHub's `fluidicon.png`, Gitea's `/assets/img/favicon.png` — `.ico` favicons are silently ignored); Discord shows the footer icon + name, Telegram a linked name +- Delivery queue: when the `QUEUE` binding is present, webhook ingress enqueues one message per event (not per target) to a Cloudflare Queue (`webhooker-delivery`); a Nitro `cloudflare:queue` plugin consumes batches, dispatches, and retries retryable failures (5xx/network/429-exhaustion) with exponential backoff (5s/30s/2m/10m) up to the queue `max_retries`, after which the DLQ (`webhooker-delivery-dlq`) marks the delivery dead. Oversized payloads (>~100 KB) are parked in KV (`queue:payload:*`) and resolved by the consumer; delivery state is tracked in KV (`delivery-state:*`). Without the `QUEUE` binding, dispatch stays inline (existing behavior) - Local dev: wrangler + Miniflare ## Architecture @@ -42,13 +43,17 @@ server/ # Nitro server ├── routes/ # H3 handlers: /health, /webhook[/:groupId], /discord/interactions, /telegram/webhook, │ # /auth/github*, /admin/{login,logout,invite,api/**}, /api/{comment,merge,close,react,richheader} ├── tasks/ # scheduled (cron */5): discord-sync, telegram-sync, audit-prune +├── plugins/ # Nitro plugins: queue-consumer (hooks cloudflare:queue → handleQueueBatch) ├── error-handler.ts # JSON error handler └── lib/ ├── types.ts # Env, Config, Route, Filter, Group, WebhookEvent, NeutralMessage ├── config.ts # loadRoutes/saveRoutes (KV config:routes, cache w/ 60s TTL), loadConfig from env ├── cf.ts # cfEnv(event) — env bindings from event.context.cloudflare ├── http.ts # shared HTTP helpers - ├── webhook.ts # processWebhook/handleWebhook: tenant lookup, provider detect/verify/parse, dedup, scoped dispatch + ├── webhook.ts # processWebhook/handleWebhook: tenant lookup, provider detect/verify/parse, dedup, enqueue (or inline dispatch) + ├── queue/ # Cloudflare Queue delivery pipeline + │ ├── delivery.ts # DeliveryMessage, enqueueWebhook, retry backoff, delivery-state KV, payload-overflow parking + │ └── consumer.ts # handleQueueBatch: resolve payload → dispatch → classify → ack/retry/DLQ ├── core/ │ └── dispatch.ts # Platform-neutral dispatch: match routes → formatEvent → driver.send/edit (recordSend + group filter + per-group webhook log) ├── events/ @@ -133,6 +138,8 @@ tests/ # bun test unit tests (webhook, formatter, discord, tel - Record every dispatch attempt to D1 `send_logs` (route id, event, target, ok/error, duration, error code) - Serve a per-group webhook ingress (`POST /webhook/{groupId}`, per-group secret in KV `tenant:{groupId}`) for Gitea/classic-GitHub/custom senders; only that group's routes fire; dedup keys are provider- and tenant-scoped (`delivery:{provider}:{groupId}:{id}` via `kvIdempotencyStore`) - Issue a per-request correlation id (`requestId`) in webhook responses and dispatch logs +- When the `QUEUE` binding is present, enqueue each verified webhook as a single Queue message (`webhooker-delivery`) instead of dispatching inline; the consumer resolves the payload, re-scopes routes to the tenant group, and dispatches; retryable failures (5xx/network/429-exhaustion) are retried with exponential backoff (5s/30s/2m/10m) up to the queue `max_retries`, then the DLQ marks the delivery dead +- Track delivery state in KV (`delivery-state:*`: pending/processing/delivered/retrying/failed/dead) so redelivered messages are skipped idempotently; oversized payloads are parked in KV (`queue:payload:*`) and deleted after dispatch - Send a per-event summary (event, repo, delivery id, per route×target ✅/❌ outcome) to the group's `logTarget` when configured - Serve `/gh` slash commands + message context-menu commands + PR merge/close buttons + comment modals - Serve Telegram `/gh` commands (login/logout/comment/merge/close) via reply-message parsing @@ -200,6 +207,7 @@ Rule: no functional change ships without its documentation; docs and code must n - **Routes**: KV key `config:routes` (JSON array, empty until configured) - **KV namespace**: Required binding for token/state/config/session storage - **D1 database**: Binding `DB` (database `webhooker`, id `214a0104-3235-47c0-b7bf-ddda95f3c8ac`) for `send_logs` + `audit_logs` + `discord_links` + `telegram_links` tables +- **Queue**: optional `QUEUE` producer binding plus consumers `webhooker-delivery` and its DLQ `webhooker-delivery-dlq` (declared in `wrangler.jsonc`); when absent, webhook dispatch stays inline - **Access control**: `ADMIN_USER_IDS` (super admins), `ALLOW_SELF_SIGNUP` (optional personal group on first login), `AUDIT_RETENTION_DAYS` (default 90) — all plain env vars, not secrets - **Discord**: `DISCORD_PUBLIC_KEY` (Interactions Endpoint signature verification, from Discord Developer Portal) and `DISCORD_APPLICATION_ID` (optional, auto-resolved via `GET /oauth2/applications/@me` when omitted) are required for interactions - **Telegram**: `TELEGRAM_TOKEN` (Bot API token from BotFather) required for Telegram routes; `TELEGRAM_WEBHOOK_SECRET` (optional secret token for `POST /telegram/webhook` verification); avatars are sent as a link-preview card via the built-in `GET /api/richheader` (overridable with `TELEGRAM_RICH_HEADER_HOST`) @@ -215,6 +223,9 @@ bunx wrangler kv namespace create KV # Update wrangler.jsonc with KV ID bunx wrangler d1 create webhooker # Update wrangler.jsonc d1_databases with the database ID +bunx wrangler queues create webhooker-delivery +bunx wrangler queues create webhooker-delivery-dlq +# Queues are declared in wrangler.jsonc (QUEUE binding); no env var needed bun run db:migrate:prod # wrangler d1 migrations apply webhooker --remote (migrations/0001..0005) bunx wrangler deploy ``` diff --git a/README.md b/README.md index 5ab702d..abebc0d 100644 --- a/README.md +++ b/README.md @@ -19,13 +19,14 @@ GitHub / Gitea webhook → Discord / Telegram dispatcher. Receives webhook event - **Discord Interactions Endpoint** (Ed25519-verified) for `/gh` slash commands, message context-menu commands, PR merge/close buttons, and comment modals - **Telegram `/gh` commands** (login/logout/comment/merge/close) via the Telegram webhook, with avatar link-preview cards - Cloudflare KV for token/state/config/session storage + D1 for send logs and platform account links +- **Async delivery via Cloudflare Queues** — when the `QUEUE` binding is present, verified webhooks are enqueued to `webhooker-delivery` and dispatched by a consumer with exponential retry backoff (5s/30s/2m/10m) and a dead-letter queue (`webhooker-delivery-dlq`); oversized payloads are parked in KV. Without the binding, dispatch stays inline - Graceful degradation (webhook-only mode if Discord unavailable) ## Architecture ```text GitHub Webhook → Cloudflare Worker (Nuxt 4 / Nitro) - ├── POST /webhook → verify → dedup → filter → format → Discord (REST) / Telegram (Bot API) + ├── POST /webhook → verify → dedup → enqueue (Queue) → dispatch → Discord (REST) / Telegram (Bot API) ├── POST /discord/interactions → verify (Ed25519) → handle command/button/modal ├── POST /telegram/webhook → verify (secret token) → handle /gh commands ├── GET /auth/github → OAuth flow @@ -37,8 +38,9 @@ GitHub Webhook → Cloudflare Worker (Nuxt 4 / Nitro) - **Cloudflare Worker** — HTTP ingress, signature verification, routing, platform dispatch - **Interactions Endpoint** — HTTPS callback (no Discord Gateway connection, no Durable Object); the bot stays offline and commands are registered via the API -- **KV** — token storage (`token:{userId}`), OAuth state (`state:{hex}`), route config (`config:routes`), group config (`config:groups`), admin sessions (`session:{id}`), delivery dedup (`delivery:{id}`), message-update tracking (`msg:*`) +- **KV** — token storage (`token:{userId}`), OAuth state (`state:{hex}`), route config (`config:routes`), group config (`config:groups`), admin sessions (`session:{id}`), delivery dedup (`delivery:{provider}:{groupId}:{id}`), delivery state (`delivery-state:*`), message-update tracking (`msg:*`) - **D1** — send logs (`send_logs`), Discord↔GitHub links (`discord_links`), Telegram↔GitHub links (`telegram_links`) +- **Queue** — async delivery when `QUEUE` is bound: `webhooker-delivery` (exponential retry) + DLQ `webhooker-delivery-dlq`; oversized payloads parked in KV (`queue:payload:*`) ## Quick Start @@ -156,7 +158,7 @@ Every filter supports plain text, `*`/`?` globs, and `/regex/` patterns (case-in - **GitHub App** — create the app, subscribe to events, configure OAuth, and set the _Setup URL_ for tenant isolation: see [GitHub App Setup](https://webhooker.docs.worldexecute.me/guide/deployment#github-app-setup) - **Discord bot** — create the bot, invite it with `applications.commands` (combined permission integer `274877910016`), and configure the Interactions Endpoint: see [Discord Bot Setup](https://webhooker.docs.worldexecute.me/guide/deployment#discord-bot-setup). The bot never connects to the Discord Gateway, so it shows as **offline** — messaging is unaffected (always REST). - **Telegram bot** — create the bot with [@BotFather](https://t.me/BotFather), set `TELEGRAM_TOKEN` (optional `TELEGRAM_WEBHOOK_SECRET`); the webhook is synced automatically by the scheduled trigger: see [Telegram Bot Setup](https://webhooker.docs.worldexecute.me/guide/deployment#telegram-bot-setup) -- **Deployment** — KV namespace, D1 database + migrations, secrets, and deploy: see the [Deployment guide](https://webhooker.docs.worldexecute.me/guide/deployment) +- **Deployment** — KV namespace, D1 database + migrations, optional Queues, secrets, and deploy: see the [Deployment guide](https://webhooker.docs.worldexecute.me/guide/deployment) ### Bot Commands (comment on GitHub as yourself) diff --git a/README.zh.md b/README.zh.md index 91dd8cd..243da9f 100644 --- a/README.zh.md +++ b/README.zh.md @@ -19,13 +19,14 @@ GitHub / Gitea webhook → Discord / Telegram 分发服务。通过 Cloudflare W - **Discord Interactions Endpoint**（Ed25519 验签）支持 `/gh` 斜杠命令、消息右键菜单命令、PR 合并/关闭按钮与评论 modal - **Telegram `/gh` 命令**（login/logout/comment/merge/close），通过 Telegram webhook 接收，头像以链接预览卡片呈现 - Cloudflare KV 存储 token/状态/配置/会话 + D1 存储发送日志与平台账号绑定 +- **Cloudflare Queues 异步投递** —— 绑定 `QUEUE` 时，已验签的 webhook 会入队到 `webhooker-delivery`，由消费者分发，带指数退避重试（5s/30s/2m/10m）与死信队列（`webhooker-delivery-dlq`）；超大负载暂存于 KV。未绑定则保持同步分发 - 优雅降级（Discord 不可用时仅 webhook 模式） ## 架构 ```text GitHub Webhook → Cloudflare Worker (Nuxt 4 / Nitro) - ├── POST /webhook → 验证 → 去重 → 过滤 → 格式化 → Discord (REST) / Telegram (Bot API) + ├── POST /webhook → 验证 → 去重 → 入队 (Queue) → 分发 → Discord (REST) / Telegram (Bot API) ├── POST /discord/interactions → 验证 (Ed25519) → 处理命令/按钮/modal ├── POST /telegram/webhook → 验证 (secret token) → 处理 /gh 命令 ├── GET /auth/github → OAuth 流程 @@ -37,8 +38,9 @@ GitHub Webhook → Cloudflare Worker (Nuxt 4 / Nitro) - **Cloudflare Worker** — HTTP 入口、签名验证、路由分发 - **Interactions Endpoint** — HTTPS 回调（无 Discord Gateway 连接、无 Durable Object）；bot 保持离线，命令通过 API 注册 -- **KV** — Token 存储（`token:{userId}`）、OAuth state（`state:{hex}`）、路由配置（`config:routes`）、分组配置（`config:groups`）、管理员会话（`session:{id}`）、投递去重（`delivery:{id}`）、消息更新追踪（`msg:*`） +- **KV** — Token 存储（`token:{userId}`）、OAuth state（`state:{hex}`）、路由配置（`config:routes`）、分组配置（`config:groups`）、管理员会话（`session:{id}`）、投递去重（`delivery:{provider}:{groupId}:{id}`）、投递状态（`delivery-state:*`）、消息更新追踪（`msg:*`） - **D1** — 发送日志（`send_logs`）、Discord↔GitHub 绑定（`discord_links`）、Telegram↔GitHub 绑定（`telegram_links`） +- **Queue** — 绑定 `QUEUE` 时异步投递：`webhooker-delivery`（指数退避重试）+ 死信队列 `webhooker-delivery-dlq`；超大负载暂存于 KV（`queue:payload:*`） ## 快速开始 @@ -156,7 +158,7 @@ bunx wrangler dev # 启动本地开发服务器 - **GitHub App** — 创建应用、订阅事件、配置 OAuth 与 _Setup URL_（租户隔离）：见 [GitHub App 配置](https://webhooker.docs.worldexecute.me/zh/guide/deployment#github-app-设置) - **Discord 机器人** — 创建机器人、以 `applications.commands` scope 邀请（组合权限整数 `274877910016`）、配置 Interactions Endpoint：见 [Discord Bot 配置](https://webhooker.docs.worldexecute.me/zh/guide/deployment#discord-bot-设置)。bot 从不连接 Discord Gateway，因此显示为**离线**——消息推送不受影响（始终走 REST）。 - **Telegram 机器人** — 用 [@BotFather](https://t.me/BotFather) 创建机器人，设置 `TELEGRAM_TOKEN`（可选 `TELEGRAM_WEBHOOK_SECRET`）；webhook 由定时任务自动同步：见 [Telegram 机器人配置](https://webhooker.docs.worldexecute.me/zh/guide/deployment#telegram-机器人配置) -- **部署** — KV 命名空间、D1 数据库与迁移、密钥、部署：见[部署指南](https://webhooker.docs.worldexecute.me/zh/guide/deployment) +- **部署** — KV 命名空间、D1 数据库与迁移、可选 Queues、密钥、部署：见[部署指南](https://webhooker.docs.worldexecute.me/zh/guide/deployment) ### Bot 指令（以本人身份评论 GitHub） diff --git a/docs/guide/deployment.md b/docs/guide/deployment.md index 87f01dd..650f801 100644 --- a/docs/guide/deployment.md +++ b/docs/guide/deployment.md @@ -85,7 +85,18 @@ bunx wrangler d1 execute webhooker --remote --file ./migrations/0005_audit_logs. ::: -### 4. Deploy +### 4. Create Queues (Optional) + +The `QUEUE` binding routes webhook delivery through Cloudflare Queues (async dispatch with retry backoff and a dead-letter queue). Skip this step to keep dispatch inline (synchronous). + +```bash +bunx wrangler queues create webhooker-delivery +bunx wrangler queues create webhooker-delivery-dlq +``` + +The queues are already declared in `wrangler.jsonc` (`queues.producers` / `queues.consumers`), so no binding change is needed. The `webhooker-delivery` consumer retries retryable failures with exponential backoff (5s/30s/2m/10m) up to `max_retries`, after which the message is moved to `webhooker-delivery-dlq` and marked dead. + +### 5. Deploy ```bash bunx wrangler deploy @@ -93,13 +104,13 @@ bunx wrangler deploy Your worker is now live at `https://webhooker..workers.dev`. -### 5. Configure GitHub Webhook +### 6. Configure GitHub Webhook 1. Go to your GitHub App settings 2. Set **Webhook URL** to `https://webhooker..workers.dev/webhook` 3. Set **Webhook secret** to match `GITHUB_WEBHOOK_SECRET` -### 6. (Optional) Configure Gitea Webhook +### 7. (Optional) Configure Gitea Webhook 1. In your Gitea repo, go to **Settings → Webhooks → Add Webhook → Gitea** 2. Set **Target URL** to `https://webhooker..workers.dev/webhook` @@ -181,4 +192,4 @@ To use a custom domain instead of `*.workers.dev`: 3. Update `BASE_URL` to match > [!NOTE] -> This project is a Cloudflare Worker. It requires the KV and D1 bindings declared in `wrangler.jsonc`, so it cannot run as a standalone Node/container process. +> This project is a Cloudflare Worker. It requires the KV and D1 bindings declared in `wrangler.jsonc`, so it cannot run as a standalone Node/container process. The Queues binding (`QUEUE`) is optional — without it, webhook dispatch stays inline. diff --git a/docs/guide/storage.md b/docs/guide/storage.md index ab135b2..e36d91e 100644 --- a/docs/guide/storage.md +++ b/docs/guide/storage.md @@ -12,9 +12,11 @@ | `state:{hex}` | `{ redirectTo, expiresAt, discordUserId?, telegramUserId?, telegramChatId? }` | 600 seconds | | `invite:{token}` | `{ groupId, role, expiresAt, createdBy, note? }` | 7 days | | `invite:group:{id}` | Token index per group (keeps invite listing consistent) | Permanent | -| `delivery:{id}` | Webhook delivery id (dedup marker) | 300 seconds | -| `delivery:{groupId}:{id}` | Tenant-scoped delivery dedup for the per-group webhook ingress | 300 seconds | -| `tenant:{groupId}` | Per-group webhook secret (64-char hex, generated from the console) | Permanent | +| `delivery:{provider}:{groupId}:{id}` | Webhook delivery dedup (provider- and tenant-scoped) | 300 seconds | +| `delivery-state:{provider}:{groupId}:{id}` | Queue delivery state (`pending`/`processing`/`delivered`/`retrying`/`failed`/`dead`) | 1 day | +| `queue:payload:{provider}:{groupId}:{id}` | Oversized webhook payload parked for the queue consumer | 1 day | +| `nonce:{nonce}` | Custom-webhook replay protection nonce (single use) | 600 seconds | +| `tenant:{groupId}` | Per-group webhook secret (64-char hex, generated from the console) | Permanent | | `msg:{routeId}:{key}:{target}` | Message id tracking for in-place updates (e.g. `workflow_run` / `check_run`) | 7 days | | `cmd:guild:{id}` | Guild id whose commands were registered (dedup) | Permanent | | `cmd:registered:global` | Global command registration marker (dedup) | 1 day | diff --git a/docs/zh/guide/deployment.md b/docs/zh/guide/deployment.md index a6f67b9..1902a14 100644 --- a/docs/zh/guide/deployment.md +++ b/docs/zh/guide/deployment.md @@ -85,7 +85,18 @@ bunx wrangler d1 execute webhooker --remote --file ./migrations/0005_audit_logs. ::: -### 4. 部署 +### 4. 创建队列（可选） + +`QUEUE` 绑定会通过 Cloudflare Queues 投递 webhook（异步分发，带重试退避与死信队列）。跳过此步则保持同步内联分发。 + +```bash +bunx wrangler queues create webhooker-delivery +bunx wrangler queues create webhooker-delivery-dlq +``` + +队列已在 `wrangler.jsonc` 中声明（`queues.producers` / `queues.consumers`），无需修改绑定。`webhooker-delivery` 消费者对可重试失败做指数退避重试（5s/30s/2m/10m），达到 `max_retries` 后消息进入 `webhooker-delivery-dlq` 并标记为 dead。 + +### 5. 部署 ```bash bunx wrangler deploy @@ -93,13 +104,13 @@ bunx wrangler deploy Worker 现在可通过 `https://webhooker..workers.dev` 访问。 -### 5. 配置 GitHub Webhook +### 6. 配置 GitHub Webhook 1. 进入 GitHub App 设置页面 2. 设置 **Webhook URL** 为 `https://webhooker..workers.dev/webhook` 3. 设置 **Webhook secret** 与 `GITHUB_WEBHOOK_SECRET` 一致 -### 6.（可选）配置 Gitea Webhook +### 7.（可选）配置 Gitea Webhook 1. 在 Gitea 仓库中进入 **设置 → Web 钩子 → 添加 Web 钩子 → Gitea** 2. 设置 **目标 URL** 为 `https://webhooker..workers.dev/webhook` @@ -181,4 +192,4 @@ Worker 现在可通过 `https://webhooker..workers.dev` 访问 3. 更新 `BASE_URL` 以匹配 > [!NOTE] -> 本项目是一个 Cloudflare Worker，依赖 `wrangler.jsonc` 中声明的 KV 与 D1 绑定，无法作为独立的 Node/容器进程运行。 +> 本项目是一个 Cloudflare Worker，依赖 `wrangler.jsonc` 中声明的 KV 与 D1 绑定，无法作为独立的 Node/容器进程运行。Queues 绑定（`QUEUE`）可选——未绑定时 webhook 分发保持内联同步。 diff --git a/docs/zh/guide/storage.md b/docs/zh/guide/storage.md index eddfee5..174b9f8 100644 --- a/docs/zh/guide/storage.md +++ b/docs/zh/guide/storage.md @@ -12,9 +12,11 @@ | `state:{hex}` | `{ redirectTo, expiresAt, discordUserId?, telegramUserId?, telegramChatId? }` | 600 秒 | | `invite:{token}` | `{ groupId, role, expiresAt, createdBy, note? }` | 7 天 | | `invite:group:{id}` | 每组的 Token 索引（保证邀请列表一致性） | 永久 | -| `delivery:{id}` | Webhook 投递 id（去重标记） | 300 秒 | -| `delivery:{groupId}:{id}` | 分组级 webhook 入口的租户级投递去重 | 300 秒 | -| `tenant:{groupId}` | 分组 webhook secret（64 位 hex，控制台生成） | 永久 | +| `delivery:{provider}:{groupId}:{id}` | Webhook 投递去重（按 provider 与租户隔离） | 300 秒 | +| `delivery-state:{provider}:{groupId}:{id}` | 队列投递状态（`pending`/`processing`/`delivered`/`retrying`/`failed`/`dead`） | 1 天 | +| `queue:payload:{provider}:{groupId}:{id}` | 暂存供队列消费者读取的超大 webhook 负载 | 1 天 | +| `nonce:{nonce}` | 自定义 webhook 重放防护 nonce（一次性） | 600 秒 | +| `tenant:{groupId}` | 分组 webhook secret（64 位 hex，控制台生成） | 永久 | | `msg:{routeId}:{key}:{target}` | 原地更新用消息 id 追踪（如 `workflow_run` / `check_run`） | 7 天 | | `cmd:guild:{id}` | 已注册命令的服务器 id（去重） | 永久 | | `cmd:registered:global` | 全局命令注册标记（去重） | 1 天 | diff --git a/server/lib/core/dispatch.ts b/server/lib/core/dispatch.ts index 0b686ef..cf782da 100644 --- a/server/lib/core/dispatch.ts +++ b/server/lib/core/dispatch.ts @@ -13,6 +13,7 @@ import { } from "../web/groups"; import { getDriver } from "../drivers"; import type { SendResult } from "../drivers/types"; +import type { DispatchFailure, DispatchSummary } from "../queue/delivery"; /** One dispatch attempt (route × target), collected for the group webhook log. */ interface DispatchAttempt { @@ -22,6 +23,8 @@ interface DispatchAttempt { target: string; ok: boolean; error?: string; + errorCode?: string; + status?: number; } export async function dispatchEvent( @@ -29,7 +32,7 @@ export async function dispatchEvent( event: WebhookEvent, env: Env, groups?: Group[], -): Promise { +): Promise { const loadedGroups = groups ?? (await loadGroups(env.KV)); const groupById = new Map(loadedGroups.map((g) => [g.id, g])); @@ -76,6 +79,16 @@ export async function dispatchEvent( await sendGroupLogs(attempts); + const failures: DispatchFailure[] = attempts + .filter((a) => !a.ok) + .map((a) => ({ + target: a.target, + error: a.error, + errorCode: a.errorCode, + status: a.status, + })); + return { attempts: attempts.length, failures }; + async function sendGroupLogs(list: DispatchAttempt[]): Promise { const byGroup = new Map(); for (const a of list) { @@ -309,6 +322,8 @@ export async function dispatchEvent( target: targetStr, ok: false, error, + errorCode: result.errorCode, + status: result.status, }); await recordSend(env.DB, { ...base, diff --git a/server/lib/queue/consumer.ts b/server/lib/queue/consumer.ts new file mode 100644 index 0000000..8cdaa14 --- /dev/null +++ b/server/lib/queue/consumer.ts @@ -0,0 +1,90 @@ +import type { Env, WebhookEvent, WebhookProvider } from "../types"; +import { dispatchEvent } from "../core/dispatch"; +import { loadConfig } from "../config"; +import { loadGroups } from "../web/groups"; +import { log } from "../lib/log"; +import { + DELIVERY_DLQ, + type DeliveryMessage, + classifyDelivery, + deliveryStateKey, + discardPayload, + getDeliveryState, + resolvePayload, + retryDelay, + setDeliveryState, +} from "./delivery"; + +export async function handleQueueBatch( + batch: MessageBatch, + env: Env, +): Promise { + for (const message of batch.messages) { + const body = message.body; + if (batch.queue === DELIVERY_DLQ) { + await markDead(env, body); + message.ack(); + continue; + } + await processMessage(env, body, message); + } +} + +async function processMessage( + env: Env, + body: DeliveryMessage, + message: Message, +): Promise { + const key = deliveryStateKey(body.provider, body.groupId, body.deliveryId); + const prior = await getDeliveryState(env, key); + if (prior === "delivered" || prior === "dead") { + message.ack(); + return; + } + + await setDeliveryState(env, key, "processing"); + + const payload = await resolvePayload(env, body); + const event: WebhookEvent = { + event: body.event, + payload, + deliveryId: body.deliveryId, + provider: body.provider as WebhookProvider, + installationId: body.installationId, + }; + + try { + const config = await loadConfig(env); + if (body.groupId) { + config.routes = config.routes.filter((r) => r.groupId === body.groupId); + } + const groups = await loadGroups(env.KV); + const summary = await dispatchEvent(config, event, env, groups); + const { failed, retryable } = classifyDelivery(summary); + + if (!failed) { + await setDeliveryState(env, key, "delivered"); + await discardPayload(env, body); + message.ack(); + return; + } + if (retryable) { + await setDeliveryState(env, key, "retrying"); + message.retry({ delaySeconds: retryDelay(message.attempts) }); + return; + } + await setDeliveryState(env, key, "failed"); + await discardPayload(env, body); + message.ack(); + } catch (err) { + log.error({ deliveryId: body.deliveryId, err }, "Queue delivery failed"); + await setDeliveryState(env, key, "retrying"); + message.retry({ delaySeconds: retryDelay(message.attempts) }); + } +} + +async function markDead(env: Env, body: DeliveryMessage): Promise { + const key = deliveryStateKey(body.provider, body.groupId, body.deliveryId); + await setDeliveryState(env, key, "dead"); + await discardPayload(env, body); +} diff --git a/server/lib/queue/delivery.ts b/server/lib/queue/delivery.ts new file mode 100644 index 0000000..ed12e7a --- /dev/null +++ b/server/lib/queue/delivery.ts @@ -0,0 +1,133 @@ +import type { Env } from "../types"; + +export type DeliveryStatus = + "pending" | "processing" | "delivered" | "retrying" | "failed" | "dead"; + +export interface DeliveryMessage { + deliveryId: string; + groupId?: string; + provider: string; + event: string; + payload?: Record; + payloadRef?: string; + installationId?: number; + receivedAt: number; + requestId?: string; +} + +export interface DispatchFailure { + target: string; + error?: string; + errorCode?: string; + status?: number; +} + +export interface DispatchSummary { + attempts: number; + failures: DispatchFailure[]; +} + +export const DELIVERY_QUEUE = "webhooker-delivery"; +export const DELIVERY_DLQ = "webhooker-delivery-dlq"; + +const MAX_QUEUE_MESSAGE_BYTES = 100_000; +const PAYLOAD_KV_TTL_SECONDS = 60 * 60 * 24; +const STATE_KV_TTL_SECONDS = 60 * 60 * 24; + +const RETRYABLE_ERROR_CODES = new Set(["DISCORD_5XX", "TELEGRAM_5XX", "NETWORK", "RETRIES"]); + +const RETRY_DELAYS_SECONDS = [5, 30, 120, 600]; + +export function isRetryableError(code?: string): boolean { + return code == null || RETRYABLE_ERROR_CODES.has(code); +} + +export function classifyDelivery(summary: DispatchSummary): { + failed: boolean; + retryable: boolean; +} { + if (summary.failures.length === 0) return { failed: false, retryable: false }; + const retryable = summary.failures.every((f) => isRetryableError(f.errorCode)); + const permanent = summary.failures.some((f) => !isRetryableError(f.errorCode)); + return { failed: true, retryable: retryable && !permanent }; +} + +export function retryDelay(attempt: number): number { + if (attempt < 1) return RETRY_DELAYS_SECONDS[0]; + const idx = Math.min(attempt - 1, RETRY_DELAYS_SECONDS.length - 1); + return RETRY_DELAYS_SECONDS[idx]; +} + +function scopeKey(provider: string, groupId: string | undefined, deliveryId: string): string { + return `${provider}:${groupId ?? "global"}:${deliveryId}`; +} + +export function deliveryStateKey( + provider: string, + groupId: string | undefined, + deliveryId: string, +): string { + return `delivery-state:${scopeKey(provider, groupId, deliveryId)}`; +} + +function payloadKey(provider: string, groupId: string | undefined, deliveryId: string): string { + return `queue:payload:${scopeKey(provider, groupId, deliveryId)}`; +} + +export async function setDeliveryState( + env: Env, + key: string, + status: DeliveryStatus, +): Promise { + await env.KV.put(key, JSON.stringify({ status, at: Date.now() }), { + expirationTtl: STATE_KV_TTL_SECONDS, + }); +} + +export async function getDeliveryState(env: Env, key: string): Promise { + const raw = await env.KV.get(key); + if (!raw) return null; + try { + return (JSON.parse(raw) as { status?: DeliveryStatus }).status ?? null; + } catch { + return null; + } +} + +export async function enqueueWebhook(env: Env, message: DeliveryMessage): Promise { + const queue = env.QUEUE; + if (!queue) throw new Error("QUEUE binding is not configured"); + const { payload, ...rest } = message; + const direct: DeliveryMessage = { ...rest, payload }; + if (JSON.stringify(direct).length <= MAX_QUEUE_MESSAGE_BYTES) { + await queue.send(direct); + return; + } + const payloadRef = payloadKey(message.provider, message.groupId, message.deliveryId); + await env.KV.put(payloadRef, JSON.stringify(payload ?? {}), { + expirationTtl: PAYLOAD_KV_TTL_SECONDS, + }); + await queue.send({ ...rest, payloadRef }); +} + +export async function resolvePayload( + env: Env, + message: DeliveryMessage, +): Promise> { + if (message.payload) return message.payload; + if (message.payloadRef) { + const raw = await env.KV.get(message.payloadRef); + if (raw) { + try { + return JSON.parse(raw) as Record; + } catch { + return {}; + } + } + } + return {}; +} + +export async function discardPayload(env: Env, message: DeliveryMessage): Promise { + if (message.payloadRef) await env.KV.delete(message.payloadRef); +} diff --git a/server/lib/types.ts b/server/lib/types.ts index 95ca0d5..4e41c2b 100644 --- a/server/lib/types.ts +++ b/server/lib/types.ts @@ -27,6 +27,7 @@ export interface Env { ASSETS?: Fetcher; KV: KVNamespace; DB: D1Database; + QUEUE?: Queue; } export interface Config { diff --git a/server/lib/webhook.ts b/server/lib/webhook.ts index d115ac7..4be55e4 100644 --- a/server/lib/webhook.ts +++ b/server/lib/webhook.ts @@ -11,6 +11,7 @@ import { cfEnv, cfWaitUntil, headersFrom } from "./cf"; import { log } from "./lib/log"; import { deliveryKey, kvIdempotencyStore } from "./lib/idempotency"; import { newCorrelationId } from "./lib/correlation"; +import { enqueueWebhook, type DeliveryMessage } from "./queue/delivery"; const MAX_BODY_SIZE = 1024 * 1024; @@ -135,6 +136,24 @@ export async function processWebhook( config.routes = config.routes.filter((r) => r.groupId === tenantId); } + if (env.QUEUE) { + const message: DeliveryMessage = { + deliveryId: event.deliveryId ?? requestId, + groupId: tenantId, + provider: provider.id, + event: event.event, + payload: event.payload, + installationId: event.installationId, + receivedAt: Date.now(), + requestId, + }; + const enqueue = enqueueWebhook(env, message).catch((err) => + log.error({ requestId, err }, "Failed to enqueue webhook"), + ); + waitUntil(enqueue); + return { status: 200, body: { ok: true, requestId } }; + } + const dispatch = dispatchEvent(config, event, env, groups).catch((err) => log.error({ requestId, err }, "Dispatch failed"), ); diff --git a/server/plugins/queue-consumer.ts b/server/plugins/queue-consumer.ts new file mode 100644 index 0000000..5751249 --- /dev/null +++ b/server/plugins/queue-consumer.ts @@ -0,0 +1,8 @@ +import type { Env } from "../lib/types"; +import { handleQueueBatch } from "../lib/queue/consumer"; + +export default defineNitroPlugin((nitroApp) => { + nitroApp.hooks.hook("cloudflare:queue", async ({ batch, env }) => { + await handleQueueBatch(batch, env as Env); + }); +}); diff --git a/tests/delivery.test.ts b/tests/delivery.test.ts new file mode 100644 index 0000000..272b64b --- /dev/null +++ b/tests/delivery.test.ts @@ -0,0 +1,211 @@ +import { describe, it, expect } from "bun:test"; +import { + classifyDelivery, + deliveryStateKey, + discardPayload, + enqueueWebhook, + getDeliveryState, + isRetryableError, + resolvePayload, + retryDelay, + setDeliveryState, + type DeliveryMessage, + type DispatchSummary, +} from "../server/lib/queue/delivery"; +import type { Env } from "../server/lib/types"; + +function createMockKV(): { kv: KVNamespace; store: Map } { + const store = new Map(); + const kv = { + get: async (key: string) => (store.has(key) ? store.get(key)! : null), + put: async (key: string, value: string) => { + store.set(key, value); + }, + delete: async (key: string) => { + store.delete(key); + }, + list: async () => ({ keys: [], list_complete: true, cacheStatus: null }), + } as unknown as KVNamespace; + return { kv, store }; +} + +function createMockQueue(): { queue: Queue; sent: DeliveryMessage[] } { + const sent: DeliveryMessage[] = []; + const queue = { + send: async (message: DeliveryMessage) => { + sent.push(message); + }, + sendBatch: async () => {}, + } as unknown as Queue; + return { queue, sent }; +} + +function createEnv(kv: KVNamespace, queue?: Queue): Env { + return { + GITHUB_WEBHOOK_SECRET: "secret", + KV: kv, + DB: {} as D1Database, + QUEUE: queue, + }; +} + +function message(overrides: Partial = {}): DeliveryMessage { + return { + deliveryId: "d1", + provider: "github", + event: "push", + payload: { ref: "refs/heads/main" }, + receivedAt: Date.now(), + ...overrides, + }; +} + +describe("isRetryableError", () => { + it("treats undefined as retryable", () => { + expect(isRetryableError(undefined)).toBe(true); + }); + it("treats 5xx/network/retries as retryable", () => { + expect(isRetryableError("DISCORD_5XX")).toBe(true); + expect(isRetryableError("TELEGRAM_5XX")).toBe(true); + expect(isRetryableError("NETWORK")).toBe(true); + expect(isRetryableError("RETRIES")).toBe(true); + }); + it("treats 4xx and config errors as permanent", () => { + expect(isRetryableError("DISCORD_ERROR")).toBe(false); + expect(isRetryableError("TELEGRAM_ERROR")).toBe(false); + expect(isRetryableError("NO_TOKEN")).toBe(false); + expect(isRetryableError("NO_TARGET")).toBe(false); + }); +}); + +describe("retryDelay", () => { + it("follows the exponential backoff schedule", () => { + expect(retryDelay(1)).toBe(5); + expect(retryDelay(2)).toBe(30); + expect(retryDelay(3)).toBe(120); + expect(retryDelay(4)).toBe(600); + expect(retryDelay(10)).toBe(600); + }); + it("clamps sub-one attempts to the first delay", () => { + expect(retryDelay(0)).toBe(5); + }); +}); + +describe("classifyDelivery", () => { + it("reports success when no failures", () => { + expect(classifyDelivery({ attempts: 1, failures: [] })).toEqual({ + failed: false, + retryable: false, + }); + }); + it("is retryable when every failure is retryable", () => { + const summary: DispatchSummary = { + attempts: 2, + failures: [{ target: "c1", errorCode: "DISCORD_5XX" }], + }; + expect(classifyDelivery(summary)).toEqual({ failed: true, retryable: true }); + }); + it("is not retryable when any failure is permanent", () => { + const summary: DispatchSummary = { + attempts: 2, + failures: [ + { target: "c1", errorCode: "DISCORD_5XX" }, + { target: "c2", errorCode: "DISCORD_ERROR" }, + ], + }; + expect(classifyDelivery(summary)).toEqual({ failed: true, retryable: false }); + }); +}); + +describe("delivery state", () => { + it("round-trips status via KV", async () => { + const { kv } = createMockKV(); + const env = createEnv(kv); + const key = deliveryStateKey("github", "g1", "d1"); + await setDeliveryState(env, key, "processing"); + expect(await getDeliveryState(env, key)).toBe("processing"); + await setDeliveryState(env, key, "delivered"); + expect(await getDeliveryState(env, key)).toBe("delivered"); + }); + + it("returns null for an unknown key", async () => { + const { kv } = createMockKV(); + expect(await getDeliveryState(createEnv(kv), "delivery-state:nope")).toBeNull(); + }); + + it("scopes state keys by provider, group and delivery id", () => { + expect(deliveryStateKey("github", "g1", "d1")).toBe("delivery-state:github:g1:d1"); + expect(deliveryStateKey("github", undefined, "d1")).toBe("delivery-state:github:global:d1"); + }); +}); + +describe("enqueueWebhook", () => { + it("sends the message directly when it fits", async () => { + const { kv } = createMockKV(); + const { queue, sent } = createMockQueue(); + const env = createEnv(kv, queue); + await enqueueWebhook(env, message()); + expect(sent).toHaveLength(1); + expect(sent[0].deliveryId).toBe("d1"); + expect(sent[0].payload).toEqual({ ref: "refs/heads/main" }); + expect(sent[0].payloadRef).toBeUndefined(); + }); + + it("stores the payload in KV when the message overflows", async () => { + const { kv, store } = createMockKV(); + const { queue, sent } = createMockQueue(); + const env = createEnv(kv, queue); + const large = "x".repeat(150_000); + await enqueueWebhook(env, message({ payload: { data: large } })); + expect(sent).toHaveLength(1); + expect(sent[0].payload).toBeUndefined(); + expect(sent[0].payloadRef).toBe("queue:payload:github:global:d1"); + const stored = store.get("queue:payload:github:global:d1"); + expect(stored).toBeDefined(); + expect(JSON.parse(stored!).data).toBe(large); + }); + + it("throws when the QUEUE binding is missing", async () => { + const { kv } = createMockKV(); + await expect(enqueueWebhook(createEnv(kv), message())).rejects.toThrow( + "QUEUE binding is not configured", + ); + }); +}); + +describe("resolvePayload / discardPayload", () => { + it("reads the payload from the message", async () => { + const { kv } = createMockKV(); + const env = createEnv(kv); + expect(await resolvePayload(env, message())).toEqual({ ref: "refs/heads/main" }); + }); + + it("reads the payload from KV when a payloadRef is set", async () => { + const { kv, store } = createMockKV(); + store.set("queue:payload:github:global:d1", JSON.stringify({ a: 1 })); + const env = createEnv(kv); + expect( + await resolvePayload( + env, + message({ payload: undefined, payloadRef: "queue:payload:github:global:d1" }), + ), + ).toEqual({ a: 1 }); + }); + + it("returns an empty object when the payloadRef is missing", async () => { + const { kv } = createMockKV(); + expect( + await resolvePayload( + createEnv(kv), + message({ payload: undefined, payloadRef: "queue:payload:missing" }), + ), + ).toEqual({}); + }); + + it("deletes the payloadRef on discard", async () => { + const { kv, store } = createMockKV(); + store.set("queue:payload:github:global:d1", "{}"); + await discardPayload(createEnv(kv), message({ payloadRef: "queue:payload:github:global:d1" })); + expect(store.has("queue:payload:github:global:d1")).toBe(false); + }); +}); diff --git a/tests/queue-consumer.test.ts b/tests/queue-consumer.test.ts new file mode 100644 index 0000000..a1f9f1e --- /dev/null +++ b/tests/queue-consumer.test.ts @@ -0,0 +1,139 @@ +import { describe, it, expect, beforeEach, mock } from "bun:test"; +import { handleQueueBatch } from "../server/lib/queue/consumer"; +import { invalidateConfigCache } from "../server/lib/config"; +import type { Env } from "../server/lib/types"; +import type { DeliveryMessage, DispatchSummary } from "../server/lib/queue/delivery"; + +let summary: DispatchSummary; +let dispatchCalls: number; + +mock.module("../server/lib/core/dispatch", () => ({ + dispatchEvent: async (): Promise => { + dispatchCalls += 1; + return summary; + }, +})); + +function createMockKV(): { kv: KVNamespace; store: Map } { + const store = new Map(); + const kv = { + get: async (key: string) => (store.has(key) ? store.get(key)! : null), + put: async (key: string, value: string) => { + store.set(key, value); + }, + delete: async (key: string) => { + store.delete(key); + }, + list: async () => ({ keys: [], list_complete: true, cacheStatus: null }), + } as unknown as KVNamespace; + return { kv, store }; +} + +function createEnv(kv: KVNamespace): Env { + return { + GITHUB_WEBHOOK_SECRET: "secret", + KV: kv, + DB: {} as D1Database, + }; +} + +interface FakeMessage { + body: DeliveryMessage; + attempts: number; + acked: boolean; + retried: { delaySeconds?: number } | null; + ack: () => void; + retry: (opts?: { delaySeconds?: number }) => void; +} + +function makeMessage(body: DeliveryMessage, attempts = 1): FakeMessage { + return { + body, + attempts, + acked: false, + retried: null, + ack(): void { + this.acked = true; + }, + retry(opts): void { + this.retried = opts ?? {}; + }, + }; +} + +function makeBatch(queue: string, messages: FakeMessage[]): MessageBatch { + return { + queue, + messages: messages as never, + ackAll: () => {}, + retryAll: () => {}, + } as unknown as MessageBatch; +} + +function body(overrides: Partial = {}): DeliveryMessage { + return { + deliveryId: "d1", + provider: "github", + event: "push", + payload: { ref: "refs/heads/main" }, + receivedAt: Date.now(), + ...overrides, + }; +} + +const STATE_KEY = "delivery-state:github:global:d1"; + +describe("handleQueueBatch", () => { + beforeEach(() => { + invalidateConfigCache(); + summary = { attempts: 0, failures: [] }; + dispatchCalls = 0; + }); + + it("marks DLQ messages dead and acks them", async () => { + const { kv, store } = createMockKV(); + const msg = makeMessage(body()); + await handleQueueBatch(makeBatch("webhooker-delivery-dlq", [msg]), createEnv(kv)); + expect(msg.acked).toBe(true); + expect(JSON.parse(store.get(STATE_KEY)!)).toMatchObject({ status: "dead" }); + }); + + it("acks and marks delivered on success", async () => { + const { kv, store } = createMockKV(); + summary = { attempts: 1, failures: [] }; + const msg = makeMessage(body()); + await handleQueueBatch(makeBatch("webhooker-delivery", [msg]), createEnv(kv)); + expect(msg.acked).toBe(true); + expect(msg.retried).toBeNull(); + expect(JSON.parse(store.get(STATE_KEY)!)).toMatchObject({ status: "delivered" }); + }); + + it("retries with backoff on a retryable failure", async () => { + const { kv, store } = createMockKV(); + summary = { attempts: 2, failures: [{ target: "c1", errorCode: "DISCORD_5XX" }] }; + const msg = makeMessage(body(), 1); + await handleQueueBatch(makeBatch("webhooker-delivery", [msg]), createEnv(kv)); + expect(msg.acked).toBe(false); + expect(msg.retried).toEqual({ delaySeconds: 5 }); + expect(JSON.parse(store.get(STATE_KEY)!)).toMatchObject({ status: "retrying" }); + }); + + it("acks and marks failed on a permanent failure", async () => { + const { kv, store } = createMockKV(); + summary = { attempts: 2, failures: [{ target: "c1", errorCode: "DISCORD_ERROR" }] }; + const msg = makeMessage(body()); + await handleQueueBatch(makeBatch("webhooker-delivery", [msg]), createEnv(kv)); + expect(msg.acked).toBe(true); + expect(msg.retried).toBeNull(); + expect(JSON.parse(store.get(STATE_KEY)!)).toMatchObject({ status: "failed" }); + }); + + it("skips already-delivered deliveries without re-dispatching", async () => { + const { kv, store } = createMockKV(); + store.set(STATE_KEY, JSON.stringify({ status: "delivered", at: Date.now() })); + const msg = makeMessage(body()); + await handleQueueBatch(makeBatch("webhooker-delivery", [msg]), createEnv(kv)); + expect(msg.acked).toBe(true); + expect(dispatchCalls).toBe(0); + }); +}); diff --git a/wrangler.jsonc b/wrangler.jsonc index 9cc1af4..7b5b8f3 100644 --- a/wrangler.jsonc +++ b/wrangler.jsonc @@ -29,4 +29,27 @@ "database_id": "214a0104-3235-47c0-b7bf-ddda95f3c8ac" }, ], + "queues": { + "producers": [ + { + "binding": "QUEUE", + "queue": "webhooker-delivery" + } + ], + "consumers": [ + { + "queue": "webhooker-delivery", + "max_batch_size": 10, + "max_batch_timeout": 30, + "max_retries": 5, + "dead_letter_queue": "webhooker-delivery-dlq" + }, + { + "queue": "webhooker-delivery-dlq", + "max_batch_size": 10, + "max_batch_timeout": 30, + "max_retries": 1 + } + ] + }, }