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.
This commit is contained in:
RhenCloud 2026-08-02 06:50:09 +08:00
parent 4d5c90773d
commit cfdf37e273
No known key found for this signature in database
GPG key ID: A574A617378C4E0B
4 changed files with 184 additions and 56 deletions

49
src/discord-rest.ts Normal file
View file

@ -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" };
}

View file

@ -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<DurableObjectStub> {
const id = env.DISCORD_GATEWAY.idFromName("discord-gateway");
@ -10,6 +16,7 @@ async function getGatewayProxy(env: Env): Promise<DurableObjectStub> {
}
export async function initGateway(env: Env): Promise<void> {
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<void> {
}
export async function dispatchEvent(config: Config, event: WebhookEvent, env: Env): Promise<void> {
const langs = [...new Set(config.routes.map((r) => r.lang ?? "en"))];
const trMap = new Map<string, Translations>();
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<void> {
return new Promise((resolve) => setTimeout(resolve, ms));
}
async function sendWithRetry(
channelId: string,
message: FormattedMessage,
env: Env,
threadId?: string,
maxRetries = 3,
): Promise<void> {
for (let attempt = 0; attempt <= maxRetries; attempt++) {
try {
await sendToChannel(channelId, message, env, threadId);
return;
} catch (err: unknown) {
const error = err as Record<string, unknown>;
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<void> {
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", {

49
src/send-log.ts Normal file
View file

@ -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<void> {
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<SendRecord[]> {
try {
const list = await kv.list({ prefix: KEY_PREFIX, limit: MAX_READ });
const records = await Promise.all(
list.keys.map((k) => kv.get<SendRecord>(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 [];
}
}

View file

@ -2,7 +2,21 @@
"name": "webhooker",
"main": "src/index.ts",
"compatibility_date": "2025-01-01",
"compatibility_flags": ["nodejs_compat"],
"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": [
{
@ -14,13 +28,15 @@
"migrations": [
{
"tag": "v1",
"new_classes": ["DiscordGateway"],
"new_sqlite_classes": [
"DiscordGateway"
],
},
],
"kv_namespaces": [
{
"binding": "KV",
"id": "placeholder-will-be-replaced",
"id": "53abb6d985a44b80b2b08d510f73928a"
},
],
}