From eaec039ad413c3221dbd2e546c42833c3904d694 Mon Sep 17 00:00:00 2001 From: RhenCloud Date: Sat, 15 Aug 2026 15:42:24 +0800 Subject: [PATCH] refactor(core): message tracker abstraction and formatter plugin registry --- AGENTS.md | 5 +- server/lib/core/dispatch.ts | 17 ++-- server/lib/formatters/index.ts | 90 ++------------------ server/lib/formatters/registry.ts | 135 ++++++++++++++++++++++++++++++ server/lib/formatters/types.ts | 16 ++++ server/lib/lib/message-tracker.ts | 24 ++++++ server/lib/types.ts | 16 ++++ tests/formatter-registry.test.ts | 89 ++++++++++++++++++++ tests/message-tracker.test.ts | 59 +++++++++++++ 9 files changed, 361 insertions(+), 90 deletions(-) create mode 100644 server/lib/formatters/registry.ts create mode 100644 server/lib/formatters/types.ts create mode 100644 server/lib/lib/message-tracker.ts create mode 100644 tests/formatter-registry.test.ts create mode 100644 tests/message-tracker.test.ts diff --git a/AGENTS.md b/AGENTS.md index 38b784a..5c35156 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -70,7 +70,9 @@ server/ # Nitro server │ │ └── parse.ts # parse + normalize Gitea payloads to GitHub shape │ └── custom/ # X-WebHooker-Signature (sha256= HMAC) + arbitrary JSON → `custom` events ├── formatters/ # Platform-neutral message formatters (was formatter.ts) - │ ├── index.ts # formatEvent: 28-event switch + custom → NeutralMessage + re-exports + │ ├── index.ts # formatEvent: builds FormatContext → findFormatter → .format (falls back to formatGeneric) + │ ├── types.ts # FormatContext + EventFormatter (formatter plugin interface) + │ ├── registry.ts # eventFormatters[] + findFormatter() — 29-event formatter plugin registry │ ├── colors.ts # GITHUB_COLORS + WORKFLOW_CONCLUSION_EMOJI │ ├── helpers.ts # emojiPrefix, T, buildMessage, commitLink/branchLink/tagLink │ └── *.ts # push, pull-request, issues, comments, workflow, release, create, @@ -108,6 +110,7 @@ server/ # Nitro server ├── i18n.ts # loadTranslations (KV i18n:{lang} overrides), t() with param interpolation ├── idempotency.ts # IdempotencyStore interface + kvIdempotencyStore (delivery dedup via claim/has) + 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) ├── send-log.ts # SendRecord, recordSend/getSendLog/getSendLogById (D1 send_logs) ├── audit.ts # recordAudit/getAuditLog/pruneAuditLogs (D1 audit_logs, best-effort writes) ├── log.ts # JSON console logger (info/warn/error/fatal) diff --git a/server/lib/core/dispatch.ts b/server/lib/core/dispatch.ts index cf782da..e47cec1 100644 --- a/server/lib/core/dispatch.ts +++ b/server/lib/core/dispatch.ts @@ -5,6 +5,7 @@ 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 { loadGroups, groupAcceptsOwners, @@ -35,6 +36,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); // Message language is configured per group (Group.lang), not per route. const langs = [...new Set(loadedGroups.map((g) => g.lang ?? "en"))]; @@ -209,7 +211,8 @@ export async function dispatchEvent( let result: SendResult; if (message.updateKey) { const groupPrefix = route.groupId ? `${route.groupId}:` : ""; - const kvKey = `msg:${groupPrefix}${route.id}:${message.updateKey}:${targetStr}`; + const eventId = `${groupPrefix}${route.id}:${message.updateKey}`; + const kvKey = `msg:${eventId}:${targetStr}`; const lockKey = `msg:lock:${kvKey}`; // Acquire a short-lived lock so concurrent events for the same @@ -220,7 +223,7 @@ export async function dispatchEvent( for (let attempt = 0; attempt < 3; attempt++) { const holder = await env.KV.get(lockKey); if (holder) { - const existing = await env.KV.get(kvKey); + const existing = await tracker.get(eventId, targetStr); if (existing) { result = await driver.edit(message, target, env, existing); if (result.ok || /not modified/i.test(result.error ?? "")) { @@ -242,10 +245,10 @@ export async function dispatchEvent( durationMs: Date.now() - started, errorCode: result.errorCode, }); - if (!ok) await env.KV.delete(kvKey); + if (!ok) await tracker.delete(eventId, targetStr); continue; } - await env.KV.delete(kvKey); + await tracker.delete(eventId, targetStr); break; } await new Promise((r) => setTimeout(r, 50 * (attempt + 1))); @@ -257,7 +260,7 @@ export async function dispatchEvent( } try { - const existingId = await env.KV.get(kvKey); + const existingId = await tracker.get(eventId, targetStr); if (existingId) { result = await driver.edit(message, target, env, existingId); if (result.ok) { @@ -300,11 +303,11 @@ export async function dispatchEvent( }); continue; } - await env.KV.delete(kvKey); + await tracker.delete(eventId, targetStr); } result = await driver.send(message, target, env); if (result.ok && result.messageId) { - await env.KV.put(kvKey, result.messageId, { expirationTtl: 604800 }); + await tracker.set(eventId, targetStr, result.messageId); } } finally { if (locked) await env.KV.delete(lockKey); diff --git a/server/lib/formatters/index.ts b/server/lib/formatters/index.ts index f3063cf..2ab1206 100644 --- a/server/lib/formatters/index.ts +++ b/server/lib/formatters/index.ts @@ -1,27 +1,9 @@ import type { Route, WebhookEvent, NeutralMessage, NeutralAuthor } from "../types"; import type { Translations } from "../lib/i18n"; import { makeT, senderProfileUrl, type T } from "./helpers"; -import { formatPush } from "./push"; -import { formatPullRequest } from "./pull-request"; -import { formatPullRequestReview, formatPullRequestReviewComment } from "./review"; -import { formatIssues } from "./issues"; -import { formatIssueComment } from "./comments"; -import { formatWorkflowRun, formatWorkflowJob } from "./workflow"; -import { formatRelease } from "./release"; -import { formatCreate, formatDelete } from "./create"; -import { formatStar, formatFork } from "./repo"; -import { formatCheckRun, formatCheckSuite, formatStatus } from "./check"; -import { formatCommitComment } from "./commit-comment"; -import { formatDeployment, formatDeploymentStatus } from "./deployment"; -import { formatMember } from "./member"; -import { formatLabel } from "./label"; -import { formatMilestone } from "./milestone"; -import { formatDiscussion, formatDiscussionComment } from "./discussion"; -import { formatRepository } from "./repository"; -import { formatCodeScanningAlert, formatDependabotAlert } from "./security"; import { formatGeneric } from "./generic"; -import { formatPing } from "./ping"; -import { formatCustom } from "./custom"; +import type { FormatContext } from "./types"; +import { findFormatter } from "./registry"; export function formatEvent( route: Route, @@ -44,66 +26,10 @@ export function formatEvent( url: senderUrl ?? (sender ? senderProfileUrl(repoUrl, sender) : undefined), }; - switch (eventType) { - case "push": - return formatPush(payload, repo, author, t, showEmoji); - case "pull_request": - return formatPullRequest(payload, repo, author, t, showEmoji); - case "pull_request_review": - return formatPullRequestReview(payload, repo, author, t, showEmoji); - case "pull_request_review_comment": - return formatPullRequestReviewComment(payload, repo, author, t, showEmoji); - case "issues": - return formatIssues(payload, repo, author, t, showEmoji); - case "issue_comment": - return formatIssueComment(payload, repo, author, t, showEmoji); - case "workflow_run": - return formatWorkflowRun(payload, repo, author, t, showEmoji); - case "workflow_job": - return formatWorkflowJob(payload, repo, author, t, showEmoji); - case "status": - return formatStatus(payload, repo, author, t, showEmoji); - case "deployment": - return formatDeployment(payload, repo, author, t, showEmoji); - case "ping": - return formatPing(payload, repo, author, t, showEmoji); - case "release": - return formatRelease(payload, repo, author, t, showEmoji); - case "create": - return formatCreate(payload, repo, author, t, showEmoji); - case "delete": - return formatDelete(payload, repo, author, t, showEmoji); - case "star": - return formatStar(payload, repo, repoUrl, author, t, showEmoji); - case "fork": - return formatFork(payload, repo, repoUrl, author, t, showEmoji); - case "check_run": - return formatCheckRun(payload, repo, author, t, showEmoji); - case "check_suite": - return formatCheckSuite(payload, repo, author, t, showEmoji); - case "commit_comment": - return formatCommitComment(payload, repo, author, t, showEmoji); - case "deployment_status": - return formatDeploymentStatus(payload, repo, author, t, showEmoji); - case "member": - return formatMember(payload, repo, author, t, showEmoji); - case "label": - return formatLabel(payload, repo, author, t, showEmoji); - case "milestone": - return formatMilestone(payload, repo, author, t, showEmoji); - case "discussion": - return formatDiscussion(payload, repo, author, t, showEmoji); - case "discussion_comment": - return formatDiscussionComment(payload, repo, author, t, showEmoji); - case "repository": - return formatRepository(payload, repo, repoUrl, author, t, showEmoji); - case "code_scanning_alert": - return formatCodeScanningAlert(payload, repo, author, t, showEmoji); - case "dependabot_alert": - return formatDependabotAlert(payload, repo, author, t, showEmoji); - case "custom": - return formatCustom(payload, repo, author, t, showEmoji); - default: - return formatGeneric(eventType, payload, repo, author, t, showEmoji); - } + const ctx: FormatContext = { payload, repo, repoUrl, author, t, showEmoji }; + + return ( + findFormatter(eventType)?.format(ctx) ?? + formatGeneric(eventType, payload, repo, author, t, showEmoji) + ); } diff --git a/server/lib/formatters/registry.ts b/server/lib/formatters/registry.ts new file mode 100644 index 0000000..fe6a5ed --- /dev/null +++ b/server/lib/formatters/registry.ts @@ -0,0 +1,135 @@ +import type { EventFormatter } from "./types"; +import { formatPush } from "./push"; +import { formatPullRequest } from "./pull-request"; +import { formatPullRequestReview, formatPullRequestReviewComment } from "./review"; +import { formatIssues } from "./issues"; +import { formatIssueComment } from "./comments"; +import { formatWorkflowRun, formatWorkflowJob } from "./workflow"; +import { formatRelease } from "./release"; +import { formatCreate, formatDelete } from "./create"; +import { formatStar, formatFork } from "./repo"; +import { formatCheckRun, formatCheckSuite, formatStatus } from "./check"; +import { formatCommitComment } from "./commit-comment"; +import { formatDeployment, formatDeploymentStatus } from "./deployment"; +import { formatMember } from "./member"; +import { formatLabel } from "./label"; +import { formatMilestone } from "./milestone"; +import { formatDiscussion, formatDiscussionComment } from "./discussion"; +import { formatRepository } from "./repository"; +import { formatCodeScanningAlert, formatDependabotAlert } from "./security"; +import { formatPing } from "./ping"; +import { formatCustom } from "./custom"; + +export const eventFormatters: EventFormatter[] = [ + { events: ["push"], format: (c) => formatPush(c.payload, c.repo, c.author, c.t, c.showEmoji) }, + { + events: ["pull_request"], + format: (c) => formatPullRequest(c.payload, c.repo, c.author, c.t, c.showEmoji), + }, + { + events: ["pull_request_review"], + format: (c) => formatPullRequestReview(c.payload, c.repo, c.author, c.t, c.showEmoji), + }, + { + events: ["pull_request_review_comment"], + format: (c) => formatPullRequestReviewComment(c.payload, c.repo, c.author, c.t, c.showEmoji), + }, + { + events: ["issues"], + format: (c) => formatIssues(c.payload, c.repo, c.author, c.t, c.showEmoji), + }, + { + events: ["issue_comment"], + format: (c) => formatIssueComment(c.payload, c.repo, c.author, c.t, c.showEmoji), + }, + { + events: ["workflow_run"], + format: (c) => formatWorkflowRun(c.payload, c.repo, c.author, c.t, c.showEmoji), + }, + { + events: ["workflow_job"], + format: (c) => formatWorkflowJob(c.payload, c.repo, c.author, c.t, c.showEmoji), + }, + { + events: ["status"], + format: (c) => formatStatus(c.payload, c.repo, c.author, c.t, c.showEmoji), + }, + { + events: ["deployment"], + format: (c) => formatDeployment(c.payload, c.repo, c.author, c.t, c.showEmoji), + }, + { events: ["ping"], format: (c) => formatPing(c.payload, c.repo, c.author, c.t, c.showEmoji) }, + { + events: ["release"], + format: (c) => formatRelease(c.payload, c.repo, c.author, c.t, c.showEmoji), + }, + { + events: ["create"], + format: (c) => formatCreate(c.payload, c.repo, c.author, c.t, c.showEmoji), + }, + { + events: ["delete"], + format: (c) => formatDelete(c.payload, c.repo, c.author, c.t, c.showEmoji), + }, + { + events: ["star"], + format: (c) => formatStar(c.payload, c.repo, c.repoUrl, c.author, c.t, c.showEmoji), + }, + { + events: ["fork"], + format: (c) => formatFork(c.payload, c.repo, c.repoUrl, c.author, c.t, c.showEmoji), + }, + { + events: ["check_run"], + format: (c) => formatCheckRun(c.payload, c.repo, c.author, c.t, c.showEmoji), + }, + { + events: ["check_suite"], + format: (c) => formatCheckSuite(c.payload, c.repo, c.author, c.t, c.showEmoji), + }, + { + events: ["commit_comment"], + format: (c) => formatCommitComment(c.payload, c.repo, c.author, c.t, c.showEmoji), + }, + { + events: ["deployment_status"], + format: (c) => formatDeploymentStatus(c.payload, c.repo, c.author, c.t, c.showEmoji), + }, + { + events: ["member"], + format: (c) => formatMember(c.payload, c.repo, c.author, c.t, c.showEmoji), + }, + { events: ["label"], format: (c) => formatLabel(c.payload, c.repo, c.author, c.t, c.showEmoji) }, + { + events: ["milestone"], + format: (c) => formatMilestone(c.payload, c.repo, c.author, c.t, c.showEmoji), + }, + { + events: ["discussion"], + format: (c) => formatDiscussion(c.payload, c.repo, c.author, c.t, c.showEmoji), + }, + { + events: ["discussion_comment"], + format: (c) => formatDiscussionComment(c.payload, c.repo, c.author, c.t, c.showEmoji), + }, + { + events: ["repository"], + format: (c) => formatRepository(c.payload, c.repo, c.repoUrl, c.author, c.t, c.showEmoji), + }, + { + events: ["code_scanning_alert"], + format: (c) => formatCodeScanningAlert(c.payload, c.repo, c.author, c.t, c.showEmoji), + }, + { + events: ["dependabot_alert"], + format: (c) => formatDependabotAlert(c.payload, c.repo, c.author, c.t, c.showEmoji), + }, + { + events: ["custom"], + format: (c) => formatCustom(c.payload, c.repo, c.author, c.t, c.showEmoji), + }, +]; + +export function findFormatter(eventType: string): EventFormatter | undefined { + return eventFormatters.find((f) => f.events.includes(eventType)); +} diff --git a/server/lib/formatters/types.ts b/server/lib/formatters/types.ts new file mode 100644 index 0000000..27d1880 --- /dev/null +++ b/server/lib/formatters/types.ts @@ -0,0 +1,16 @@ +import type { NeutralAuthor, NeutralMessage } from "../types"; +import type { T } from "./helpers"; + +export interface FormatContext { + payload: Record; + repo?: string; + repoUrl?: string; + author: NeutralAuthor; + t: T; + showEmoji: boolean; +} + +export interface EventFormatter { + readonly events: readonly string[]; + format(ctx: FormatContext): NeutralMessage; +} diff --git a/server/lib/lib/message-tracker.ts b/server/lib/lib/message-tracker.ts new file mode 100644 index 0000000..0495db4 --- /dev/null +++ b/server/lib/lib/message-tracker.ts @@ -0,0 +1,24 @@ +export interface MessageTracker { + get(eventId: string, targetId: string): Promise; + set(eventId: string, targetId: string, messageId: string): Promise; + delete(eventId: string, targetId: string): Promise; +} + +const MESSAGE_KEY_TTL_SECONDS = 604800; + +export function kvMessageTracker(kv: KVNamespace): MessageTracker { + const key = (eventId: string, targetId: string): string => `msg:${eventId}:${targetId}`; + return { + async get(eventId: string, targetId: string): Promise { + return kv.get(key(eventId, targetId)); + }, + async set(eventId: string, targetId: string, messageId: string): Promise { + await kv.put(key(eventId, targetId), messageId, { + expirationTtl: MESSAGE_KEY_TTL_SECONDS, + }); + }, + async delete(eventId: string, targetId: string): Promise { + await kv.delete(key(eventId, targetId)); + }, + }; +} diff --git a/server/lib/types.ts b/server/lib/types.ts index 4e41c2b..ea1c644 100644 --- a/server/lib/types.ts +++ b/server/lib/types.ts @@ -169,6 +169,22 @@ export interface WebhookEvent { installationId?: number; } +/** + * A forge-agnostic webhook event after provider verification and + * normalization: the canonical shape every provider's `parse` output conforms + * to, plus the tenant scope (`groupId`) and receipt time carried through the + * delivery pipeline. + */ +export interface NormalizedWebhook { + provider: WebhookProvider; + deliveryId: string; + event: string; + groupId?: string; + payload: Record; + installationId?: number; + receivedAt: number; +} + export interface NeutralAuthor { name: string; iconUrl?: string; diff --git a/tests/formatter-registry.test.ts b/tests/formatter-registry.test.ts new file mode 100644 index 0000000..5c52cf4 --- /dev/null +++ b/tests/formatter-registry.test.ts @@ -0,0 +1,89 @@ +import { describe, it, expect } from "bun:test"; +import { findFormatter } from "../server/lib/formatters/registry"; +import { formatEvent } from "../server/lib/formatters"; +import type { Route, WebhookEvent } from "../server/lib/types"; + +const SUPPORTED_EVENTS = [ + "push", + "pull_request", + "pull_request_review", + "pull_request_review_comment", + "issues", + "issue_comment", + "workflow_run", + "workflow_job", + "status", + "deployment", + "ping", + "release", + "create", + "delete", + "star", + "fork", + "check_run", + "check_suite", + "commit_comment", + "deployment_status", + "member", + "label", + "milestone", + "discussion", + "discussion_comment", + "repository", + "code_scanning_alert", + "dependabot_alert", + "custom", +]; + +function makeRoute(): Route { + return { + id: "r1", + name: "test", + enabled: true, + filters: [], + targets: [{ platform: "discord", channelId: "c1" }], + }; +} + +describe("findFormatter", () => { + it("maps every supported event type to a formatter", () => { + for (const event of SUPPORTED_EVENTS) { + const formatter = findFormatter(event); + expect(formatter, `missing formatter for ${event}`).toBeDefined(); + expect(formatter!.events).toContain(event); + } + }); + + it("returns undefined for an unknown event type", () => { + expect(findFormatter("totally_unknown_event")).toBeUndefined(); + }); +}); + +describe("formatEvent", () => { + it("falls back to the generic formatter for an unknown event", () => { + const route = makeRoute(); + const event: WebhookEvent = { + event: "totally_unknown_event", + payload: { action: "opened" }, + }; + const message = formatEvent(route, event); + expect(message).toBeDefined(); + expect(typeof message.title).toBe("string"); + expect(message.color).toBe(9147550); + }); + + it("routes a known event through its dedicated formatter", () => { + const route = makeRoute(); + const event: WebhookEvent = { + event: "issues", + payload: { + action: "opened", + repository: { full_name: "acme/widget" }, + issue: { number: 7, title: "Add feature" }, + sender: { login: "alice" }, + }, + }; + const message = formatEvent(route, event); + expect(message.title).toContain("acme/widget#7"); + }); +}); diff --git a/tests/message-tracker.test.ts b/tests/message-tracker.test.ts new file mode 100644 index 0000000..1ff5047 --- /dev/null +++ b/tests/message-tracker.test.ts @@ -0,0 +1,59 @@ +import { describe, it, expect } from "bun:test"; +import { kvMessageTracker } from "../server/lib/lib/message-tracker"; + +function createMockKV(): KVNamespace { + const store = new Map(); + const ttl = new Map(); + return { + get: async (key: string) => (store.has(key) ? store.get(key)! : null), + put: async (key: string, value: string, opts?: { expirationTtl?: number }) => { + store.set(key, value); + if (opts?.expirationTtl) ttl.set(key, opts.expirationTtl); + }, + delete: async (key: string) => { + store.delete(key); + }, + list: async () => ({ keys: [], list_complete: true, cacheStatus: null }), + } as unknown as KVNamespace; +} + +describe("kvMessageTracker", () => { + it("stores and retrieves a message id by event and target", async () => { + const kv = createMockKV(); + const tracker = kvMessageTracker(kv); + await tracker.set("route-1:workflow-9", "channel-1", "msg-42"); + expect(await tracker.get("route-1:workflow-9", "channel-1")).toBe("msg-42"); + }); + + it("returns null for an unknown event/target", async () => { + const kv = createMockKV(); + const tracker = kvMessageTracker(kv); + expect(await tracker.get("missing", "channel-1")).toBeNull(); + }); + + it("uses the msg:{eventId}:{targetId} key with a 7-day TTL", async () => { + const kv = createMockKV(); + const keys = new Map(); + const ttls = new Map(); + (kv.put as unknown) = async ( + key: string, + value: string, + opts?: { expirationTtl?: number }, + ): Promise => { + keys.set(key, value); + ttls.set(key, opts?.expirationTtl ?? 0); + }; + const tracker = kvMessageTracker(kv); + await tracker.set("event-1", "target-1", "msg-7"); + expect(keys.has("msg:event-1:target-1")).toBe(true); + expect(ttls.get("msg:event-1:target-1")).toBe(604800); + }); + + it("deletes tracked entries", async () => { + const kv = createMockKV(); + const tracker = kvMessageTracker(kv); + await tracker.set("event-1", "target-1", "msg-7"); + await tracker.delete("event-1", "target-1"); + expect(await tracker.get("event-1", "target-1")).toBeNull(); + }); +});