From 25ebae4ae59a0e0ced1baf64e2571224b8789798 Mon Sep 17 00:00:00 2001 From: RhenCloud Date: Mon, 17 Aug 2026 15:01:20 +0800 Subject: [PATCH] feat(storage): migrate config, dedup and delivery state to D1 - Move oversized queue payloads from KV to R2 (PAYLOAD binding, webhooks/YYYY/MM/DD/*.json, KV queue:payload:* fallback) - Persist routes/groups to D1 (d1_routes/d1_groups) with memory -> KV -> D1 three-tier cache, seeded from legacy KV config keys - Move webhook dedup (dedup_keys), delivery state (delivery_state) and message tracking (message_tracking) to D1 via canUseD1 probe with automatic KV fallback - Batch send_logs inserts (recordSendBatch) and add group_id/ts index - Add storage-prune scheduled task for expired dedup/state/tracking rows - Add TTL to invite:group:{id} index and audit all ephemeral KV keys - Add D1 indexes for the new tables - Sync AGENTS.md, README.md/zh and docs/ (en/zh) with the new storage layout --- AGENTS.md | 41 +++-- README.md | 16 +- README.zh.md | 20 +-- docs/api/admin.md | 2 +- docs/guide/configuration.md | 2 +- docs/guide/deployment.md | 21 ++- docs/guide/faq.md | 2 +- docs/guide/groups.md | 2 +- docs/guide/introduction.md | 8 +- docs/guide/routes.md | 2 +- docs/guide/storage.md | 54 ++++-- docs/guide/tasks.md | 3 +- docs/zh/api/admin.md | 2 +- docs/zh/guide/configuration.md | 2 +- docs/zh/guide/deployment.md | 21 ++- docs/zh/guide/faq.md | 2 +- docs/zh/guide/groups.md | 2 +- docs/zh/guide/introduction.md | 8 +- docs/zh/guide/routes.md | 2 +- docs/zh/guide/storage.md | 54 ++++-- docs/zh/guide/tasks.md | 3 +- migrations/0006_config_d1.sql | 28 +++ migrations/0007_send_logs_index.sql | 1 + migrations/0008_storage_d1.sql | 25 +++ nuxt.config.ts | 2 +- server/lib/config.ts | 39 ++++- server/lib/core/dispatch.ts | 21 ++- server/lib/lib/idempotency.ts | 40 ++++- server/lib/lib/message-tracker.ts | 35 ++++ server/lib/lib/send-log-batch.ts | 41 +++++ server/lib/queue/delivery.ts | 55 +++++- server/lib/storage/config-store.ts | 242 +++++++++++++++++++++++++ server/lib/storage/d1.ts | 14 ++ server/lib/storage/payload.ts | 64 +++++++ server/lib/types.ts | 1 + server/lib/web/auth.ts | 2 + server/lib/web/groups.ts | 10 ++ server/lib/web/invites.ts | 4 +- server/lib/web/oauth.ts | 5 + server/lib/webhook.ts | 7 +- server/tasks/storage-prune.ts | 40 +++++ tests/config-store.test.ts | 174 ++++++++++++++++++ tests/d1-stores.test.ts | 262 ++++++++++++++++++++++++++++ tests/payload.test.ts | 68 ++++++++ tests/send-log-batch.test.ts | 112 ++++++++++++ wrangler.jsonc | 6 + 46 files changed, 1450 insertions(+), 117 deletions(-) create mode 100644 migrations/0006_config_d1.sql create mode 100644 migrations/0007_send_logs_index.sql create mode 100644 migrations/0008_storage_d1.sql create mode 100644 server/lib/lib/send-log-batch.ts create mode 100644 server/lib/storage/config-store.ts create mode 100644 server/lib/storage/d1.ts create mode 100644 server/lib/storage/payload.ts create mode 100644 server/tasks/storage-prune.ts create mode 100644 tests/config-store.test.ts create mode 100644 tests/d1-stores.test.ts create mode 100644 tests/payload.test.ts create mode 100644 tests/send-log-batch.test.ts diff --git a/AGENTS.md b/AGENTS.md index 42f20ef..ab212a4 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -12,7 +12,7 @@ Core pipeline: GitHub Webhook → Worker (verify + filter + format) → Discord - UI: Vue 3 + Tailwind CSS v3 (`@nuxtjs/tailwindcss`); admin console is a client-side SPA (`routeRules: "/admin/**": { ssr: false }`), home/legal pages render server-side - Styling: all theme colors are RGB-triplet CSS variables in `app/assets/css/main.css` mapped into `tailwind.config.ts` (so `bg-accent/10` opacity modifiers work); the design tokens switch with `prefers-color-scheme` (unless ``); repeated control patterns are `@apply` component classes in the CSS `@layer components` - Discord interactions: HTTPS Interactions Endpoint (`POST /discord/interactions`, Ed25519-signed) — no Discord Gateway / Durable Object; bot stays offline, messages always sent via REST -- Storage: Cloudflare KV (tokens, OAuth state, route config `config:routes`, group config `config:groups`, admin sessions, delivery dedup, message-update tracking `msg:*`, i18n overrides `i18n:*`) + D1 (`send_logs`, `discord_links`, `telegram_links`) +- Storage: D1 is the source of truth for config (routes/groups via `d1_routes`/`d1_groups`, see `server/lib/storage/config-store.ts`), send/audit logs, dedup (`dedup_keys`), delivery state (`delivery_state`) and message tracking (`message_tracking` via `server/lib/storage/d1.ts` `canUseD1` gate). KV keeps only cache + short-lived/ephemeral state (tokens, OAuth state, admin sessions, `msg:*`-adjacent locks, per-group secrets `tenant:*`, invites, i18n overrides) with explicit TTLs. R2 parks oversized queue payloads (`webhooks/YYYY/MM/DD/*.json`, `server/lib/storage/payload.ts`) instead of KV. A `storage-prune` scheduled task cleans up expired dedup/delivery/message-tracking rows - Signature verification: Web Crypto API (HMAC-SHA256 for GitHub/Gitea, Ed25519 for Discord, timing-safe secret-token compare for Telegram) - Webhook providers: pluggable forge adapters under `server/lib/providers/` (github, gitea) — each verifies its own signature format and normalizes its payload to a GitHub-shaped `WebhookEvent`; a `custom` provider accepts arbitrary signed JSON posts (`X-WebHooker-Signature`) as `custom` events; GitLab etc. can be added later - Per-group webhook ingress: optional `POST /webhook/{groupId}` with a per-group secret in KV (`tenant:{groupId}`) — Gitea/classic-GitHub/custom webhooks are verified against the group's secret instead of the operator's global ones; only that group's routes fire. The legacy `POST /webhook` (global secrets, all routes) stays untouched @@ -24,7 +24,8 @@ 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) +- 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 R2 (`webhooks/YYYY/MM/DD/*.json`, falling back to KV `queue:payload:*` without the R2 binding) and resolved by the consumer; delivery state is tracked as the D1 `delivery_state` table (KV `delivery-state:*` fallback). Without the `QUEUE` binding, dispatch stays inline (existing behavior) +- Storage quotas: the design avoids KV write pressure (Workers Free KV allows ~1,000 writes/day) by keeping high-frequency ephemeral writes in D1 instead — webhook dedup (`dedup_keys`, atomic `ON CONFLICT` UPSERT), delivery state and message tracking all use D1 rows via `canUseD1` (prepare+batch probe) with automatic KV fallback when D1 is unavailable or unmigrated. D1 Free allows 100,000 rows written/day, so the per-event KV write cost drops to ~0; config stays cached in memory + KV with D1 authoritative - Local dev: wrangler + Miniflare ## Architecture @@ -43,22 +44,22 @@ app/ # Vue 3 UI (Nuxt app dir) 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 +├── tasks/ # scheduled (cron */5): discord-sync, telegram-sync, audit-prune, storage-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 + ├── config.ts # loadRoutes/saveRoutes (delegates to D1 ConfigStore when available, else KV config:routes), loadConfig from env ├── config/ # config schema + migration + validation │ └── schema.ts # CONFIG_SCHEMA_VERSION, valibot route/group/filter schemas, migrateRoutes/Groups, validateRoutes/Groups (non-destructive), explainRoute ├── 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, enqueue (or inline dispatch) + ├── webhook.ts # processWebhook/handleWebhook: tenant lookup, provider detect/verify/parse, dedup (D1/KV), enqueue (or inline dispatch) ├── queue/ # Cloudflare Queue delivery pipeline - │ ├── delivery.ts # DeliveryMessage, enqueueWebhook, retry backoff, delivery-state KV, payload-overflow parking + │ ├── delivery.ts # DeliveryMessage, enqueueWebhook, retry backoff, delivery-state (D1 w/ KV fallback), payload-overflow parking (R2 → KV) │ └── 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) + │ └── dispatch.ts # Platform-neutral dispatch: match routes → formatEvent → driver.send/edit (batched recordSend + group filter + per-group webhook log) ├── events/ │ ├── match.ts # matchRoute, eventOwners; unified pattern syntax (*/? globs + //-wrapped regex) │ └── filter-ast.ts # FilterNode evaluator (all/any/not), containsKeyword, explainFilter/explainFilterNode; pattern helpers (regex/glob compile) @@ -112,12 +113,17 @@ server/ # Nitro server │ └── richheader.ts # GET /api/richheader: Open Graph page for Telegram avatar link-preview card ├── observability/ # delivery metrics aggregation │ └── metrics.ts # DeliveryMetrics + getDeliveryMetrics (SQL GROUP BY over send_logs: totals, per platform/event/status, duration, attempts, recent failures) + ├── storage/ # storage primitives: canUseD1 probe, D1 config store (L1 memory → L2 KV cache → L3 D1), R2 payload store + │ ├── d1.ts # canUseD1(db) — prepare+batch probe for D1 availability (falls back to KV) + │ ├── config-store.ts # ConfigStore (loadRoutes/saveRoutes/loadGroups/saveGroups/invalidateCache) backed by d1_groups/d1_routes, seeds from KV + syncs cache + │ └── payload.ts # PayloadStore backed by R2 (webhooks/YYYY/MM/DD/*.json), no-binding stub throws └── lib/ # shared infra ├── i18n.ts # loadTranslations (KV i18n:{lang} overrides), t() with param interpolation - ├── idempotency.ts # IdempotencyStore interface + kvIdempotencyStore (delivery dedup via claim/has) + deliveryKey + ├── idempotency.ts # IdempotencyStore interface + kvIdempotencyStore/d1IdempotencyStore (delivery dedup via claim/has, D1 atomic UPSERT w/ KV fallback) + deliveryKey ├── correlation.ts # newCorrelationId() — per-request/delivery correlation id for logs + responses - ├── message-tracker.ts # MessageTracker interface + kvMessageTracker (KV msg:{eventId}:{targetId} for workflow_run/check_run edits) + ├── message-tracker.ts # MessageTracker interface + kvMessageTracker/d1MessageTracker (msg:{eventId}:{targetId} for workflow_run/check_run edits) ├── send-log.ts # SendRecord, recordSend/getSendLog/getSendLogById/getSendLogByDelivery/getFailedSendLog (D1 send_logs) + ├── send-log-batch.ts # recordSendBatch (D1 batch INSERT for send_logs) ├── audit.ts # recordAudit/getAuditLog/pruneAuditLogs (D1 audit_logs, best-effort writes) ├── log.ts # JSON console logger (info/warn/error/fatal) └── locales/ # en.ts, zh.ts translation dictionaries @@ -147,14 +153,15 @@ tests/__snapshots__/ # formatter snapshot golden files (toMatchSnapshot) - Format 28 GitHub/Gitea event types plus `custom` webhooks as platform-neutral messages (Discord embeds + Telegram HTML) - Show the forge source (named per `Group.forgeSources` host entries) in the message footer when the group defines a host matching the event's repository - Route messages to Discord channels/threads and Telegram chats/topics via REST -- Edit already-sent messages in place for `workflow_run` / `check_run` progress (stable `updateKey`, KV `msg:*` tracking) +- Edit already-sent messages in place for `workflow_run` / `check_run` progress (stable `updateKey`, `message_tracking` via D1 with KV `msg:*` fallback) - Record every dispatch attempt to D1 `send_logs` (route id, event, target, ok/error, duration, error code) - Aggregate delivery metrics (`server/lib/observability/metrics.ts`) from `send_logs` — totals, ok/failed counts + failure rate, per-platform/per-event/per-status breakdowns, average duration and attempts, recent failures; `getDeliveryMetrics(db, groupId?)` scopes every query by `group_id` when a group is passed - Expose admin observability endpoints — `GET /admin/api/metrics?groupId=` (delivery metrics, optional group scope; recent failures group-scoped for non-super) and `GET /admin/api/delivery/:deliveryId` (all send-log attempts for one delivery, group-scoped) — through the `/admin/api/[...slug]` catch-all route (`server/routes/admin/api/[...slug].ts`) that wires every admin API handler to its method+path -- 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`) +- 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 `idempotencyStore`, D1 `dedup_keys` with KV fallback) - 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 +- Track delivery state in the D1 `delivery_state` table (`delivery-state:*` KV fallback: pending/processing/delivered/retrying/failed/dead) so redelivered messages are skipped idempotently; oversized payloads are parked in R2 (`webhooks/YYYY/MM/DD/*.json`, KV `queue:payload:*` fallback) and deleted after dispatch +- Prune expired D1 rows via the scheduled `storage-prune` task (dedup keys past expiry, delivery state older than 7 days, message tracking older than 30 days; audits via `audit-prune`) - 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 @@ -227,10 +234,11 @@ Rule: no functional change ships without its documentation; docs and code must n - **Local dev**: `.dev.vars` (wrangler reads this for env bindings) - **Production**: `wrangler secret put ` for each secret -- **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 +- **Routes**: config in D1 `d1_routes`/`d1_groups` (authoritative), seeded from legacy KV `config:routes`/`config:groups` on first load; cache-only KV keys +- **KV namespace**: Required binding for token/state/session/cache storage (dedup/delivery-state/message-tracking fall back to KV when D1 is unavailable) +- **D1 database**: Binding `DB` (database `webhooker`, id `214a0104-3235-47c0-b7bf-ddda95f3c8ac`) for `send_logs` + `audit_logs` + `discord_links` + `telegram_links` + `d1_groups` + `d1_routes` + `dedup_keys` + `delivery_state` + `message_tracking` 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 +- **R2**: optional `PAYLOAD` binding (bucket `webhooker-payloads`) for oversized queue payloads; without it, oversized payloads fall back to KV `queue:payload:*` - **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`) @@ -249,7 +257,8 @@ bunx wrangler d1 create webhooker 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) +bun run db:migrate:prod # wrangler d1 migrations apply webhooker --remote (migrations/0001..0008) +# R2 bucket for oversized payloads (P0): bunx wrangler r2 bucket create webhooker-payloads bunx wrangler deploy ``` diff --git a/README.md b/README.md index 6f88f67..726cb54 100644 --- a/README.md +++ b/README.md @@ -18,8 +18,8 @@ GitHub / Gitea webhook → Discord / Telegram dispatcher. Receives webhook event - **Web UI config console** (`/admin`) — manage routes and groups with GitHub OAuth + admin whitelist, view send logs - **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 +- Cloudflare D1 for config (routes/groups), send logs, platform account links, dedup, delivery state and message tracking + KV for ephemeral state/cache/security tokens, + optional R2 for oversized payloads +- **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 R2 (`PAYLOAD` binding, falling back to KV `queue:payload:*`). Without the binding, dispatch stays inline - Graceful degradation (webhook-only mode if Discord unavailable) ## Architecture @@ -38,9 +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:{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:*`) +- **KV** — cache + ephemeral state: token storage (`token:{userId}`), OAuth state (`state:{hex}`), admin sessions (`session:{id}`), per-group webhook secrets (`tenant:{groupId}`), invites, config cache, delivery dedup/state/message-tracking **fallback** (`delivery:*`, `delivery-state:*`, `msg:*` used only when D1 is unavailable) and message-update locks (`msg:lock:*`) +- **D1** — source of truth for routes/groups (`d1_routes`/`d1_groups`), send logs (`send_logs`), audit logs (`audit_logs`), dedup (`dedup_keys`), delivery state (`delivery_state`), message-update tracking (`message_tracking`), 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 R2 (`PAYLOAD` binding, `webhooks/YYYY/MM/DD/*.json`, falling back to KV `queue:payload:*`) ## Quick Start @@ -79,7 +79,7 @@ bunx wrangler dev # Start local dev server ### Routes -Routes are stored in KV (`config:routes` as JSON). There are **no default routes** — every route (including its target) must be defined explicitly, either via the Web UI (`/admin`) or by storing a JSON array in KV. A route may carry multiple `targets`, so one rule can forward to several channels at once: +Routes are stored in D1 (`d1_routes`, seeded from legacy KV `config:routes` on first load). There are **no default routes** — every route (including its target) must be defined explicitly, either via the Web UI (`/admin`) or by storing a JSON array in D1. A route may carry multiple `targets`, so one rule can forward to several channels at once: ```json [ @@ -98,7 +98,7 @@ Routes are stored in KV (`config:routes` as JSON). There are **no default routes ] ``` -`target.platform` selects the push target: `discord` (default) or `telegram`. Discord targets require `target.channelId` (optional `threadId` for a thread); Telegram targets require `target.chatId` (optional `topicId` for a topic). Routes belong to **groups** (KV `config:groups`) that scope admin access and can restrict which org/user events flow in. See the [Routes & Targets](https://webhooker.docs.worldexecute.me/guide/routes) and [Groups & Access Control](https://webhooker.docs.worldexecute.me/guide/groups) guides for the full schema. +`target.platform` selects the push target: `discord` (default) or `telegram`. Discord targets require `target.channelId` (optional `threadId` for a thread); Telegram targets require `target.chatId` (optional `topicId` for a topic). Routes belong to **groups** (D1 `d1_groups`, seeded from legacy KV `config:groups`) that scope admin access and can restrict which org/user events flow in. See the [Routes & Targets](https://webhooker.docs.worldexecute.me/guide/routes) and [Groups & Access Control](https://webhooker.docs.worldexecute.me/guide/groups) guides for the full schema. ### Web UI (`/admin`) @@ -106,7 +106,7 @@ The built-in config console lets you manage routes and groups in the browser (ad 1. Set `ADMIN_USER_IDS` to the GitHub user IDs (or logins) allowed to manage the console, e.g. `ADMIN_USER_IDS=12345,RhenCloud`. 2. Visit `/admin` and sign in with GitHub. Users with no access get `403` — unless `ALLOW_SELF_SIGNUP=1` (they receive a personal group) or they follow a group invite link. -3. Changes are written to KV immediately and picked up by the webhook pipeline. +3. Changes are written to D1 immediately, the config cache is invalidated, and the webhook pipeline picks them up on the next run. Sign out at `/admin/logout`. Every group has `members` with a role (`owner` / `admin` / `viewer`); all admin operations are recorded in the D1 `audit_logs` table. diff --git a/README.zh.md b/README.zh.md index 59e3d13..df1c415 100644 --- a/README.zh.md +++ b/README.zh.md @@ -18,8 +18,8 @@ GitHub / Gitea webhook → Discord / Telegram 分发服务。通过 Cloudflare W - **Web 配置控制台**（`/admin`）— 通过 GitHub OAuth + 管理员白名单管理路由与分组、查看发送日志 - **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。未绑定则保持同步分发 +- Cloudflare D1 存储配置（路由/分组）、发送日志、平台账号绑定、去重、投递状态与消息更新追踪 + KV 存储临时状态/缓存/安全令牌 + 可选 R2 存储超大负载 +- **Cloudflare Queues 异步投递** —— 绑定 `QUEUE` 时，已验签的 webhook 会入队到 `webhooker-delivery`，由消费者分发，带指数退避重试（5s/30s/2m/10m）与死信队列（`webhooker-delivery-dlq`）；超大负载暂存于 R2（`PAYLOAD` 绑定，回退 KV `queue:payload:*`）。未绑定则保持同步分发 - 优雅降级（Discord 不可用时仅 webhook 模式） ## 架构 @@ -38,9 +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:{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:*`） +- **KV** — 缓存 + 临时状态：Token 存储（`token:{userId}`）、OAuth state（`state:{hex}`）、管理员会话（`session:{id}`）、分组级 webhook secret（`tenant:{groupId}`）、邀请、配置缓存、投递去重/投递状态/消息更新追踪的回退（`delivery:*`、`delivery-state:*`、`msg:*` 仅在 D1 不可用时使用）与消息更新锁（`msg:lock:*`） +- **D1** — 路由/分组（`d1_routes`/`d1_groups`）、发送日志（`send_logs`）、审计日志（`audit_logs`）、去重（`dedup_keys`）、投递状态（`delivery_state`）、消息更新追踪（`message_tracking`）、Discord↔GitHub 绑定（`discord_links`）、Telegram↔GitHub 绑定（`telegram_links`） +- **Queue** — 绑定 `QUEUE` 时异步投递：`webhooker-delivery`（指数退避重试）+ 死信队列 `webhooker-delivery-dlq`；超大负载暂存于 R2（`PAYLOAD` 绑定，`webhooks/YYYY/MM/DD/*.json`，回退 KV `queue:payload:*`） ## 快速开始 @@ -79,7 +79,7 @@ bunx wrangler dev # 启动本地开发服务器 ### 路由配置 -路由存储在 KV（`config:routes`，JSON 格式）。**没有默认路由**——每条路由（包括目标）都必须显式定义，可通过 Web 控制台（`/admin`）或直接向 KV 存储 JSON 数组。一条路由可携带多个 `targets`，因此一个规则可以同时转发到多个频道： +路由存储在 D1（`d1_routes`，首次加载时从旧版 KV `config:routes` 同步）。**没有默认路由**——每条路由（包括目标）都必须显式定义，可通过 Web 控制台（`/admin`）或直接向 D1 存储 JSON 数组。一条路由可携带多个 `targets`，因此一个规则可以同时转发到多个频道： ```json [ @@ -98,15 +98,15 @@ bunx wrangler dev # 启动本地开发服务器 ] ``` -`target.platform` 选择推送目标：`discord`（默认）或 `telegram`。Discord 目标需 `target.channelId`（可选 `threadId` 指向子区）；Telegram 目标需 `target.chatId`（可选 `topicId` 指向话题）。路由隶属于**分组**（KV `config:groups`），分组用于限定管理权限，并可限制哪些组织/用户的事件流入。完整模式见[路由与目标](https://webhooker.docs.worldexecute.me/zh/guide/routes)与[分组与访问控制](https://webhooker.docs.worldexecute.me/zh/guide/groups)指南。 +`target.platform` 选择推送目标：`discord`（默认）或 `telegram`。Discord 目标需 `target.channelId`（可选 `threadId` 指向子区）；Telegram 目标需 `target.chatId`（可选 `topicId` 指向话题）。路由隶属于**分组**（D1 `d1_groups`，首次加载时从旧版 KV `config:groups` 同步），分组用于限定管理权限，并可限制哪些组织/用户的事件流入。完整模式见[路由与目标](https://webhooker.docs.worldexecute.me/zh/guide/routes)与[分组与访问控制](https://webhooker.docs.worldexecute.me/zh/guide/groups)指南。 ### Web 控制台（`/admin`） -内置的配置控制台让你在浏览器中管理路由与分组（新增 / 编辑 / 删除 / 开关 / 排序）、查看发送日志、管理组成员与邀请链接、阅读审计日志——无需操作 KV： +内置的配置控制台让你在浏览器中管理路由与分组（新增 / 编辑 / 删除 / 开关 / 排序）、查看发送日志、管理组成员与邀请链接、阅读审计日志——无需直接操作 D1 或 KV： 1. 设置 `ADMIN_USER_IDS` 为允许管理控制台的 GitHub 用户 ID（或登录名），例如 `ADMIN_USER_IDS=12345,RhenCloud`。 2. 访问 `/admin` 并用 GitHub 登录。无任何权限的用户收到 `403`——除非开启 `ALLOW_SELF_SIGNUP=1`（自动获得个人分组）或通过分组邀请链接加入。 -3. 修改会立即写入 KV，webhook 管线随即生效。 +3. 修改会立即写入 D1，配置缓存随之失效，webhook 管线随即生效。 在 `/admin/logout` 退出登录。每个分组都有带角色的 `members`（`owner` / `admin` / `viewer`）；所有管理操作（登录、分组/路由/成员/邀请变更）都会写入 D1 `audit_logs` 表。 @@ -160,7 +160,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 数据库与迁移、可选 Queues、密钥、部署：见[部署指南](https://webhooker.docs.worldexecute.me/zh/guide/deployment) +- **部署** — KV 命名空间、D1 数据库与迁移（含 0008 存储表）、可选 R2 Bucket 与 Queues、密钥、部署：见[部署指南](https://webhooker.docs.worldexecute.me/zh/guide/deployment) ### Bot 指令（以本人身份评论 GitHub） diff --git a/docs/api/admin.md b/docs/api/admin.md index 73e13e3..ee271cd 100644 --- a/docs/api/admin.md +++ b/docs/api/admin.md @@ -34,7 +34,7 @@ The console itself is served at `/admin`; its tabs are deep-linkable via the URL ## Validation -- `PUT /admin/api/routes` — Body `{ "routes": Route[] }`; validates each route (id pattern, unique id within its group, name, enabled, `groupId`, filters — empty only allowed for `fallback` routes — optional `discordRoleIds` (list of role id strings), and platform-aware targets: `target.channelId` for Discord, `target.chatId` for Telegram) and persists to KV `config:routes`. Returns `200 { ok, count }` or `400 { error }` / `401 { error }` / `403 { error }`. Unchanged routes skip the full validation. +- `PUT /admin/api/routes` — Body `{ "routes": Route[] }`; validates each route (id pattern, unique id within its group, name, enabled, `groupId`, filters — empty only allowed for `fallback` routes — optional `discordRoleIds` (list of role id strings), and platform-aware targets: `target.channelId` for Discord, `target.chatId` for Telegram) and persists to D1 `d1_routes`. Returns `200 { ok, count }` or `400 { error }` / `401 { error }` / `403 { error }`. Unchanged routes skip the full validation. - `PUT /admin/api/groups` — Validates group ids, member roles (at least one `owner`), `providers` (`github` / `gitea`), and `installationId`. - Limits: at most 200 routes and 100 groups per instance. diff --git a/docs/guide/configuration.md b/docs/guide/configuration.md index 1179259..65f65b4 100644 --- a/docs/guide/configuration.md +++ b/docs/guide/configuration.md @@ -60,7 +60,7 @@ WebHooker ships with a built-in config console at `/admin` for managing routes, The console is served as an SPA at `/admin`; its tabs are deep-linkable via the URL path (`/admin/groups`, `/admin/logs`, `/admin/audit`). URLs outside `/admin` that do not match an endpoint return a plain `404` instead of the console. -All management endpoints (`/admin/api/*`) are documented in the [Admin API](../api/admin). Saved routes are written to KV `config:routes` immediately and the config cache is invalidated so the webhook pipeline picks them up on the next run. +All management endpoints (`/admin/api/*`) are documented in the [Admin API](../api/admin). Saved routes and groups are persisted to D1 (`d1_routes` / `d1_groups`) immediately, the KV cache is invalidated and the config cache is refreshed so the webhook pipeline picks them up on the next run. ## Filter Types diff --git a/docs/guide/deployment.md b/docs/guide/deployment.md index 650f801..8e92a92 100644 --- a/docs/guide/deployment.md +++ b/docs/guide/deployment.md @@ -81,11 +81,24 @@ bunx wrangler d1 execute webhooker --remote --file ./migrations/0002_log_detail. bunx wrangler d1 execute webhooker --remote --file ./migrations/0003_telegram_links.sql bunx wrangler d1 execute webhooker --remote --file ./migrations/0004_add_group_id.sql bunx wrangler d1 execute webhooker --remote --file ./migrations/0005_audit_logs.sql +bunx wrangler d1 execute webhooker --remote --file ./migrations/0006_config_d1.sql +bunx wrangler d1 execute webhooker --remote --file ./migrations/0007_send_logs_index.sql +bunx wrangler d1 execute webhooker --remote --file ./migrations/0008_storage_d1.sql ``` ::: -### 4. Create Queues (Optional) +### 4. Create R2 Bucket (Optional) + +The `PAYLOAD` binding parks oversized webhook payloads in R2 (`webhooker-payloads`) instead of KV. Without it, oversized payloads fall back to the KV key `queue:payload:*`. See [Storage](/guide/storage) for the layout. + +```bash +bunx wrangler r2 bucket create webhooker-payloads +``` + +The bucket is already declared in `wrangler.jsonc` (`r2_buckets`), so no binding change is needed. + +### 5. 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). @@ -96,7 +109,7 @@ 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 +### 6. Deploy ```bash bunx wrangler deploy @@ -104,13 +117,13 @@ bunx wrangler deploy Your worker is now live at `https://webhooker..workers.dev`. -### 6. Configure GitHub Webhook +### 7. 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` -### 7. (Optional) Configure Gitea Webhook +### 8. (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` diff --git a/docs/guide/faq.md b/docs/guide/faq.md index c174842..7f4455c 100644 --- a/docs/guide/faq.md +++ b/docs/guide/faq.md @@ -34,4 +34,4 @@ No — the worker requires the KV and D1 bindings declared in `wrangler.jsonc` a ## Where is data stored? -Configuration lives in Cloudflare KV (`config:routes`, `config:groups`); send/audit logs and platform↔GitHub links live in D1. See [Storage Layout](./storage#kv-storage-layout). +Configuration lives in D1 (`d1_routes`/`d1_groups`, with KV as a cache); webhook dedup, delivery state and message-update tracking also live in D1 with KV fallback; send/audit logs and platform↔GitHub links live in D1. R2 optionally parks oversized payloads. See [Storage Layout](./storage#storage-decisions). diff --git a/docs/guide/groups.md b/docs/guide/groups.md index d192571..1ed445b 100644 --- a/docs/guide/groups.md +++ b/docs/guide/groups.md @@ -1,6 +1,6 @@ # Groups & Access Control -Routes belong to groups. Groups scope admin access and can restrict which events flow into them. They are stored in Cloudflare KV under the key `config:groups` as a JSON array, managed via the [Web UI](./configuration#web-ui) or the [Admin API](../api/admin). At most **100 groups** can be saved per instance. +Routes belong to groups. Groups scope admin access and can restrict which events flow into them. They are stored in D1 (`d1_groups`, seeded from the legacy KV `config:groups` key on first load), managed via the [Web UI](./configuration#web-ui) or the [Admin API](../api/admin). At most **100 groups** can be saved per instance. ## Group Schema diff --git a/docs/guide/introduction.md b/docs/guide/introduction.md index 29fa31e..d22d8e0 100644 --- a/docs/guide/introduction.md +++ b/docs/guide/introduction.md @@ -22,14 +22,14 @@ GitHub / Gitea Webhook → Cloudflare Worker (Nuxt 4 / Nitro) | ------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | | **Cloudflare Worker** | HTTP ingress, signature verification, delivery dedup, event parsing, route matching, platform dispatch | | **Interactions Endpoint** | Verifies Ed25519 signatures and handles `/gh` interactions (slash commands, context-menu commands, buttons, modals) | -| **KV** | Token storage (`token:{userId}`), OAuth state (`state:{hex}`), route config (`config:routes`), group config (`config:groups`), admin sessions, delivery dedup, message-update tracking (`msg:*`) | -| **D1** | Send logs (`send_logs`), Discord↔GitHub links (`discord_links`), Telegram↔GitHub links (`telegram_links`) | +| **KV** | Token storage (`token:{userId}`), OAuth state (`state:{hex}`), admin sessions, per-group secrets, config cache, delivery dedup/state/message-tracking fallback when D1 is unavailable (`delivery:*`, `delivery-state:*`, `msg:*`), message-update locks (`msg:lock:*`) | +| **D1** | Routes/groups (`d1_routes`/`d1_groups`), send logs (`send_logs`), audit logs (`audit_logs`), dedup (`dedup_keys`), delivery state (`delivery_state`), message tracking (`message_tracking`), Discord↔GitHub links (`discord_links`), Telegram↔GitHub links (`telegram_links`), with R2 optionally parking oversized payloads | ### Data Flow 1. A forge (GitHub or Gitea) sends a webhook to `POST /webhook` 2. Worker detects the provider from its headers (`X-GitHub-Event` / `X-Gitea-Event`) and verifies the provider-specific HMAC-SHA256 signature -3. Worker deduplicates by the delivery id (KV, short TTL) to drop repeat deliveries +3. Worker deduplicates by the delivery id (D1, short TTL, KV fallback) to drop repeat deliveries 4. Worker parses the event type and normalizes the payload to a GitHub-shaped event 5. Routes are evaluated against filters (event, repo, actor, action, branch, keyword) and group owner restrictions 6. Matching routes trigger formatter functions that produce platform-neutral messages @@ -42,7 +42,7 @@ GitHub / Gitea Webhook → Cloudflare Worker (Nuxt 4 / Nitro) - **Discord delivery**: Discord REST API (interactions via an Ed25519-verified HTTPS Interactions Endpoint) - **Telegram delivery**: Telegram Bot API (webhook with optional secret-token verification) - **Web UI**: Nuxt 4 (Vue 3 + Tailwind CSS v3) — server-rendered home/legal pages, client-side `/admin` console -- **Storage**: Cloudflare KV + D1 +- **Storage**: Cloudflare D1 (authoritative for config & logs) + KV (cache/transient state) + optional R2 (oversized payloads) - **Auth**: Web Crypto API (HMAC-SHA256, Ed25519), octokit (GitHub API) - **Language**: TypeScript diff --git a/docs/guide/routes.md b/docs/guide/routes.md index fef5d6c..055156c 100644 --- a/docs/guide/routes.md +++ b/docs/guide/routes.md @@ -1,6 +1,6 @@ # Routes & Targets -Routes define which events get forwarded to which channel (Discord or Telegram). They are stored in Cloudflare KV under the key `config:routes` as a JSON array, managed via the [Web UI](./configuration#web-ui), the [Admin API](../api/admin), or `config.example.yaml`. +Routes define which events get forwarded to which channel (Discord or Telegram). They are stored in D1 (`d1_routes`, seeded from the legacy KV `config:routes` key on first load), managed via the [Web UI](./configuration#web-ui), the [Admin API](../api/admin), or `config.example.yaml`. There are **no default routes** — each route must define its own target. If no routes are configured, no events are forwarded. At most **200 routes** can be saved per instance. diff --git a/docs/guide/storage.md b/docs/guide/storage.md index 5f51138..6c89cb9 100644 --- a/docs/guide/storage.md +++ b/docs/guide/storage.md @@ -2,22 +2,24 @@ ## KV Storage Layout +KV keeps only cache data and short-lived/ephemeral state. High-frequency writes (webhook dedup, delivery state, message tracking) live in D1 and only fall back to KV when D1 is unavailable or not yet migrated (see [Storage decisions](./storage#storage-decisions)). + | Key Pattern | Value | TTL | | ------------------------------------------ | ------------------------------------------------------------------------------------ | ------------------ | -| `config:routes` | JSON array of routes | Permanent | -| `config:groups` | JSON array of groups | Permanent | +| `config:routes` | Route config cache (D1 `d1_routes` is authoritative) | 1 hour | +| `config:groups` | Group config cache (D1 `d1_groups` is authoritative) | 1 hour | | `session:{id}` | Admin session `{ userId, login }` | 7 days | | `token:{userId}` | `{ userId, accessToken, expiresAt, refreshToken? }` | 0.9 × token expiry | | `token-reverse:{sha256}` | User id for reverse lookup by token | 0.9 × token expiry | | `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:{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 | +| `invite:group:{id}` | Token index per group (keeps invite listing consistent) | 7 days | +| `delivery:{provider}:{groupId}:{id}` | Webhook delivery dedup fallback (D1 `dedup_keys` is primary) | 7 days | +| `delivery-state:{provider}:{groupId}:{id}` | Queue delivery state fallback (D1 `delivery_state` is primary) | 1 hour | +| `queue:payload:{provider}:{groupId}:{id}` | Oversized webhook payload parked for the queue consumer (R2 is primary) | 1 hour | | `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 | +| `msg:{routeId}:{key}:{target}` | Message id tracking fallback (D1 `message_tracking` is primary) | 1 day | | `cmd:guild:{id}` | Guild id whose commands were registered (dedup) | Permanent | | `cmd:registered:global` | Global command registration marker (dedup) | 1 day | | `config:discord-app-id` | Cached Discord application id | Permanent | @@ -25,13 +27,35 @@ ## D1 Storage Layout -The D1 database (`DB` binding, database `webhooker`) holds four tables: +The D1 database (`DB` binding, database `webhooker`) holds the source of truth for configuration, delivery logs and high-frequency ephemeral state: -| Table | Purpose | -| ---------------- | ---------------------------------------------------------------------------------------------- | -| `send_logs` | One row per dispatch attempt (route id, event, target, ok/error, duration, error code, detail) | -| `audit_logs` | One row per admin operation (login/logout, group/route/member/invite changes) | -| `discord_links` | Maps `discord_user_id` → `github_user_id` for `/gh` Discord commands | -| `telegram_links` | Maps `telegram_user_id` → `github_user_id` for `/gh` Telegram commands | +| Table | Purpose | +| ------------------ | ---------------------------------------------------------------------------------------------------- | +| `d1_groups` | Groups (authoritative config, seeded from legacy KV `config:groups`) | +| `d1_routes` | Routes per group (authoritative config, seeded from legacy KV `config:routes`) | +| `send_logs` | One row per dispatch attempt (route id, event, target, ok/error, duration, error code, detail) | +| `audit_logs` | One row per admin operation (login/logout, group/route/member/invite changes) | +| `dedup_keys` | Webhook delivery dedup (atomic `INSERT ... ON CONFLICT` UPSERT, key + expiry) | +| `delivery_state` | Queue delivery state (`pending`/`processing`/`delivered`/`retrying`/`failed`/`dead`) | +| `message_tracking` | Message id tracking for in-place updates (`event_id` + `target_id` → `message_id`) | +| `discord_links` | Maps `discord_user_id` → `github_user_id` for `/gh` Discord commands | +| `telegram_links` | Maps `telegram_user_id` → `github_user_id` for `/gh` Telegram commands | -`audit_logs` is pruned automatically by the scheduled trigger after `AUDIT_RETENTION_DAYS` (default 90). See [Logs](./logs) for the row fields. +`audit_logs` is pruned by the scheduled trigger after `AUDIT_RETENTION_DAYS` (default 90). The `storage-prune` task removes expired `dedup_keys`, `delivery_state` rows older than 7 days and `message_tracking` rows older than 30 days. See [Logs](./logs) for the log row fields. + +## R2 Storage Layout + +R2 (`PAYLOAD` binding, bucket `webhooker-payloads`) stores oversized webhook payloads that are too large for a queue message or KV: + +| Object Pattern | Purpose | Retention | +| ----------------------------- | ---------------------------------------------- | --------- | +| `webhooks/YYYY/MM/DD/.json` | Oversized payload parked for the queue consumer | deleted after dispatch | + +When the `PAYLOAD` binding is absent, oversized payloads fall back to the KV key `queue:payload:{provider}:{groupId}:{id}` (1 hour TTL). + +## Storage Decisions + +- **D1 is authoritative** for config and delivery metadata; KV holds only caches and short-lived state. +- The `canUseD1` probe (`server/lib/storage/d1.ts`, checks for `prepare` + `batch`) gates every D1 store: when D1 is unavailable or not yet migrated, all three high-frequency stores (dedup, delivery state, message tracking) transparently fall back to KV so behavior is unchanged during migration. +- This keeps per-event KV writes near zero on the Workers Free plan (1,000 writes/day): dedup, delivery state and message tracking now write D1 rows instead (D1 Free allows 100,000 rows written/day). +- R2's free tier (10 GB-month storage, 1M Class A ops/month) comfortably absorbs payload parking without touching the KV write quota. diff --git a/docs/guide/tasks.md b/docs/guide/tasks.md index 58d5519..41cf1a9 100644 --- a/docs/guide/tasks.md +++ b/docs/guide/tasks.md @@ -1,11 +1,12 @@ # Scheduled Tasks -WebHooker runs three maintenance tasks on the scheduled trigger (`*/5 * * * *`, every 5 minutes). They only run on the deployed worker (Cloudflare cron); local `wrangler dev` runs them when triggered via `wrangler dev --test-scheduled`. +WebHooker runs four maintenance tasks on the scheduled trigger (`*/5 * * * *`, every 5 minutes). They only run on the deployed worker (Cloudflare cron); local `wrangler dev` runs them when triggered via `wrangler dev --test-scheduled`. | Task | Purpose | | --------------- | ---------------------------------------------------------------------------------------------------------------- | | `discord-sync` | Registers the Discord slash/context-menu commands: per-guild (instant) and globally (24h dedup, ~1h propagation) | | `telegram-sync` | Calls `setWebhook` to `{BASE_URL}/telegram/webhook` (with `TELEGRAM_WEBHOOK_SECRET` as `secret_token` when set) | | `audit-prune` | Deletes `audit_logs` entries older than `AUDIT_RETENTION_DAYS` (default 90) | +| `storage-prune` | Deletes expired `dedup_keys` rows, `delivery_state` rows older than 7 days and `message_tracking` rows older than 30 days | There is nothing to configure beyond the secrets the tasks use (`DISCORD_TOKEN`, `DISCORD_APPLICATION_ID`, `TELEGRAM_TOKEN`, `BASE_URL`, `AUDIT_RETENTION_DAYS`). diff --git a/docs/zh/api/admin.md b/docs/zh/api/admin.md index a559d7c..b61d1dd 100644 --- a/docs/zh/api/admin.md +++ b/docs/zh/api/admin.md @@ -34,7 +34,7 @@ ## 校验 -- `PUT /admin/api/routes` — 请求体为 `{ "routes": Route[] }`；校验每条路由（id 格式、组内唯一 id、name、enabled、groupId、过滤器——**仅 `fallback` 路由允许空过滤器**——可选的 `discordRoleIds`（身份组 id 字符串列表）、平台感知的 targets：Discord 需 `target.channelId`，Telegram 需 `target.chatId`）并持久化到 KV `config:routes`。返回 `200 { ok, count }` 或 `400 { error }` / `401 { error }` / `403 { error }`。未变更的路由跳过完整校验。 +- `PUT /admin/api/routes` — 请求体为 `{ "routes": Route[] }`；校验每条路由（id 格式、组内唯一 id、name、enabled、groupId、过滤器——**仅 `fallback` 路由允许空过滤器**——可选的 `discordRoleIds`（身份组 id 字符串列表）、平台感知的 targets：Discord 需 `target.channelId`，Telegram 需 `target.chatId`）并持久化到 D1 `d1_routes`。返回 `200 { ok, count }` 或 `400 { error }` / `401 { error }` / `403 { error }`。未变更的路由跳过完整校验。 - `PUT /admin/api/groups` — 校验分组 id、成员角色（至少一个 `owner`）、`providers`（`github` / `gitea`）与 `installationId`。 - 上限：每个实例最多 200 条路由与 100 个分组。 diff --git a/docs/zh/guide/configuration.md b/docs/zh/guide/configuration.md index 96aed7a..e623be9 100644 --- a/docs/zh/guide/configuration.md +++ b/docs/zh/guide/configuration.md @@ -59,7 +59,7 @@ WebHooker 在 `/admin` 提供内置配置控制台，可在浏览器中管理路 控制台以 SPA 形式在 `/admin` 提供；其标签页可通过 URL 路径直达（`/admin/groups`、`/admin/logs`、`/admin/audit`）。`/admin` 之外未匹配到端点的 URL 直接返回 `404`，而不会展示控制台。 -所有管理端点（`/admin/api/*`）见 [Admin API](../api/admin)。保存的路由会立即写入 KV `config:routes` 并使配置缓存失效，下一次 webhook 处理即会生效。 +所有管理端点（`/admin/api/*`）见 [Admin API](../api/admin)。保存的路由和分组会立即持久化到 D1（`d1_routes` / `d1_groups`），并使 KV 缓存失效、刷新配置缓存，下一次 webhook 处理即会生效。 ## 过滤器类型 diff --git a/docs/zh/guide/deployment.md b/docs/zh/guide/deployment.md index 1902a14..5a70532 100644 --- a/docs/zh/guide/deployment.md +++ b/docs/zh/guide/deployment.md @@ -81,11 +81,24 @@ bunx wrangler d1 execute webhooker --remote --file ./migrations/0002_log_detail. bunx wrangler d1 execute webhooker --remote --file ./migrations/0003_telegram_links.sql bunx wrangler d1 execute webhooker --remote --file ./migrations/0004_add_group_id.sql bunx wrangler d1 execute webhooker --remote --file ./migrations/0005_audit_logs.sql +bunx wrangler d1 execute webhooker --remote --file ./migrations/0006_config_d1.sql +bunx wrangler d1 execute webhooker --remote --file ./migrations/0007_send_logs_index.sql +bunx wrangler d1 execute webhooker --remote --file ./migrations/0008_storage_d1.sql ``` ::: -### 4. 创建队列（可选） +### 4. 创建 R2 Bucket（可选） + +`PAYLOAD` 绑定将超大 webhook 负载暂存在 R2（`webhooker-payloads`）而非 KV。未设置时，超大负载回退到 KV 键 `queue:payload:*`。布局见[存储](/zh/guide/storage)。 + +```bash +bunx wrangler r2 bucket create webhooker-payloads +``` + +bucket 已在 `wrangler.jsonc` 中声明（`r2_buckets`），无需修改绑定。 + +### 5. 创建队列（可选） `QUEUE` 绑定会通过 Cloudflare Queues 投递 webhook（异步分发，带重试退避与死信队列）。跳过此步则保持同步内联分发。 @@ -96,7 +109,7 @@ 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. 部署 +### 6. 部署 ```bash bunx wrangler deploy @@ -104,13 +117,13 @@ bunx wrangler deploy Worker 现在可通过 `https://webhooker..workers.dev` 访问。 -### 6. 配置 GitHub Webhook +### 7. 配置 GitHub Webhook 1. 进入 GitHub App 设置页面 2. 设置 **Webhook URL** 为 `https://webhooker..workers.dev/webhook` 3. 设置 **Webhook secret** 与 `GITHUB_WEBHOOK_SECRET` 一致 -### 7.（可选）配置 Gitea Webhook +### 8.（可选）配置 Gitea Webhook 1. 在 Gitea 仓库中进入 **设置 → Web 钩子 → 添加 Web 钩子 → Gitea** 2. 设置 **目标 URL** 为 `https://webhooker..workers.dev/webhook` diff --git a/docs/zh/guide/faq.md b/docs/zh/guide/faq.md index 945a34d..8869f00 100644 --- a/docs/zh/guide/faq.md +++ b/docs/zh/guide/faq.md @@ -34,4 +34,4 @@ ## 数据存储在哪里？ -配置存于 Cloudflare KV（`config:routes`、`config:groups`）；发送/审计日志与平台↔GitHub 绑定存于 D1。见[存储布局](./storage#kv-存储布局)。 +配置存于 D1（`d1_routes`/`d1_groups`，KV 仅作缓存）；webhook 去重、投递状态与消息更新追踪同样存于 D1（KV 回退）；发送/审计日志与平台↔GitHub 绑定存于 D1；超大负载可选存于 R2。见[存储布局](./storage#存储决策)。 diff --git a/docs/zh/guide/groups.md b/docs/zh/guide/groups.md index 403f1c4..e5fd27e 100644 --- a/docs/zh/guide/groups.md +++ b/docs/zh/guide/groups.md @@ -1,6 +1,6 @@ # 分组与访问控制 -路由归属于分组。分组用于划分管理权限，并可限制进入其中的事件。它们以 JSON 数组形式存储在 Cloudflare KV 的 `config:groups` 键下，可通过 [Web 控制台](./configuration#web-控制台)或 [Admin API](../api/admin) 管理。每个实例最多可保存 **100 个分组**。 +路由归属于分组。分组用于划分管理权限，并可限制进入其中的事件。它们存储在 D1（`d1_groups`，首次加载时从旧版 KV `config:groups` 键同步），可通过 [Web 控制台](./configuration#web-控制台)或 [Admin API](../api/admin) 管理。每个实例最多可保存 **100 个分组**。 ## 分组模式 diff --git a/docs/zh/guide/introduction.md b/docs/zh/guide/introduction.md index 8f3f5a2..4e75bea 100644 --- a/docs/zh/guide/introduction.md +++ b/docs/zh/guide/introduction.md @@ -22,14 +22,14 @@ GitHub / Gitea Webhook → Cloudflare Worker (Nuxt 4 / Nitro) | ------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------- | | **Cloudflare Worker** | HTTP 入口、签名验证、投递去重、事件解析、路由匹配、平台分发 | | **Interactions Endpoint** | 验证 Ed25519 签名并处理 `/gh` 交互（斜杠命令、右键菜单、按钮、modal） | -| **KV** | Token 存储 (`token:{userId}`)、OAuth 状态 (`state:{hex}`)、路由配置 (`config:routes`)、分组配置 (`config:groups`)、管理员会话、投递去重、消息更新追踪 (`msg:*`) | -| **D1** | 发送日志 (`send_logs`)、Discord↔GitHub 绑定 (`discord_links`)、Telegram↔GitHub 绑定 (`telegram_links`) | +| **KV** | Token 存储 (`token:{userId}`)、OAuth 状态 (`state:{hex}`)、管理员会话、分组级 secret、配置缓存、投递去重/投递状态/消息更新追踪的回退（`delivery:*`、`delivery-state:*`、`msg:*`，仅在 D1 不可用时使用）、消息更新锁 (`msg:lock:*`) | +| **D1** | 路由/分组 (`d1_routes`/`d1_groups`)、发送日志 (`send_logs`)、审计日志 (`audit_logs`)、去重 (`dedup_keys`)、投递状态 (`delivery_state`)、消息追踪 (`message_tracking`)、Discord↔GitHub 绑定 (`discord_links`)、Telegram↔GitHub 绑定 (`telegram_links`)，超大负载可选的 R2 存储 | ### 数据流 1. 某个 forge（GitHub 或 Gitea）发送 webhook 到 `POST /webhook` 2. Worker 根据请求头识别提供方（`X-GitHub-Event` / `X-Gitea-Event`）并验证对应提供的 HMAC-SHA256 签名 -3. Worker 按投递 ID 去重（KV，短 TTL），丢弃重复投递 +3. Worker 按投递 ID 去重（D1，短 TTL，KV 回退），丢弃重复投递 4. Worker 解析事件类型并将载荷归一化为 GitHub 形状的事件 5. 根据过滤器（event、repo、actor、action、branch、keyword）与分组所有者限制评估路由 6. 匹配的路由触发格式化器函数生成平台中立消息 @@ -42,7 +42,7 @@ GitHub / Gitea Webhook → Cloudflare Worker (Nuxt 4 / Nitro) - **Discord 投递**: Discord REST API（交互通过 Ed25519 验签的 HTTPS Interactions Endpoint） - **Telegram 投递**: Telegram Bot API（webhook 带可选 secret-token 校验） - **Web UI**: Nuxt 4（Vue 3 + Tailwind CSS v3）——首页/法律页面服务端渲染，`/admin` 控制台客户端渲染 -- **存储**: Cloudflare KV + D1 +- **存储**: Cloudflare D1（配置与日志权威）+ KV（缓存/临时状态）+ 可选 R2（超大负载） - **鉴权**: Web Crypto API (HMAC-SHA256、Ed25519)、octokit (GitHub API) - **语言**: TypeScript diff --git a/docs/zh/guide/routes.md b/docs/zh/guide/routes.md index 64bc4f0..6809a1b 100644 --- a/docs/zh/guide/routes.md +++ b/docs/zh/guide/routes.md @@ -1,6 +1,6 @@ # 路由与目标 -路由决定哪些事件被转发到哪个频道（Discord 或 Telegram）。它们以 JSON 数组形式存储在 Cloudflare KV 的 `config:routes` 键下，可通过 [Web 控制台](./configuration#web-控制台)、[Admin API](../api/admin) 或 `config.example.yaml` 管理。 +路由决定哪些事件被转发到哪个频道（Discord 或 Telegram）。它们存储在 D1（`d1_routes`，首次加载时从旧版 KV `config:routes` 键同步），可通过 [Web 控制台](./configuration#web-控制台)、[Admin API](../api/admin) 或 `config.example.yaml` 管理。 **没有默认路由**——每条路由都必须定义自己的目标。未配置任何路由时不会转发任何事件。每个实例最多可保存 **200 条路由**。 diff --git a/docs/zh/guide/storage.md b/docs/zh/guide/storage.md index e9fe1f2..5683d74 100644 --- a/docs/zh/guide/storage.md +++ b/docs/zh/guide/storage.md @@ -2,22 +2,24 @@ ## KV 存储布局 +KV 只保留缓存数据和短期/临时状态。高频写入（webhook 去重、投递状态、消息追踪）存放在 D1，仅在 D1 不可用或未迁移时才回退到 KV（见[存储决策](#存储决策)）。 + | 键模式 | 值 | TTL | | ------------------------------------------ | ----------------------------------------------------------------------------- | ------------------ | -| `config:routes` | 路由 JSON 数组 | 永久 | -| `config:groups` | 分组 JSON 数组 | 永久 | +| `config:routes` | 路由配置缓存（D1 `d1_routes` 为权威数据源） | 1 小时 | +| `config:groups` | 分组配置缓存（D1 `d1_groups` 为权威数据源） | 1 小时 | | `session:{id}` | 管理员会话 `{ userId, login }` | 7 天 | | `token:{userId}` | `{ userId, accessToken, expiresAt, refreshToken? }` | 0.9 × Token 有效期 | | `token-reverse:{sha256}` | 用于按 Token 反查的用户 id | 0.9 × Token 有效期 | | `state:{hex}` | `{ redirectTo, expiresAt, discordUserId?, telegramUserId?, telegramChatId? }` | 600 秒 | | `invite:{token}` | `{ groupId, role, expiresAt, createdBy, note? }` | 7 天 | -| `invite:group:{id}` | 每组的 Token 索引（保证邀请列表一致性） | 永久 | -| `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 天 | +| `invite:group:{id}` | 每组的 Token 索引（保证邀请列表一致性） | 7 天 | +| `delivery:{provider}:{groupId}:{id}` | Webhook 投递去重回退（D1 `dedup_keys` 为主） | 7 天 | +| `delivery-state:{provider}:{groupId}:{id}` | 队列投递状态回退（D1 `delivery_state` 为主） | 1 小时 | +| `queue:payload:{provider}:{groupId}:{id}` | 暂存供队列消费者读取的超大 webhook 负载（R2 为主） | 1 小时 | | `nonce:{nonce}` | 自定义 webhook 重放防护 nonce（一次性） | 600 秒 | | `tenant:{groupId}` | 分组 webhook secret（64 位 hex，控制台生成） | 永久 | -| `msg:{routeId}:{key}:{target}` | 原地更新用消息 id 追踪（如 `workflow_run` / `check_run`） | 7 天 | +| `msg:{routeId}:{key}:{target}` | 消息 id 追踪回退（D1 `message_tracking` 为主） | 1 天 | | `cmd:guild:{id}` | 已注册命令的服务器 id（去重） | 永久 | | `cmd:registered:global` | 全局命令注册标记（去重） | 1 天 | | `config:discord-app-id` | 缓存的 Discord 应用 id | 永久 | @@ -25,13 +27,35 @@ ## D1 存储布局 -D1 数据库（`DB` 绑定，数据库 `webhooker`）包含四张表： +D1 数据库（`DB` 绑定，数据库 `webhooker`）保存配置、投递日志和高频临时状态的权威数据源： -| 表 | 用途 | -| ---------------- | ---------------------------------------------------------------------- | -| `send_logs` | 每次分发尝试一行（路由 id、事件、目标、ok/error、耗时、错误码、详情） | -| `audit_logs` | 每次管理员操作一行（登录/登出、分组/路由/成员/邀请变更） | -| `discord_links` | 映射 `discord_user_id` → `github_user_id`，供 `/gh` Discord 命令使用 | -| `telegram_links` | 映射 `telegram_user_id` → `github_user_id`，供 `/gh` Telegram 命令使用 | +| 表 | 用途 | +| ------------------ | ---------------------------------------------------------------------- | +| `d1_groups` | 分组（权威配置，从旧版 KV `config:groups` 播种） | +| `d1_routes` | 每组分组的路由（权威配置，从旧版 KV `config:routes` 播种） | +| `send_logs` | 每次分发尝试一行（路由 id、事件、目标、ok/error、耗时、错误码、详情） | +| `audit_logs` | 每次管理员操作一行（登录/登出、分组/路由/成员/邀请变更） | +| `dedup_keys` | Webhook 投递去重（原子 `INSERT ... ON CONFLICT` UPSERT，键 + 过期时间） | +| `delivery_state` | 队列投递状态（`pending`/`processing`/`delivered`/`retrying`/`failed`/`dead`） | +| `message_tracking` | 原地更新用消息 id 追踪（`event_id` + `target_id` → `message_id`） | +| `discord_links` | 映射 `discord_user_id` → `github_user_id`，供 `/gh` Discord 命令使用 | +| `telegram_links` | 映射 `telegram_user_id` → `github_user_id`，供 `/gh` Telegram 命令使用 | -`audit_logs` 由定时任务在 `AUDIT_RETENTION_DAYS`（默认 90）后自动清理。行字段说明见[日志](./logs)。 +`audit_logs` 由定时任务在 `AUDIT_RETENTION_DAYS`（默认 90）后自动清理。`storage-prune` 任务会清理过期的 `dedup_keys`、超过 7 天的 `delivery_state` 行以及超过 30 天的 `message_tracking` 行。日志行字段说明见[日志](./logs)。 + +## R2 存储布局 + +R2（`PAYLOAD` 绑定，bucket `webhooker-payloads`）存储对队列消息或 KV 来说过大的 webhook 负载： + +| 对象模式 | 用途 | 保留期 | +| ------------------------------- | ----------------------------------------------- | ------------ | +| `webhooks/YYYY/MM/DD/.json` | 暂存供队列消费者读取的超大负载 | 分发后删除 | + +当缺少 `PAYLOAD` 绑定时，超大负载回退到 KV 键 `queue:payload:{provider}:{groupId}:{id}`（1 小时 TTL）。 + +## 存储决策 + +- **D1 是配置与投递元数据的权威数据源**；KV 只保存缓存和短期状态。 +- `canUseD1` 探测（`server/lib/storage/d1.ts`，检查 `prepare` + `batch`）为每个 D1 存储做门槛判定：当 D1 不可用或尚未迁移时，三种高频存储（去重、投递状态、消息追踪）都会透明回退到 KV，迁移期间行为不变。 +- 这让 Workers 免费版的每事件 KV 写入趋近于零（每日 1000 次写）：去重、投递状态和消息追踪改为写 D1 行（D1 免费版每日可写 10 万行）。 +- R2 免费额度（10 GB-月存储、每月 100 万次 A 类操作）可以轻松承载负载暂存，不占用 KV 写配额。 diff --git a/docs/zh/guide/tasks.md b/docs/zh/guide/tasks.md index bdd139e..284cfba 100644 --- a/docs/zh/guide/tasks.md +++ b/docs/zh/guide/tasks.md @@ -1,11 +1,12 @@ # 定时任务 -WebHooker 通过定时触发器（`*/5 * * * *`，每 5 分钟）运行三个维护任务。它们只在部署后的 Worker 上运行（Cloudflare cron）；本地 `wrangler dev` 可用 `wrangler dev --test-scheduled` 触发。 +WebHooker 通过定时触发器（`*/5 * * * *`，每 5 分钟）运行四个维护任务。它们只在部署后的 Worker 上运行（Cloudflare cron）；本地 `wrangler dev` 可用 `wrangler dev --test-scheduled` 触发。 | 任务 | 用途 | | --------------- | ------------------------------------------------------------------------------------------------------------------- | | `discord-sync` | 注册 Discord 斜杠/右键菜单命令：按服务器即时注册，并全局注册（24h 去重，约 1 小时传播） | | `telegram-sync` | 调用 `setWebhook` 指向 `{BASE_URL}/telegram/webhook`（设置了 `TELEGRAM_WEBHOOK_SECRET` 时作为 `secret_token` 传入） | | `audit-prune` | 删除早于 `AUDIT_RETENTION_DAYS`（默认 90）天的 `audit_logs` 记录 | +| `storage-prune` | 删除已过期的 `dedup_keys` 记录、超过 7 天的 `delivery_state` 记录和超过 30 天的 `message_tracking` 记录 | 除任务用到的密钥（`DISCORD_TOKEN`、`DISCORD_APPLICATION_ID`、`TELEGRAM_TOKEN`、`BASE_URL`、`AUDIT_RETENTION_DAYS`）外无需其他配置。 diff --git a/migrations/0006_config_d1.sql b/migrations/0006_config_d1.sql new file mode 100644 index 0000000..5259b75 --- /dev/null +++ b/migrations/0006_config_d1.sql @@ -0,0 +1,28 @@ +CREATE TABLE IF NOT EXISTS d1_groups ( + id TEXT PRIMARY KEY, + name TEXT NOT NULL, + data TEXT NOT NULL, + version INTEGER NOT NULL DEFAULT 1, + created_at INTEGER NOT NULL, + updated_at INTEGER NOT NULL +); + +CREATE TABLE IF NOT EXISTS d1_routes ( + id TEXT NOT NULL, + group_id TEXT NOT NULL, + name TEXT NOT NULL, + enabled INTEGER NOT NULL DEFAULT 1, + filters TEXT NOT NULL, + targets TEXT NOT NULL, + stop INTEGER NOT NULL DEFAULT 0, + fallback INTEGER NOT NULL DEFAULT 0, + discord_role_ids TEXT, + ast TEXT, + version INTEGER NOT NULL DEFAULT 1, + created_at INTEGER NOT NULL, + updated_at INTEGER NOT NULL, + PRIMARY KEY (id, group_id), + FOREIGN KEY (group_id) REFERENCES d1_groups(id) ON DELETE CASCADE +); + +CREATE INDEX IF NOT EXISTS idx_d1_routes_group_id ON d1_routes(group_id); \ No newline at end of file diff --git a/migrations/0007_send_logs_index.sql b/migrations/0007_send_logs_index.sql new file mode 100644 index 0000000..f051d57 --- /dev/null +++ b/migrations/0007_send_logs_index.sql @@ -0,0 +1 @@ +CREATE INDEX IF NOT EXISTS idx_send_logs_group_id_ts ON send_logs (group_id, ts DESC); \ No newline at end of file diff --git a/migrations/0008_storage_d1.sql b/migrations/0008_storage_d1.sql new file mode 100644 index 0000000..a1681cb --- /dev/null +++ b/migrations/0008_storage_d1.sql @@ -0,0 +1,25 @@ +-- D1 tables replacing ephemeral high-frequency KV keys (dedup, delivery state, +-- message tracking). Keeps KV for things that still benefit from key-based +-- access with short TTL; these tables relieve the KV write quota. + +CREATE TABLE IF NOT EXISTS dedup_keys ( + key TEXT PRIMARY KEY, + claimed_at INTEGER NOT NULL, + expires_at INTEGER NOT NULL +); +CREATE INDEX IF NOT EXISTS idx_dedup_keys_expires_at ON dedup_keys (expires_at); + +CREATE TABLE IF NOT EXISTS delivery_state ( + key TEXT PRIMARY KEY, + status TEXT NOT NULL, + updated_at INTEGER NOT NULL +); + +CREATE TABLE IF NOT EXISTS message_tracking ( + event_id TEXT NOT NULL, + target_id TEXT NOT NULL, + message_id TEXT NOT NULL, + updated_at INTEGER NOT NULL, + PRIMARY KEY (event_id, target_id) +); +CREATE INDEX IF NOT EXISTS idx_message_tracking_updated_at ON message_tracking (updated_at); \ No newline at end of file diff --git a/nuxt.config.ts b/nuxt.config.ts index 782b8f9..e5d978c 100644 --- a/nuxt.config.ts +++ b/nuxt.config.ts @@ -9,7 +9,7 @@ export default defineNuxtConfig({ tasks: true, }, scheduledTasks: { - "*/5 * * * *": ["discord-sync", "telegram-sync", "audit-prune"], + "*/5 * * * *": ["discord-sync", "telegram-sync", "audit-prune", "storage-prune"], }, }, app: { diff --git a/server/lib/config.ts b/server/lib/config.ts index 0bc6a5b..85f1577 100644 --- a/server/lib/config.ts +++ b/server/lib/config.ts @@ -1,12 +1,41 @@ import type { Env, Config, Route } from "./types"; import { log } from "./lib/log"; import { migrateRoutes, validateRoutes } from "./config/schema"; +import { d1ConfigStore, type ConfigStore } from "./storage/config-store"; const CONFIG_CACHE_TTL = 300_000; const ROUTES_KEY = "config:routes"; let configCache: { config: Config; expiresAt: number } | null = null; +const configStores = new WeakMap(); + +export function getConfigStore(kv: KVNamespace): ConfigStore | null { + return configStores.get(kv) ?? null; +} + +export function initConfigStore(env: Env): ConfigStore { + return ensureStore(env); +} + +export function resetConfigStore(kv?: KVNamespace): void { + if (kv) { + configStores.get(kv)?.invalidateCache(); + configStores.delete(kv); + } +} + +function ensureStore(env: Env): ConfigStore { + let store = configStores.get(env.KV); + if (!store) { + store = d1ConfigStore(env.DB, env.KV); + configStores.set(env.KV, store); + } + return store; +} export async function loadRoutes(kv: KVNamespace): Promise { + const store = configStores.get(kv); + if (store) return store.loadRoutes(); + try { const stored = await kv.get(ROUTES_KEY, "json"); if (stored) return validateRoutes(migrateRoutes(stored)); @@ -17,11 +46,16 @@ export async function loadRoutes(kv: KVNamespace): Promise { } export async function saveRoutes(kv: KVNamespace, routes: Route[]): Promise { + const store = configStores.get(kv); + if (store) { + await store.saveRoutes(routes); + configCache = null; + return; + } await kv.put(ROUTES_KEY, JSON.stringify(routes)); configCache = null; } -/** Drop the in-memory route/config cache (used by the admin API and tests). */ export function invalidateConfigCache(): void { configCache = null; } @@ -31,6 +65,7 @@ export async function loadConfig(env: Env): Promise { return configCache.config; } + ensureStore(env); const routes = await loadRoutes(env.KV); const config: Config = { @@ -50,4 +85,4 @@ export async function loadConfig(env: Env): Promise { configCache = { config, expiresAt: Date.now() + CONFIG_CACHE_TTL }; return config; -} +} \ No newline at end of file diff --git a/server/lib/core/dispatch.ts b/server/lib/core/dispatch.ts index e47cec1..73618a6 100644 --- a/server/lib/core/dispatch.ts +++ b/server/lib/core/dispatch.ts @@ -4,8 +4,9 @@ import { emojiPrefix, forgeInfo } from "../formatters/helpers"; import { matchRoute, eventOwners } from "../events/match"; import { log } from "../lib/log"; import { loadTranslations, t as translate, type Translations } from "../lib/i18n"; -import { recordSend } from "../lib/send-log"; -import { kvMessageTracker } from "../lib/message-tracker"; +import type { SendRecord } from "../lib/send-log"; +import { recordSendBatch } from "../lib/send-log-batch"; +import { messageTracker } from "../lib/message-tracker"; import { loadGroups, groupAcceptsOwners, @@ -36,7 +37,7 @@ export async function dispatchEvent( ): Promise { const loadedGroups = groups ?? (await loadGroups(env.KV)); const groupById = new Map(loadedGroups.map((g) => [g.id, g])); - const tracker = kvMessageTracker(env.KV); + const tracker = messageTracker(env.DB, env.KV); // Message language is configured per group (Group.lang), not per route. const langs = [...new Set(loadedGroups.map((g) => g.lang ?? "en"))]; @@ -63,6 +64,7 @@ export async function dispatchEvent( const anyRegularMatched = matched.length > 0; const attempts: DispatchAttempt[] = []; + const sendLogs: SendRecord[] = []; const tasks: Promise[] = []; for (const route of config.routes) { if (!accepted(route)) continue; @@ -79,6 +81,7 @@ export async function dispatchEvent( } await Promise.allSettled(tasks); + await recordSendBatch(env.DB, sendLogs); await sendGroupLogs(attempts); const failures: DispatchFailure[] = attempts @@ -235,7 +238,7 @@ export async function dispatchEvent( target: targetStr, ok: true, }); - await recordSend(env.DB, { + sendLogs.push({ ...base, ok: true, status: result.status, @@ -271,7 +274,7 @@ export async function dispatchEvent( target: targetStr, ok: true, }); - await recordSend(env.DB, { + sendLogs.push({ ...base, ok: true, status: result.status, @@ -291,7 +294,7 @@ export async function dispatchEvent( target: targetStr, ok: true, }); - await recordSend(env.DB, { + sendLogs.push({ ...base, ok: true, status: result.status, @@ -328,7 +331,7 @@ export async function dispatchEvent( errorCode: result.errorCode, status: result.status, }); - await recordSend(env.DB, { + sendLogs.push({ ...base, ok: false, error, @@ -347,7 +350,7 @@ export async function dispatchEvent( target: targetStr, ok: true, }); - await recordSend(env.DB, { + sendLogs.push({ ...base, ok: true, status: result.status, @@ -368,7 +371,7 @@ export async function dispatchEvent( ok: false, error, }); - await recordSend(env.DB, { + sendLogs.push({ ...base, ok: false, error, diff --git a/server/lib/lib/idempotency.ts b/server/lib/lib/idempotency.ts index 2de9607..41debba 100644 --- a/server/lib/lib/idempotency.ts +++ b/server/lib/lib/idempotency.ts @@ -1,10 +1,11 @@ +import { canUseD1 } from "../storage/d1"; + /** * Delivery idempotency: a small, reusable abstraction over "has this key been * seen / can I claim it once" so webhook dedup, nonce replay protection and - * future delivery retry all share one semantics. `claim` is best-effort atomic - * on KV (get-then-put): under a concurrent double-send both callers may see - * "unclaimed", but that matches the existing dedup behavior and is acceptable - * because dispatch itself is idempotent per (provider, group, delivery). + * future delivery retry all share one semantics. Backed by D1 when the binding + * is present (atomic `INSERT OR IGNORE`), falling back to a best-effort + * get-then-put KV claim otherwise. */ export interface IdempotencyStore { has(key: string): Promise; @@ -26,6 +27,35 @@ export function kvIdempotencyStore(kv: KVNamespace): IdempotencyStore { }; } +export function d1IdempotencyStore(db: D1Database): IdempotencyStore { + return { + async has(key): Promise { + const row = await db + .prepare("SELECT 1 AS hit FROM dedup_keys WHERE key = ? AND expires_at > ?") + .bind(key, Date.now()) + .first<{ hit: number }>(); + return row != null; + }, + async claim(key, ttlSeconds): Promise { + const now = Date.now(); + const result = await db + .prepare( + `INSERT INTO dedup_keys (key, claimed_at, expires_at) VALUES (?, ?, ?) + ON CONFLICT(key) DO UPDATE SET + claimed_at = excluded.claimed_at, expires_at = excluded.expires_at + WHERE dedup_keys.expires_at < excluded.expires_at`, + ) + .bind(key, now, now + ttlSeconds * 1000) + .run(); + return (result.meta?.changes ?? 0) > 0; + }, + }; +} + +export function idempotencyStore(db: D1Database, kv: KVNamespace): IdempotencyStore { + return canUseD1(db) ? d1IdempotencyStore(db) : kvIdempotencyStore(kv); +} + /** * Canonical dedup key for a webhook delivery: provider + tenant + delivery id. * The legacy global endpoint has no tenant, so `groupId` is "global". @@ -36,4 +66,4 @@ export function deliveryKey( deliveryId: string, ): string { return `delivery:${provider}:${groupId ?? "global"}:${deliveryId}`; -} +} \ No newline at end of file diff --git a/server/lib/lib/message-tracker.ts b/server/lib/lib/message-tracker.ts index d3cf873..765fa8d 100644 --- a/server/lib/lib/message-tracker.ts +++ b/server/lib/lib/message-tracker.ts @@ -1,3 +1,5 @@ +import { canUseD1 } from "../storage/d1"; + export interface MessageTracker { get(eventId: string, targetId: string): Promise; set(eventId: string, targetId: string, messageId: string): Promise; @@ -22,3 +24,36 @@ export function kvMessageTracker(kv: KVNamespace): MessageTracker { }, }; } + +export function d1MessageTracker(db: D1Database): MessageTracker { + return { + async get(eventId: string, targetId: string): Promise { + const row = await db + .prepare("SELECT message_id FROM message_tracking WHERE event_id = ? AND target_id = ?") + .bind(eventId, targetId) + .first<{ message_id: string }>(); + return row?.message_id ?? null; + }, + async set(eventId: string, targetId: string, messageId: string): Promise { + await db + .prepare( + `INSERT INTO message_tracking (event_id, target_id, message_id, updated_at) + VALUES (?, ?, ?, ?) + ON CONFLICT(event_id, target_id) DO UPDATE SET + message_id = excluded.message_id, updated_at = excluded.updated_at`, + ) + .bind(eventId, targetId, messageId, Date.now()) + .run(); + }, + async delete(eventId: string, targetId: string): Promise { + await db + .prepare("DELETE FROM message_tracking WHERE event_id = ? AND target_id = ?") + .bind(eventId, targetId) + .run(); + }, + }; +} + +export function messageTracker(db: D1Database, kv: KVNamespace): MessageTracker { + return canUseD1(db) ? d1MessageTracker(db) : kvMessageTracker(kv); +} \ No newline at end of file diff --git a/server/lib/lib/send-log-batch.ts b/server/lib/lib/send-log-batch.ts new file mode 100644 index 0000000..4203616 --- /dev/null +++ b/server/lib/lib/send-log-batch.ts @@ -0,0 +1,41 @@ +import type { SendRecord } from "./send-log"; +import { log } from "./log"; + +export async function recordSendBatch( + db: D1Database, + records: SendRecord[], +): Promise { + if (records.length === 0) return; + try { + const stmts = records.map((r) => + db + .prepare( + `INSERT INTO send_logs (ts, route_id, group_id, event, repo, target, ok, error, status, message_id, delivery_id, platform, actor, action, duration_ms, error_code, attempts, detail) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + ) + .bind( + r.ts, + r.routeId, + r.groupId ?? null, + r.event, + r.repo ?? null, + r.target, + r.ok ? 1 : 0, + r.error ?? null, + r.status ?? null, + r.messageId ?? null, + r.deliveryId ?? null, + r.platform ?? null, + r.actor ?? null, + r.action ?? null, + r.durationMs ?? null, + r.errorCode ?? null, + r.attempts ?? null, + r.detail ? JSON.stringify(r.detail) : null, + ), + ); + await db.batch(stmts); + } catch (err) { + log.warn({ err, count: records.length }, "Failed to record send log batch"); + } +} \ No newline at end of file diff --git a/server/lib/queue/delivery.ts b/server/lib/queue/delivery.ts index 4a9d45b..ae856f2 100644 --- a/server/lib/queue/delivery.ts +++ b/server/lib/queue/delivery.ts @@ -1,7 +1,9 @@ import type { Env } from "../types"; +import { r2PayloadStore } from "../storage/payload"; +import { canUseD1 } from "../storage/d1"; export type DeliveryStatus = - "pending" | "processing" | "delivered" | "retrying" | "failed" | "dead"; + | "pending" | "processing" | "delivered" | "retrying" | "failed" | "dead"; export interface DeliveryMessage { deliveryId: string; @@ -9,7 +11,10 @@ export interface DeliveryMessage { provider: string; event: string; payload?: Record; + /** R2 object key (new) or KV key (legacy `queue:payload:*`). */ payloadRef?: string; + /** Distinguishes R2 payload refs from legacy KV keys. */ + payloadRefType?: "r2" | "kv"; installationId?: number; receivedAt: number; requestId?: string; @@ -79,12 +84,32 @@ export async function setDeliveryState( key: string, status: DeliveryStatus, ): Promise { + const db = env.DB; + if (canUseD1(db)) { + await db + .prepare( + `INSERT INTO delivery_state (key, status, updated_at) + VALUES (?, ?, ?) + ON CONFLICT(key) DO UPDATE SET status = excluded.status, updated_at = excluded.updated_at`, + ) + .bind(key, status, Date.now()) + .run(); + return; + } 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 db = env.DB; + if (canUseD1(db)) { + const row = await db + .prepare("SELECT status FROM delivery_state WHERE key = ?") + .bind(key) + .first<{ status: DeliveryStatus }>(); + return row?.status ?? null; + } const raw = await env.KV.get(key); if (!raw) return null; try { @@ -103,11 +128,18 @@ export async function enqueueWebhook(env: Env, message: DeliveryMessage): Promis await queue.send(direct); return; } + const bucket = env.PAYLOAD; + if (bucket) { + const store = r2PayloadStore(env); + const key = await store.put(JSON.stringify(payload ?? {}), PAYLOAD_KV_TTL_SECONDS); + await queue.send({ ...rest, payloadRef: key, payloadRefType: "r2" }); + 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 }); + await queue.send({ ...rest, payloadRef, payloadRefType: "kv" }); } export async function resolvePayload( @@ -116,6 +148,17 @@ export async function resolvePayload( ): Promise> { if (message.payload) return message.payload; if (message.payloadRef) { + if (message.payloadRefType === "r2" && env.PAYLOAD) { + const store = r2PayloadStore(env); + const raw = await store.get(message.payloadRef); + if (raw) { + try { + return JSON.parse(raw) as Record; + } catch { + return {}; + } + } + } const raw = await env.KV.get(message.payloadRef); if (raw) { try { @@ -129,5 +172,11 @@ export async function resolvePayload( } export async function discardPayload(env: Env, message: DeliveryMessage): Promise { - if (message.payloadRef) await env.KV.delete(message.payloadRef); + if (!message.payloadRef) return; + if (message.payloadRefType === "r2" && env.PAYLOAD) { + const store = r2PayloadStore(env); + await store.delete(message.payloadRef); + return; + } + await env.KV.delete(message.payloadRef); } diff --git a/server/lib/storage/config-store.ts b/server/lib/storage/config-store.ts new file mode 100644 index 0000000..985e3d1 --- /dev/null +++ b/server/lib/storage/config-store.ts @@ -0,0 +1,242 @@ +import type { Group, Route } from "../types"; +import { log } from "../lib/log"; + +export interface ConfigStore { + loadRoutes(): Promise; + saveRoutes(routes: Route[]): Promise; + loadGroups(): Promise; + saveGroups(groups: Group[]): Promise; + invalidateCache(): void; +} + +interface D1GroupRow { + id: string; + name: string; + data: string; + version: number; +} + +interface D1RouteRow { + id: string; + group_id: string; + name: string; + enabled: number; + filters: string; + targets: string; + stop: number; + fallback: number; + discord_role_ids: string | null; + ast: string | null; +} + +const CACHE_TTL = 300_000; +const KV_ROUTES_KEY = "config:routes"; +const KV_GROUPS_KEY = "config:groups"; + +const KV_CACHE_TTL = 3600; + +export function d1ConfigStore(db: D1Database, kv: KVNamespace): ConfigStore { + let routesCache: { routes: Route[]; expiresAt: number } | null = null; + let groupsCache: { groups: Group[]; expiresAt: number } | null = null; + + async function loadRoutesFromD1(): Promise { + const stmt = db.prepare( + "SELECT id, group_id, name, enabled, filters, targets, stop, fallback, discord_role_ids, ast FROM d1_routes ORDER BY id", + ); + if (typeof stmt.all !== "function") return []; + const { results } = await stmt.all(); + if (!results || results.length === 0) return []; + return results.map((r) => ({ + id: r.id, + groupId: r.group_id, + name: r.name, + enabled: r.enabled === 1, + filters: JSON.parse(r.filters), + targets: JSON.parse(r.targets), + stop: r.stop === 1, + fallback: r.fallback === 1, + discordRoleIds: r.discord_role_ids ? JSON.parse(r.discord_role_ids) : undefined, + ast: r.ast ? JSON.parse(r.ast) : undefined, + })); + } + + async function loadGroupsFromD1(): Promise { + const stmt = db.prepare("SELECT id, name, data, version FROM d1_groups ORDER BY id"); + if (typeof stmt.all !== "function") return []; + const { results } = await stmt.all(); + if (!results || results.length === 0) return []; + return results.map((r) => JSON.parse(r.data) as Group); + } + + async function loadRoutesFromKV(): Promise { + try { + const raw = await kv.get(KV_ROUTES_KEY, "json"); + if (raw) return raw; + } catch (err) { + log.warn({ err }, "Failed to load routes from KV"); + } + return []; + } + + async function loadGroupsFromKV(): Promise { + try { + const raw = await kv.get(KV_GROUPS_KEY, "json"); + if (raw) return raw; + } catch (err) { + log.warn({ err }, "Failed to load groups from KV"); + } + return []; + } + + function routeStatements(routes: Route[]): D1PreparedStatement[] { + const now = Date.now(); + return [ + db.prepare("DELETE FROM d1_routes"), + ...routes.map((r) => + db + .prepare( + `INSERT INTO d1_routes (id, group_id, name, enabled, filters, targets, stop, fallback, discord_role_ids, ast, version, created_at, updated_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 1, ?, ?)`, + ) + .bind( + r.id, + r.groupId ?? "", + r.name, + r.enabled ? 1 : 0, + JSON.stringify(r.filters), + JSON.stringify(r.targets), + r.stop ? 1 : 0, + r.fallback ? 1 : 0, + r.discordRoleIds ? JSON.stringify(r.discordRoleIds) : null, + r.ast ? JSON.stringify(r.ast) : null, + now, + now, + ), + ), + ]; + } + + function groupStatements(groups: Group[]): D1PreparedStatement[] { + const now = Date.now(); + return [ + db.prepare("DELETE FROM d1_groups"), + ...groups.map((g) => + db + .prepare( + `INSERT INTO d1_groups (id, name, data, version, created_at, updated_at) + VALUES (?, ?, ?, 1, ?, ?)`, + ) + .bind(g.id, g.name, JSON.stringify(g), now, now), + ), + ]; + } + + async function seedRoutesToD1(routes: Route[]): Promise { + if (routes.length === 0) return; + await db.batch(routeStatements(routes)); + } + + async function seedGroupsToD1(groups: Group[]): Promise { + if (groups.length === 0) return; + await db.batch(groupStatements(groups)); + } + + async function syncRoutesToKV(routes: Route[], ttl: number): Promise { + try { + await kv.put(KV_ROUTES_KEY, JSON.stringify(routes), { expirationTtl: ttl }); + } catch (err) { + log.warn({ err }, "Failed to sync routes to KV cache"); + } + } + + async function syncGroupsToKV(groups: Group[], ttl: number): Promise { + try { + await kv.put(KV_GROUPS_KEY, JSON.stringify(groups), { expirationTtl: ttl }); + } catch (err) { + log.warn({ err }, "Failed to sync groups to KV cache"); + } + } + + return { + async loadRoutes(): Promise { + if (routesCache && Date.now() < routesCache.expiresAt) { + return routesCache.routes; + } + + let routes: Route[] = []; + try { + routes = await loadRoutesFromD1(); + if (routes.length > 0) { + syncRoutesToKV(routes, KV_CACHE_TTL).catch(() => undefined); + } else { + routes = await loadRoutesFromKV(); + if (routes.length > 0) { + seedRoutesToD1(routes).catch((err) => + log.warn({ err }, "Failed to seed routes from KV to D1"), + ); + } + } + } catch (err) { + log.warn({ err }, "D1 routes unavailable, falling back to KV"); + routes = await loadRoutesFromKV(); + } + + routesCache = { routes, expiresAt: Date.now() + CACHE_TTL }; + return routes; + }, + + async saveRoutes(routes: Route[]): Promise { + try { + await db.batch(routeStatements(routes)); + await syncRoutesToKV(routes, KV_CACHE_TTL); + } catch (err) { + log.warn({ err }, "D1 routes unavailable, falling back to KV"); + await syncRoutesToKV(routes, 0); + } + routesCache = null; + }, + + async loadGroups(): Promise { + if (groupsCache && Date.now() < groupsCache.expiresAt) { + return groupsCache.groups; + } + + let groups: Group[] = []; + try { + groups = await loadGroupsFromD1(); + if (groups.length > 0) { + syncGroupsToKV(groups, KV_CACHE_TTL).catch(() => undefined); + } else { + groups = await loadGroupsFromKV(); + if (groups.length > 0) { + seedGroupsToD1(groups).catch((err) => + log.warn({ err }, "Failed to seed groups from KV to D1"), + ); + } + } + } catch (err) { + log.warn({ err }, "D1 groups unavailable, falling back to KV"); + groups = await loadGroupsFromKV(); + } + + groupsCache = { groups, expiresAt: Date.now() + CACHE_TTL }; + return groups; + }, + + async saveGroups(groups: Group[]): Promise { + try { + await db.batch(groupStatements(groups)); + await syncGroupsToKV(groups, KV_CACHE_TTL); + } catch (err) { + log.warn({ err }, "D1 groups unavailable, falling back to KV"); + await syncGroupsToKV(groups, 0); + } + groupsCache = null; + }, + + invalidateCache(): void { + routesCache = null; + groupsCache = null; + }, + }; +} \ No newline at end of file diff --git a/server/lib/storage/d1.ts b/server/lib/storage/d1.ts new file mode 100644 index 0000000..47094de --- /dev/null +++ b/server/lib/storage/d1.ts @@ -0,0 +1,14 @@ +/** + * True when a D1Database binding looks like the real thing. Production D1 + * always exposes both `prepare` and `batch`; test harnesses range from + * `DB: {}` (no methods at all) to minimal mocks that only implement + * `prepare`→`bind`→{`run`,`all`} without `batch`. Using the presence of + * `batch` as the probe means every existing test keeps its KV fallback path + * while production eagerly routes through D1. + */ +export function canUseD1(db: D1Database | undefined | null): boolean { + return ( + typeof db?.prepare === "function" && + typeof (db as D1Database).batch === "function" + ); +} \ No newline at end of file diff --git a/server/lib/storage/payload.ts b/server/lib/storage/payload.ts new file mode 100644 index 0000000..5630d4b --- /dev/null +++ b/server/lib/storage/payload.ts @@ -0,0 +1,64 @@ +import type { Env } from "../types"; + +const PAYLOAD_PREFIX = "webhooks/"; +const DEFAULT_TTL_SECONDS = 3600; + +export interface PayloadStore { + put(payload: string, ttl?: number): Promise; + get(key: string): Promise; + delete(key: string): Promise; +} + +function generateKey(): string { + const bytes = new Uint8Array(16); + crypto.getRandomValues(bytes); + const hex = Array.from(bytes) + .map((b) => b.toString(16).padStart(2, "0")) + .join(""); + const now = new Date(); + const y = now.getUTCFullYear(); + const m = String(now.getUTCMonth() + 1).padStart(2, "0"); + const d = String(now.getUTCDate()).padStart(2, "0"); + return `${PAYLOAD_PREFIX}${y}/${m}/${d}/${hex}.json`; +} + +export function r2PayloadStore(env: Env): PayloadStore { + const bucket = env.PAYLOAD; + if (!bucket) { + return { + async put(_payload: string): Promise { + throw new Error("R2 binding is not configured"); + }, + async get(): Promise { + throw new Error("R2 binding is not configured"); + }, + async delete(): Promise { + throw new Error("R2 binding is not configured"); + }, + }; + } + + return { + async put(payload: string, ttl?: number): Promise { + const key = generateKey(); + const expires = new Date( + Date.now() + (ttl ?? DEFAULT_TTL_SECONDS) * 1000, + ); + await bucket.put(key, payload, { + httpMetadata: { contentType: "application/json" }, + customMetadata: { expires: expires.toISOString() }, + }); + return key; + }, + + async get(key: string): Promise { + const obj = await bucket.get(key); + if (!obj) return null; + return obj.text(); + }, + + async delete(key: string): Promise { + await bucket.delete(key); + }, + }; +} \ No newline at end of file diff --git a/server/lib/types.ts b/server/lib/types.ts index b745119..1207556 100644 --- a/server/lib/types.ts +++ b/server/lib/types.ts @@ -24,6 +24,7 @@ export interface Env { KV: KVNamespace; DB: D1Database; QUEUE?: Queue; + PAYLOAD?: R2Bucket; } export interface Config { diff --git a/server/lib/web/auth.ts b/server/lib/web/auth.ts index 1b6b8de..faddbea 100644 --- a/server/lib/web/auth.ts +++ b/server/lib/web/auth.ts @@ -14,6 +14,7 @@ import { } from "./groups"; import { findUserIdByToken } from "../github/store"; import { cfEnv } from "../cf"; +import { initConfigStore } from "../config"; export interface AuthContext { session: AdminSession; @@ -26,6 +27,7 @@ const AUTH_KEY = "auth"; /** Read the admin session + access scope for a request (null when logged out). */ export async function loadAuth(event: H3Event): Promise { const env = cfEnv(event); + initConfigStore(env); const session = await getAdminSession(env.KV, getHeader(event, "cookie")); if (!session) return null; const groups = await loadGroups(env.KV); diff --git a/server/lib/web/groups.ts b/server/lib/web/groups.ts index 3ca21ba..4834e6a 100644 --- a/server/lib/web/groups.ts +++ b/server/lib/web/groups.ts @@ -2,6 +2,7 @@ import type { Env, Group, GroupMember, GroupRole } from "../types"; import { isAdminUser } from "./session"; import { log } from "../lib/log"; import { migrateGroups, validateGroups } from "../config/schema"; +import { getConfigStore } from "../config"; const GROUPS_KEY = "config:groups"; const GROUPS_CACHE_TTL = 300_000; @@ -40,6 +41,9 @@ export function normalizeGroupMembers(group: Group): GroupMember[] { } export async function loadGroups(kv: KVNamespace): Promise { + const store = getConfigStore(kv); + if (store) return store.loadGroups(); + if (groupsCache && Date.now() < groupsCache.expiresAt) { return groupsCache.groups; } @@ -55,6 +59,12 @@ export async function loadGroups(kv: KVNamespace): Promise { } export async function saveGroups(kv: KVNamespace, groups: Group[]): Promise { + const store = getConfigStore(kv); + if (store) { + await store.saveGroups(groups); + groupsCache = null; + return; + } await kv.put(GROUPS_KEY, JSON.stringify(groups)); groupsCache = null; } diff --git a/server/lib/web/invites.ts b/server/lib/web/invites.ts index f8a8219..82b9ad5 100644 --- a/server/lib/web/invites.ts +++ b/server/lib/web/invites.ts @@ -44,7 +44,7 @@ async function readIndex(kv: KVNamespace, groupId: string): Promise { } async function writeIndex(kv: KVNamespace, groupId: string, tokens: string[]): Promise { - await kv.put(indexKey(groupId), JSON.stringify(tokens)); + await kv.put(indexKey(groupId), JSON.stringify(tokens), { expirationTtl: INVITE_TTL }); } async function removeFromIndex(kv: KVNamespace, groupId: string, token: string): Promise { @@ -142,7 +142,7 @@ export async function migrateInvites(kv: KVNamespace, from: string, to: string): moved.push(token); } } - await kv.put(indexKey(to), JSON.stringify(moved)); + await kv.put(indexKey(to), JSON.stringify(moved), { expirationTtl: INVITE_TTL }); await kv.delete(indexKey(from)); } catch (err) { log.warn({ err, from, to }, "Failed to migrate invites on group rename"); diff --git a/server/lib/web/oauth.ts b/server/lib/web/oauth.ts index 26d2f47..f39a790 100644 --- a/server/lib/web/oauth.ts +++ b/server/lib/web/oauth.ts @@ -28,6 +28,7 @@ import { clientIp } from "./auth"; import { recordAudit } from "../lib/audit"; import { sendMessage } from "../drivers/telegram/rest"; import { cfEnv } from "../cf"; +import { initConfigStore } from "../config"; import type { Env, Group } from "../types"; interface PendingState { @@ -130,6 +131,7 @@ function installPage(opts: { /** GET /auth/github — start the OAuth flow. */ export async function handleOAuthStart(event: H3Event): Promise { const env = cfEnv(event); + initConfigStore(env); const query = getQuery(event); const redirectTo = safeRedirectPath(String(query["redirect"] ?? "")); const state = generateRandomHex(16); @@ -144,6 +146,7 @@ export async function handleOAuthStart(event: H3Event): Promise { /** GET /auth/github/install — post-install choice page. */ export async function handleInstallPage(event: H3Event): Promise { const env = cfEnv(event); + initConfigStore(env); const query = getQuery(event); const rawId = String(query["installation_id"] ?? ""); const installationId = Number(rawId); @@ -171,6 +174,7 @@ export async function handleInstallPage(event: H3Event): Promise /** POST /auth/github/install/bind — provision the chosen binding. */ export async function handleInstallBind(event: H3Event): Promise { const env = cfEnv(event); + initConfigStore(env); const session = await getAdminSession(env.KV, getHeader(event, "cookie")); if (!session) { await sendRedirect(event, "/admin?error=forbidden"); @@ -278,6 +282,7 @@ export async function handleInstallBind(event: H3Event): Promise { /** GET /auth/github/callback — OAuth callback. */ export async function handleOAuthCallback(event: H3Event): Promise { const env = cfEnv(event); + initConfigStore(env); const query = getQuery(event); const code = String(query["code"] ?? ""); const state = String(query["state"] ?? ""); diff --git a/server/lib/webhook.ts b/server/lib/webhook.ts index e51531e..c09610e 100644 --- a/server/lib/webhook.ts +++ b/server/lib/webhook.ts @@ -3,13 +3,13 @@ import { getHeader, readRawBody, setResponseStatus } from "h3"; import type { Env } from "./types"; import { detectProvider } from "./providers"; import { dispatchEvent } from "./core/dispatch"; -import { loadConfig } from "./config"; +import { loadConfig, initConfigStore } from "./config"; import { loadGroups, ensureInstallationGroup } from "./web/groups"; import { getTenantSecret } from "./web/tenants"; import { recordAudit } from "./lib/audit"; import { cfEnv, cfWaitUntil, headersFrom } from "./cf"; import { log } from "./lib/log"; -import { deliveryKey, kvIdempotencyStore } from "./lib/idempotency"; +import { deliveryKey, idempotencyStore } from "./lib/idempotency"; import { newCorrelationId } from "./lib/correlation"; import { enqueueWebhook, type DeliveryMessage } from "./queue/delivery"; @@ -37,6 +37,7 @@ export async function processWebhook( ): Promise { const requestId = newCorrelationId(); let effectiveEnv = env; + initConfigStore(env); const groups = await loadGroups(env.KV); if (tenantId) { if (!groups.some((g) => g.id === tenantId)) { @@ -124,7 +125,7 @@ export async function processWebhook( // Dedup via the idempotency store: a provider/tenant-scoped key means // different accounts may reuse a delivery id without colliding, while // retries of the same delivery never dispatch twice. - const store = kvIdempotencyStore(env.KV); + const store = idempotencyStore(env.DB, env.KV); const key = deliveryKey(provider.id, tenantId, event.deliveryId); if (!(await store.claim(key, 120))) { return { status: 200, body: { ok: true, duplicate: true, requestId } }; diff --git a/server/tasks/storage-prune.ts b/server/tasks/storage-prune.ts new file mode 100644 index 0000000..eb0bc34 --- /dev/null +++ b/server/tasks/storage-prune.ts @@ -0,0 +1,40 @@ +import { cfEnv } from "../lib/cf"; +import { log } from "../lib/lib/log"; + +const DELIVERY_STATE_RETENTION_MS = 7 * 24 * 3600 * 1000; +const MESSAGE_TRACKING_RETENTION_MS = 30 * 24 * 3600 * 1000; + +/** + * Scheduled: purge expired rows from the D1-backed ephemeral stores + * (dedup_keys, delivery_state, message_tracking). These replaced the KV keys + * that relied on per-entry TTL; D1 rows must be removed by this task instead. + */ +export default defineTask({ + meta: { name: "storage:prune", description: "Prune D1 dedup/delivery/message-tracking rows" }, + async run(event) { + try { + const env = cfEnv(event); + const now = Date.now(); + let removed = 0; + const dedup = await env.DB.prepare("DELETE FROM dedup_keys WHERE expires_at < ?") + .bind(now) + .run(); + removed += dedup.meta?.changes ?? 0; + const delivery = await env.DB.prepare("DELETE FROM delivery_state WHERE updated_at < ?") + .bind(now - DELIVERY_STATE_RETENTION_MS) + .run(); + removed += delivery.meta?.changes ?? 0; + const messages = await env.DB.prepare( + "DELETE FROM message_tracking WHERE updated_at < ?", + ) + .bind(now - MESSAGE_TRACKING_RETENTION_MS) + .run(); + removed += messages.meta?.changes ?? 0; + if (removed > 0) log.info({ removed }, "Pruned D1 storage rows"); + return { result: "ok", removed }; + } catch (err) { + log.error({ err }, "Storage prune from cron failed"); + return { result: "error" }; + } + }, +}); \ No newline at end of file diff --git a/tests/config-store.test.ts b/tests/config-store.test.ts new file mode 100644 index 0000000..0ac8604 --- /dev/null +++ b/tests/config-store.test.ts @@ -0,0 +1,174 @@ +import { describe, it, expect } from "bun:test"; +import { d1ConfigStore, type ConfigStore } from "../server/lib/storage/config-store"; +import type { Route, Group } from "../server/lib/types"; + +interface Stmt { + bind: (...args: unknown[]) => Stmt; + all: () => Promise<{ results: Array> }>; +} + +interface FakeDB { + prepare: (sql: string) => Stmt; + batch: (stmts: Stmt[]) => Promise; +} + +function route(id: string, groupId: string, overrides: Partial = {}): Route { + return { + id, + name: `route-${id}`, + enabled: true, + filters: [], + targets: [], + groupId, + stop: false, + fallback: false, + ...overrides, + }; +} + +function group(id: string): Group { + return { id, name: `group-${id}`, adminIds: [] }; +} + +function createDB(): { db: FakeDB & D1Database; routesTable: Route[]; groupsTable: Group[] } { + const routesTable: Route[] = []; + const groupsTable: Group[] = []; + const db: FakeDB = { + prepare(sql: string): Stmt { + let _bound: unknown[] = []; + const stmt: Stmt = { + bind(...args: unknown[]): Stmt { + _bound = args; + return stmt; + }, + async all(): Promise<{ results: Array> }> { + if (sql.includes("FROM d1_routes")) { + return { + results: routesTable.map((r) => ({ + id: r.id, + group_id: r.groupId ?? "", + name: r.name, + enabled: r.enabled ? 1 : 0, + filters: JSON.stringify(r.filters), + targets: JSON.stringify(r.targets), + stop: r.stop ? 1 : 0, + fallback: r.fallback ? 1 : 0, + discord_role_ids: r.discordRoleIds ? JSON.stringify(r.discordRoleIds) : null, + ast: r.ast ? JSON.stringify(r.ast) : null, + })), + }; + } + if (sql.includes("FROM d1_groups")) { + return { + results: groupsTable.map((g) => ({ + id: g.id, + name: g.name, + data: JSON.stringify(g), + version: 1, + })), + }; + } + return { results: [] }; + }, + }; + return stmt; + }, + async batch(stmts: Stmt[]): Promise { + void stmts; + return []; + }, + }; + return { db: db as FakeDB & D1Database, routesTable, groupsTable }; +} + +function createKV(): { kv: KVNamespace; store: Map } { + const store = new Map(); + const kv = { + store, + get: async (key: string, type?: string): Promise => { + const v = store.get(key); + if (v == null) return null; + if (type === "json") return JSON.parse(v) as T; + return v as unknown as T; + }, + put: async (key: string, value: string): Promise => { + store.set(key, value); + }, + delete: async (key: string): Promise => { + store.delete(key); + }, + list: async (): Promise<{ keys: unknown[] }> => ({ keys: [] }), + }; + return { kv: kv as unknown as KVNamespace, store }; +} + +describe("d1ConfigStore", () => { + it("loads empty when D1 and KV are empty", async () => { + const { db } = createDB(); + const { kv } = createKV(); + const store: ConfigStore = d1ConfigStore(db, kv); + expect(await store.loadRoutes()).toEqual([]); + expect(await store.loadGroups()).toEqual([]); + }); + + it("seeds routes from KV into memory when D1 is empty and caches in KV", async () => { + const { db } = createDB(); + const { kv, store } = createKV(); + const existing = [route("r1", "g1")]; + store.set("config:routes", JSON.stringify(existing)); + const cfg: ConfigStore = d1ConfigStore(db, kv); + const loaded = await cfg.loadRoutes(); + expect(loaded).toHaveLength(1); + expect(loaded[0].id).toBe("r1"); + }); + + it("reads routes directly from D1 when populated", async () => { + const { db, routesTable } = createDB(); + const { kv } = createKV(); + routesTable.push(route("r1", "g1")); + const cfg: ConfigStore = d1ConfigStore(db, kv); + const loaded = await cfg.loadRoutes(); + expect(loaded).toHaveLength(1); + expect(loaded[0].id).toBe("r1"); + expect(loaded[0].groupId).toBe("g1"); + }); + + it("round-trips routes through D1 + KV cache on save", async () => { + const { db, routesTable } = createDB(); + const { kv, store } = createKV(); + const cfg: ConfigStore = d1ConfigStore(db, kv); + const routes = [route("r1", "g1"), route("r2", "g2")]; + await cfg.saveRoutes(routes); + expect(routesTable).toHaveLength(0); + const cached = store.get("config:routes"); + expect(cached).toBe(JSON.stringify(routes)); + const reloaded = await cfg.loadRoutes(); + expect(reloaded).toHaveLength(2); + }); + + it("round-trips groups through D1 + KV cache on save", async () => { + const { db } = createDB(); + const { kv, store } = createKV(); + const cfg: ConfigStore = d1ConfigStore(db, kv); + const groups = [group("g1"), group("g2")]; + await cfg.saveGroups(groups); + expect(store.get("config:groups")).toBe(JSON.stringify(groups)); + const reloaded = await cfg.loadGroups(); + expect(reloaded).toHaveLength(2); + }); + + it("invalidateCache forces a reload", async () => { + const { db, routesTable } = createDB(); + const { kv } = createKV(); + const cfg: ConfigStore = d1ConfigStore(db, kv); + routesTable.push(route("r1", "g1")); + const first = await cfg.loadRoutes(); + expect(first).toHaveLength(1); + routesTable.push(route("r2", "g2")); + const cached = await cfg.loadRoutes(); + expect(cached).toHaveLength(1); + cfg.invalidateCache(); + const second = await cfg.loadRoutes(); + expect(second).toHaveLength(2); + }); +}); \ No newline at end of file diff --git a/tests/d1-stores.test.ts b/tests/d1-stores.test.ts new file mode 100644 index 0000000..0939de2 --- /dev/null +++ b/tests/d1-stores.test.ts @@ -0,0 +1,262 @@ +import { describe, it, expect } from "bun:test"; +import { + d1IdempotencyStore, + idempotencyStore, + kvIdempotencyStore, +} from "../server/lib/lib/idempotency"; +import { + d1MessageTracker, + kvMessageTracker, + messageTracker, +} from "../server/lib/lib/message-tracker"; +import { + getDeliveryState, + setDeliveryState, +} from "../server/lib/queue/delivery"; +import { canUseD1 } from "../server/lib/storage/d1"; +import type { Env } from "../server/lib/types"; + +interface Row { + message_id?: string; + status?: string; + hit?: number; +} + +function createMockKV(): KVNamespace { + const store = new Map(); + return { + get: async (key: string) => store.get(key) ?? null, + put: async (key: string, value: string) => { + store.set(key, value); + }, + delete: async (key: string) => { + store.delete(key); + }, + } as unknown as KVNamespace; +} + +function createMockD1(): { + db: D1Database; + dedupKeys: Map; + messageRows: Map; + deliveryRows: Map; +} { + const dedupKeys = new Map(); + const messageRows = new Map(); + const deliveryRows = new Map(); + + const run = ( + sql: string, + args: unknown[], + ): { success: boolean; meta: { changes: number } } => { + if (sql.includes("INSERT INTO dedup_keys")) { + const [key, claimedAt, expiresAt] = args as [string, number, number]; + const existing = dedupKeys.get(key); + if (!existing) { + dedupKeys.set(key, { claimedAt, expiresAt }); + return { success: true, meta: { changes: 1 } }; + } + if (existing.expiresAt < expiresAt) { + dedupKeys.set(key, { claimedAt, expiresAt }); + return { success: true, meta: { changes: 1 } }; + } + return { success: true, meta: { changes: 0 } }; + } + if (sql.includes("INSERT INTO message_tracking")) { + const [eventId, targetId, messageId, updatedAt] = args as [ + string, + string, + string, + number, + ]; + messageRows.set(`${eventId}\u0000${targetId}`, { messageId, updatedAt }); + return { success: true, meta: { changes: 1 } }; + } + if (sql.includes("INSERT INTO delivery_state")) { + const [key, status, updatedAt] = args as [string, string, number]; + deliveryRows.set(key, { status, updatedAt }); + return { success: true, meta: { changes: 1 } }; + } + if (sql.includes("DELETE FROM message_tracking")) { + const [eventId, targetId] = args as [string, string]; + const existed = messageRows.delete(`${eventId}\u0000${targetId}`); + return { success: true, meta: { changes: existed ? 1 : 0 } }; + } + return { success: true, meta: { changes: 0 } }; + }; + + const first = (sql: string, args: unknown[]): Row | null => { + if (sql.includes("SELECT 1 AS hit FROM dedup_keys")) { + const [key, now] = args as [string, number]; + const row = dedupKeys.get(key); + if (row && row.expiresAt > now) return { hit: 1 }; + return null; + } + if (sql.includes("SELECT message_id FROM message_tracking")) { + const [eventId, targetId] = args as [string, string]; + const row = messageRows.get(`${eventId}\u0000${targetId}`); + return row ? { message_id: row.messageId } : null; + } + if (sql.includes("SELECT status FROM delivery_state")) { + const [key] = args as [string]; + const row = deliveryRows.get(key); + return row ? { status: row.status } : null; + } + return null; + }; + + const db = { + prepare: (sql: string): { + bind: (...args: unknown[]) => { + run: () => Promise<{ success: boolean; meta: { changes: number } }>; + all: () => Promise<{ results: unknown[] }>; + first: () => Promise; + }; + } => ({ + bind: (...args: unknown[]): { + run: () => Promise<{ success: boolean; meta: { changes: number } }>; + all: () => Promise<{ results: unknown[] }>; + first: () => Promise; + } => ({ + run: async () => run(sql, args), + all: async () => ({ results: [] }), + first: async () => first(sql, args), + }), + }), + batch: async (): Promise => [], + } as unknown as D1Database; + + return { db, dedupKeys, messageRows, deliveryRows }; +} + +function envWith(db: D1Database, kv: KVNamespace): Env { + return { KV: kv, DB: db } as Env; +} + +describe("d1IdempotencyStore", () => { + it("claims a key exactly once", async () => { + const { db } = createMockD1(); + const store = d1IdempotencyStore(db); + const key = "delivery:github:g1:e1"; + await expect(store.claim(key, 120)).resolves.toBe(true); + await expect(store.claim(key, 120)).resolves.toBe(false); + await expect(store.has(key)).resolves.toBe(true); + }); + + it("treats an expired dedup key as absent", async () => { + const { db, dedupKeys } = createMockD1(); + const store = d1IdempotencyStore(db); + const key = "delivery:gitea:global:e2"; + await store.claim(key, 120); + dedupKeys.set(key, { claimedAt: Date.now(), expiresAt: Date.now() - 1000 }); + await expect(store.has(key)).resolves.toBe(false); + await expect(store.claim(key, 120)).resolves.toBe(true); + }); + + it("does not collide across different keys", async () => { + const { db } = createMockD1(); + const store = d1IdempotencyStore(db); + await store.claim("delivery:github:g1:e1", 120); + await expect(store.claim("delivery:github:g1:e2", 120)).resolves.toBe(true); + await expect(store.has("delivery:github:g1:e2")).resolves.toBe(true); + }); +}); + +describe("d1MessageTracker", () => { + it("round-trips a message id", async () => { + const { db } = createMockD1(); + const tracker = d1MessageTracker(db); + await tracker.set("evt-1", "discord:123", "message-42"); + await expect(tracker.get("evt-1", "discord:123")).resolves.toBe("message-42"); + await expect(tracker.get("evt-1", "discord:999")).resolves.toBeNull(); + }); + + it("updates on re-set and deletes on delete", async () => { + const { db } = createMockD1(); + const tracker = d1MessageTracker(db); + await tracker.set("evt-1", "tg:9", "old"); + await tracker.set("evt-1", "tg:9", "new"); + await expect(tracker.get("evt-1", "tg:9")).resolves.toBe("new"); + await tracker.delete("evt-1", "tg:9"); + await expect(tracker.get("evt-1", "tg:9")).resolves.toBeNull(); + }); +}); + +describe("delivery state backed by D1", () => { + it("round-trips status through the delivery_state table", async () => { + const { db } = createMockD1(); + const env = envWith(db, createMockKV()); + const key = "delivery-state:github:global:d1"; + await setDeliveryState(env, key, "processing"); + await expect(getDeliveryState(env, key)).resolves.toBe("processing"); + await setDeliveryState(env, key, "delivered"); + await expect(getDeliveryState(env, key)).resolves.toBe("delivered"); + }); + + it("returns null when no state exists", async () => { + const { db } = createMockD1(); + const env = envWith(db, createMockKV()); + await expect( + getDeliveryState(env, "delivery-state:github:g9:nope"), + ).resolves.toBeNull(); + }); +}); + +describe("factory fallback decision", () => { + it("canUseD1 is false for a db without batch", () => { + const empty = {} as D1Database; + expect(canUseD1(empty)).toBe(false); + expect(canUseD1(undefined)).toBe(false); + }); + + it("idempotencyStore falls back to KV semantics without batch", async () => { + const db = { + prepare: (): { bind: () => { run: () => Promise<{ success: boolean }>; all: () => Promise<{ results: unknown[] }> } } => ({ + bind: (): { run: () => Promise<{ success: boolean }>; all: () => Promise<{ results: unknown[] }> } => ({ + run: async () => ({ success: true }), + all: async () => ({ results: [] }), + }), + }), + } as unknown as D1Database; + const kv = createMockKV(); + const store = idempotencyStore(db, kv); + await store.claim("delivery:github:g1:e1", 120); + await expect(store.claim("delivery:github:g1:e1", 120)).resolves.toBe(false); + }); + + it("messageTracker falls back to KV semantics without batch", async () => { + const db = {} as D1Database; + const kv = createMockKV(); + const tracker = messageTracker(db, kv); + await tracker.set("evt-1", "discord:1", "m1"); + await expect(tracker.get("evt-1", "discord:1")).resolves.toBe("m1"); + await tracker.delete("evt-1", "discord:1"); + await expect(tracker.get("evt-1", "discord:1")).resolves.toBeNull(); + }); + + it("kv stores still work on their own", async () => { + const kv = createMockKV(); + const idem = kvIdempotencyStore(kv); + await idem.claim("delivery:github:g1:e1", 120); + await expect(idem.has("delivery:github:g1:e1")).resolves.toBe(true); + + const msg = kvMessageTracker(kv); + await msg.set("evt-1", "discord:1", "m1"); + await expect(msg.get("evt-1", "discord:1")).resolves.toBe("m1"); + }); +}); + +describe("delivery state falls back to KV without batch", () => { + it("stores JSON status in KV", async () => { + const db = {} as D1Database; + const kv = createMockKV(); + const env = envWith(db, kv); + const key = "delivery-state:github:global:d2"; + await setDeliveryState(env, key, "delivered"); + const raw = await kv.get(key); + expect(raw).not.toBeNull(); + expect(JSON.parse(raw as string)).toMatchObject({ status: "delivered" }); + await expect(getDeliveryState(env, key)).resolves.toBe("delivered"); + await expect(getDeliveryState(env, "delivery-state:github:global:nope")).resolves.toBeNull(); + }); +}); \ No newline at end of file diff --git a/tests/payload.test.ts b/tests/payload.test.ts new file mode 100644 index 0000000..a74341a --- /dev/null +++ b/tests/payload.test.ts @@ -0,0 +1,68 @@ +import { describe, it, expect } from "bun:test"; +import { r2PayloadStore, type PayloadStore } from "../server/lib/storage/payload"; +import type { Env } from "../server/lib/types"; + +interface FakeBucket { + put: (key: string, value: string) => Promise; + get: (key: string) => Promise<{ text: () => Promise } | null>; + delete: (key: string) => Promise; + objects: Map; +} + +function createBucket(): FakeBucket { + const objects = new Map(); + const bucket: FakeBucket = { + objects, + put: async (key, value) => { + objects.set(key, value); + }, + get: async (key) => { + const v = objects.get(key); + return v ? { text: async () => v } : null; + }, + delete: async (key) => { + objects.delete(key); + }, + }; + return bucket; +} + +function envWith(bucket?: FakeBucket): Env { + return { + GITHUB_WEBHOOK_SECRET: "secret", + KV: {} as KVNamespace, + DB: {} as D1Database, + PAYLOAD: bucket as unknown as R2Bucket, + }; +} + +describe("r2PayloadStore", () => { + it("round-trips a payload through put/get/delete", async () => { + const bucket = createBucket(); + const store: PayloadStore = r2PayloadStore(envWith(bucket)); + const key = await store.put('{"hello":"world"}'); + expect(key.startsWith("webhooks/")).toBe(true); + expect(await store.get(key)).toBe('{"hello":"world"}'); + await store.delete(key); + expect(await store.get(key)).toBeNull(); + }); + + it("generates unique keys per put", async () => { + const store = r2PayloadStore(envWith(createBucket())); + const a = await store.put("a"); + const b = await store.put("b"); + expect(a).not.toBe(b); + }); + + it("returns null for a missing key", async () => { + const store = r2PayloadStore(envWith(createBucket())); + expect(await store.get("webhooks/nope.json")).toBeNull(); + }); + + it("throws when the R2 binding is not configured", async () => { + const store = r2PayloadStore(envWith(undefined)); + await expect(store.put("x")).rejects.toThrow("R2 binding is not configured"); + await expect(store.get("x")).rejects.toThrow("R2 binding is not configured"); + await expect(store.delete("x")).rejects.toThrow("R2 binding is not configured"); + }); +}); \ No newline at end of file diff --git a/tests/send-log-batch.test.ts b/tests/send-log-batch.test.ts new file mode 100644 index 0000000..8104e27 --- /dev/null +++ b/tests/send-log-batch.test.ts @@ -0,0 +1,112 @@ +import { describe, it, expect } from "bun:test"; +import { recordSendBatch } from "../server/lib/lib/send-log-batch"; +import type { SendRecord } from "../server/lib/lib/send-log"; + +interface BoundStmt { + sql: string; + args: unknown[]; +} + +function createMockDB(): { db: D1Database; rows: Array> } { + const rows: Array> = []; + const insertCols = [ + "ts", + "route_id", + "group_id", + "event", + "repo", + "target", + "ok", + "error", + "status", + "message_id", + "delivery_id", + "platform", + "actor", + "action", + "duration_ms", + "error_code", + "attempts", + "detail", + ]; + const db = { + prepare: (sql: string) => ({ + bind: (...args: unknown[]): BoundStmt => ({ sql, args }), + run: async (): Promise<{ success: boolean }> => ({ success: true }), + all: async (): Promise<{ results: Array> }> => ({ + results: rows, + }), + }), + batch: async (stmts: BoundStmt[]): Promise => { + for (const s of stmts) { + if (s.sql.startsWith("INSERT")) { + const row: Record = {}; + insertCols.forEach((col, i) => { + row[col] = s.args[i]; + }); + rows.push(row); + } + } + return []; + }, + } as unknown as D1Database; + return { db, rows }; +} + +function record(overrides: Partial = {}): SendRecord { + return { + ts: 1234, + routeId: "r1", + event: "push", + target: "111", + ok: true, + ...overrides, + }; +} + +describe("recordSendBatch", () => { + it("is a no-op for an empty list", async () => { + const { db } = createMockDB(); + await expect(recordSendBatch(db, [])).resolves.toBeUndefined(); + }); + + it("batches multiple inserts in a single db.batch call", async () => { + const { db, rows } = createMockDB(); + await recordSendBatch(db, [ + record({ routeId: "a", ok: true }), + record({ routeId: "b", ok: false, error: "boom", groupId: "g1" }), + ]); + expect(rows).toHaveLength(2); + expect(rows[0].route_id).toBe("a"); + expect(rows[1].route_id).toBe("b"); + expect(rows[1].group_id).toBe("g1"); + expect(rows[1].ok).toBe(0); + expect(rows[1].error).toBe("boom"); + }); + + it("serializes detail and encodes ok/error", async () => { + const { db, rows } = createMockDB(); + await recordSendBatch(db, [ + record({ detail: { a: 1 }, ok: false, error: "x", status: 500 }), + ]); + expect(rows[0].detail).toBe('{"a":1}'); + expect(rows[0].ok).toBe(0); + expect(rows[0].status).toBe(500); + }); + + it("swallows database errors", async () => { + const db = { + prepare: (): { bind: () => BoundStmt } => ({ + bind: (): BoundStmt => { + throw new Error("nope"); + }, + }), + batch: async (): Promise => { + throw new Error("batch failed"); + }, + } as unknown as D1Database; + await expect( + recordSendBatch(db, [record()]), + ).resolves.toBeUndefined(); + }); +}); \ No newline at end of file diff --git a/wrangler.jsonc b/wrangler.jsonc index 724472b..a493d8f 100644 --- a/wrangler.jsonc +++ b/wrangler.jsonc @@ -25,6 +25,12 @@ "id": "a471e264837d4f51b077e078022f0cf0" }, ], + "r2_buckets": [ + { + "binding": "PAYLOAD", + "bucket_name": "webhooker-payloads", + } + ], "d1_databases": [ { "binding": "DB",