From cfdf37e273c206be6399ca2cb9d9a99e1eeb49b7 Mon Sep 17 00:00:00 2001 From: RhenCloud Date: Sun, 2 Aug 2026 06:50:09 +0800 Subject: [PATCH] feat(discord): make gateway optional with REST send path and send logs Gateway is now an opt-in feature (DISCORD_GATEWAY_ENABLED); when disabled, messages are delivered via the Discord REST API (discord-rest.ts) instead of the Durable Object. Record per-route delivery results (send-log.ts) for the WebUI. Configure assets binding, DO sqlite migration and cron trigger. --- src/discord-rest.ts | 49 ++++++++++++++++++++++++++++ src/discord.ts | 78 ++++++++++++++++++++++++++------------------- src/send-log.ts | 49 ++++++++++++++++++++++++++++ wrangler.jsonc | 64 +++++++++++++++++++++++-------------- 4 files changed, 184 insertions(+), 56 deletions(-) create mode 100644 src/discord-rest.ts create mode 100644 src/send-log.ts diff --git a/src/discord-rest.ts b/src/discord-rest.ts new file mode 100644 index 0000000..d940b91 --- /dev/null +++ b/src/discord-rest.ts @@ -0,0 +1,49 @@ +import { log } from "./log"; + +const DISCORD_API = "https://discord.com/api/v10"; + +export async function sendMessage( + token: string, + channelId: string, + message: unknown, + threadId?: string, +): Promise<{ ok: boolean; error?: string }> { + const url = threadId + ? `${DISCORD_API}/channels/${threadId}/messages` + : `${DISCORD_API}/channels/${channelId}/messages`; + + for (let attempt = 0; attempt < 3; attempt++) { + try { + const res = await fetch(url, { + method: "POST", + headers: { + Authorization: `Bot ${token}`, + "Content-Type": "application/json", + }, + body: JSON.stringify(message), + }); + + if (res.status === 429) { + const rateLimit = (await res.json()) as { retry_after?: number }; + const retryAfter = (rateLimit.retry_after ?? 1) * 1000; + log.warn({ retryAfter, attempt }, "Rate limited"); + await new Promise((r) => setTimeout(r, retryAfter)); + continue; + } + + if (!res.ok) { + const err = await res.text(); + log.error({ status: res.status, err, channelId }, "Discord API error"); + return { ok: false, error: err }; + } + + return { ok: true }; + } catch (err) { + log.error({ err, attempt, channelId }, "Failed to send message"); + if (attempt === 2) return { ok: false, error: String(err) }; + await new Promise((r) => setTimeout(r, 1000 * (attempt + 1))); + } + } + + return { ok: false, error: "Max retries exceeded" }; +} diff --git a/src/discord.ts b/src/discord.ts index 6173ac1..499a47b 100644 --- a/src/discord.ts +++ b/src/discord.ts @@ -2,7 +2,13 @@ import type { Config, FormattedMessage, WebhookEvent, Env } from "./types"; import { formatEvent } from "./formatter"; import { matchRoute } from "./webhook"; import { log } from "./log"; -import { loadTranslations } from "./i18n"; +import { loadTranslations, type Translations } from "./i18n"; +import { sendMessage } from "./discord-rest"; +import { recordSend } from "./send-log"; + +export function isGatewayEnabled(env: Env): boolean { + return env.DISCORD_GATEWAY_ENABLED === "true"; +} async function getGatewayProxy(env: Env): Promise { const id = env.DISCORD_GATEWAY.idFromName("discord-gateway"); @@ -10,6 +16,7 @@ async function getGatewayProxy(env: Env): Promise { } export async function initGateway(env: Env): Promise { + if (!isGatewayEnabled(env)) return; if (!env.DISCORD_TOKEN) return; const stub = await getGatewayProxy(env); await stub.fetch( @@ -22,54 +29,61 @@ export async function initGateway(env: Env): Promise { } export async function dispatchEvent(config: Config, event: WebhookEvent, env: Env): Promise { + const langs = [...new Set(config.routes.map((r) => r.lang ?? "en"))]; + const trMap = new Map(); + await Promise.all( + langs.map(async (lang) => { + trMap.set(lang, await loadTranslations(lang, env.KV)); + }), + ); + for (const route of config.routes) { if (!matchRoute(route, event)) continue; + const target = route.target.threadId + ? `${route.target.channelId}/${route.target.threadId}` + : route.target.channelId; + try { - const tr = await loadTranslations(route.lang ?? "en", env.KV); + const tr = trMap.get(route.lang ?? "en")!; const message = formatEvent(route, event, tr); - await sendWithRetry(route.target.channelId, message, env, route.target.threadId); + await sendToChannel(route.target.channelId, message, env, route.target.threadId); + await recordSend(env.KV, { + ts: Date.now(), + routeId: route.id, + event: event.event, + repo: (event.payload.repository as { full_name?: string } | undefined)?.full_name, + target, + ok: true, + }); } catch (err) { + await recordSend(env.KV, { + ts: Date.now(), + routeId: route.id, + event: event.event, + repo: (event.payload.repository as { full_name?: string } | undefined)?.full_name, + target, + ok: false, + error: err instanceof Error ? err.message : String(err), + }); log.error({ routeId: route.id, err }, "Route failed"); } } } -async function sleep(ms: number): Promise { - return new Promise((resolve) => setTimeout(resolve, ms)); -} - -async function sendWithRetry( - channelId: string, - message: FormattedMessage, - env: Env, - threadId?: string, - maxRetries = 3, -): Promise { - for (let attempt = 0; attempt <= maxRetries; attempt++) { - try { - await sendToChannel(channelId, message, env, threadId); - return; - } catch (err: unknown) { - const error = err as Record; - const isRateLimit = error.code === 50035 || error.status === 429; - if (isRateLimit && attempt < maxRetries) { - const retryAfter = ((error.retry_after as number) ?? (attempt + 1) * 2) as number; - log.warn({ retryAfter, attempt, maxRetries }, "Rate limited, retrying"); - await sleep(retryAfter * 1000); - continue; - } - throw err; - } - } -} - async function sendToChannel( channelId: string, message: FormattedMessage, env: Env, threadId?: string, ): Promise { + const token = env.DISCORD_TOKEN ?? ""; + if (!isGatewayEnabled(env)) { + const result = await sendMessage(token, channelId, message, threadId); + if (!result.ok) throw new Error(result.error ?? "Send failed"); + return; + } + const stub = await getGatewayProxy(env); const res = await stub.fetch( new Request("https://do.internal", { diff --git a/src/send-log.ts b/src/send-log.ts new file mode 100644 index 0000000..4749f11 --- /dev/null +++ b/src/send-log.ts @@ -0,0 +1,49 @@ +import { log } from "./log"; + +export interface SendRecord { + ts: number; + routeId: string; + event: string; + repo?: string; + target: string; + ok: boolean; + error?: string; +} + +const KEY_PREFIX = "logs:send:"; +const RETENTION_TTL = 3600; +const MAX_READ = 200; + +function randomHex(): string { + const bytes = new Uint8Array(8); + crypto.getRandomValues(bytes); + return Array.from(bytes) + .map((b) => b.toString(16).padStart(2, "0")) + .join(""); +} + +export async function recordSend(kv: KVNamespace, record: SendRecord): Promise { + try { + await kv.put(`${KEY_PREFIX}${record.ts}-${randomHex()}`, JSON.stringify(record), { + expirationTtl: RETENTION_TTL, + }); + } catch (err) { + log.warn({ err }, "Failed to record send log"); + } +} + +export async function getSendLog(kv: KVNamespace, limit = 50): Promise { + try { + const list = await kv.list({ prefix: KEY_PREFIX, limit: MAX_READ }); + const records = await Promise.all( + list.keys.map((k) => kv.get(k.name, "json")), + ); + return records + .filter((r): r is SendRecord => r != null) + .sort((a, b) => b.ts - a.ts) + .slice(0, limit); + } catch (err) { + log.warn({ err }, "Failed to load send log"); + return []; + } +} diff --git a/wrangler.jsonc b/wrangler.jsonc index d1be920..3eda9d7 100644 --- a/wrangler.jsonc +++ b/wrangler.jsonc @@ -1,26 +1,42 @@ { - "name": "webhooker", - "main": "src/index.ts", - "compatibility_date": "2025-01-01", - "compatibility_flags": ["nodejs_compat"], - "durable_objects": { - "bindings": [ - { - "name": "DISCORD_GATEWAY", - "class_name": "DiscordGateway", - }, - ], - }, - "migrations": [ - { - "tag": "v1", - "new_classes": ["DiscordGateway"], - }, - ], - "kv_namespaces": [ - { - "binding": "KV", - "id": "placeholder-will-be-replaced", - }, - ], + "name": "webhooker", + "main": "src/index.ts", + "compatibility_date": "2025-01-01", + "compatibility_flags": [], + "vars": { + "DISCORD_GATEWAY_ENABLED": "true" + }, + "assets": { + "directory": "./admin/dist", + "binding": "ASSETS", + "run_worker_first": true, + "not_found_handling": "single-page-application" + }, + "triggers": { + "crons": [ + "*/5 * * * *" + ] + }, + "durable_objects": { + "bindings": [ + { + "name": "DISCORD_GATEWAY", + "class_name": "DiscordGateway", + }, + ], + }, + "migrations": [ + { + "tag": "v1", + "new_sqlite_classes": [ + "DiscordGateway" + ], + }, + ], + "kv_namespaces": [ + { + "binding": "KV", + "id": "53abb6d985a44b80b2b08d510f73928a" + }, + ], }