From b93f4bb3ac03f758cf807d109cdd3ef1702fdd6a Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Tue, 14 Jul 2026 12:55:35 -0700 Subject: [PATCH] refactor: privatize Twitch and SMS internals (#107737) * refactor(twitch): privatize internal plugin surfaces * refactor(sms): privatize internal plugin surfaces * chore(deadcode): refresh unused-export baseline --- extensions/sms/src/accounts.test.ts | 25 +++- extensions/sms/src/channel.test.ts | 7 +- extensions/sms/src/channel.ts | 2 +- extensions/sms/src/config-schema.ts | 2 +- extensions/sms/src/gateway.test.ts | 59 +++++---- extensions/sms/src/gateway.ts | 2 +- extensions/sms/src/twilio.test.ts | 52 +++++--- extensions/sms/src/twilio.ts | 6 +- .../sms/src/webhook-replay-guard.test.ts | 51 +++++++ extensions/sms/src/webhook-replay-guard.ts | 56 ++++++++ extensions/sms/src/webhook.test.ts | 116 ++++++---------- extensions/sms/src/webhook.ts | 75 +---------- extensions/twitch/src/actions.test.ts | 3 +- .../src/client-manager-registry.test.ts | 17 +-- .../twitch/src/client-manager-registry.ts | 22 ---- extensions/twitch/src/config.ts | 2 +- extensions/twitch/src/monitor.test.ts | 124 +++++++++++++++--- extensions/twitch/src/monitor.ts | 8 +- extensions/twitch/src/send.ts | 2 +- extensions/twitch/src/token.test.ts | 12 +- extensions/twitch/src/token.ts | 2 +- scripts/deadcode-exports.baseline.mjs | 15 --- 22 files changed, 361 insertions(+), 299 deletions(-) create mode 100644 extensions/sms/src/webhook-replay-guard.test.ts create mode 100644 extensions/sms/src/webhook-replay-guard.ts diff --git a/extensions/sms/src/accounts.test.ts b/extensions/sms/src/accounts.test.ts index 5a3913837aec..ece1fa3ab615 100644 --- a/extensions/sms/src/accounts.test.ts +++ b/extensions/sms/src/accounts.test.ts @@ -1,7 +1,24 @@ // Sms tests cover accounts plugin behavior. import { afterEach, describe, expect, it } from "vitest"; import { listSmsAccountIds, resolveSmsAccount } from "./accounts.js"; -import { SmsConfigSchema } from "./config-schema.js"; +import { SmsChannelConfigSchema } from "./config-schema.js"; +import type { SmsChannelConfig } from "./types.js"; + +const smsRuntimeConfigSchema = (() => { + const schema = SmsChannelConfigSchema.runtime; + if (!schema) { + throw new Error("expected SMS runtime config schema"); + } + return schema; +})(); + +function parseSmsConfig(value: unknown): SmsChannelConfig { + const parsed = smsRuntimeConfigSchema.safeParse(value); + if (!parsed.success) { + throw new Error(parsed.issues.map((issue) => issue.message).join("; ")); + } + return parsed.data as SmsChannelConfig; +} const ENV_KEYS = [ "TWILIO_ACCOUNT_SID", @@ -94,7 +111,7 @@ describe("SMS account config", () => { }, }; - expect(SmsConfigSchema.parse(cfg.channels.sms).allowFrom).toEqual([1_555_333_4444]); + expect(parseSmsConfig(cfg.channels.sms).allowFrom).toEqual([1_555_333_4444]); expect(resolveSmsAccount(cfg)).toMatchObject({ allowFrom: ["+15553334444"], }); @@ -189,7 +206,7 @@ describe("SMS account config", () => { }); it("coerces numeric allowFrom entries accepted by the config schema", () => { - const parsed = SmsConfigSchema.parse({ + const parsed = parseSmsConfig({ accountSid: "AC123", authToken: "token", fromNumber: "+15550001111", @@ -242,7 +259,7 @@ describe("SMS account config", () => { it("accepts secret references for Twilio auth tokens", () => { expect(() => - SmsConfigSchema.parse({ + parseSmsConfig({ accountSid: "AC123", authToken: { source: "env", provider: "default", id: "TWILIO_AUTH_TOKEN" }, fromNumber: "+15550001111", diff --git a/extensions/sms/src/channel.test.ts b/extensions/sms/src/channel.test.ts index 1cfd4c26076d..ecdb36cfe057 100644 --- a/extensions/sms/src/channel.test.ts +++ b/extensions/sms/src/channel.test.ts @@ -3,7 +3,6 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; type ChannelModule = typeof import("./channel.js"); -let resolveSmsTextChunkLimit: ChannelModule["resolveSmsTextChunkLimit"]; let smsPlugin: ChannelModule["smsPlugin"]; const sendSmsViaTwilio = vi.hoisted(() => @@ -21,7 +20,7 @@ beforeEach(async () => { vi.doMock("./twilio.js", () => ({ sendSmsViaTwilio, })); - ({ resolveSmsTextChunkLimit, smsPlugin } = await import("./channel.js")); + ({ smsPlugin } = await import("./channel.js")); }); afterEach(() => { @@ -71,7 +70,7 @@ describe("smsPlugin outbound", () => { expect(smsPlugin.messaging?.targetPrefixes).toEqual(["twilio-sms"]); expect(smsPlugin.outbound?.chunker?.("alpha beta", 6)).toEqual(["alpha", "beta"]); expect( - resolveSmsTextChunkLimit({ + smsPlugin.outbound?.resolveEffectiveTextChunkLimit?.({ cfg: { channels: { sms: { @@ -85,7 +84,7 @@ describe("smsPlugin outbound", () => { }), ).toBe(42); expect( - resolveSmsTextChunkLimit({ + smsPlugin.outbound?.resolveEffectiveTextChunkLimit?.({ cfg: { channels: { sms: { diff --git a/extensions/sms/src/channel.ts b/extensions/sms/src/channel.ts index 419144691a4d..0380828535c0 100644 --- a/extensions/sms/src/channel.ts +++ b/extensions/sms/src/channel.ts @@ -156,7 +156,7 @@ function createSmsReceipt(params: { }; } -export function resolveSmsTextChunkLimit(params: { +function resolveSmsTextChunkLimit(params: { cfg: OpenClawConfig; accountId?: string | null; fallbackLimit?: number; diff --git a/extensions/sms/src/config-schema.ts b/extensions/sms/src/config-schema.ts index d8bde63aea50..5a527fbfeb63 100644 --- a/extensions/sms/src/config-schema.ts +++ b/extensions/sms/src/config-schema.ts @@ -38,7 +38,7 @@ const SmsAccountConfigSchema = z }); }); -export const SmsConfigSchema = SmsAccountConfigSchema.extend({ +const SmsConfigSchema = SmsAccountConfigSchema.extend({ accounts: z.record(z.string(), SmsAccountConfigSchema.optional()).optional(), defaultAccount: z.string().optional(), }); diff --git a/extensions/sms/src/gateway.test.ts b/extensions/sms/src/gateway.test.ts index b2fddd6cd174..ae74652e04f4 100644 --- a/extensions/sms/src/gateway.test.ts +++ b/extensions/sms/src/gateway.test.ts @@ -1,10 +1,23 @@ // Sms tests cover gateway plugin behavior. import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; -import { registerSmsWebhookRoute } from "./gateway.js"; +import { startSmsGatewayAccount } from "./gateway.js"; import type { SmsChannelRuntime } from "./inbound.js"; import type { ResolvedSmsAccount } from "./types.js"; -const registerPluginHttpRoute = vi.hoisted(() => vi.fn(() => vi.fn())); +const { registeredRoutes, registerPluginHttpRoute, waitUntilAbort } = vi.hoisted(() => { + const routeCleanups: Array<() => void> = []; + return { + registeredRoutes: routeCleanups, + registerPluginHttpRoute: vi.fn(() => vi.fn()), + waitUntilAbort: vi.fn(async (_signal: AbortSignal, onAbort?: () => void) => { + if (onAbort) { + routeCleanups.push(onAbort); + } + }), + }; +}); + +vi.mock("openclaw/plugin-sdk/channel-outbound", () => ({ waitUntilAbort })); vi.mock("openclaw/plugin-sdk/webhook-ingress", () => ({ createFixedWindowRateLimiter: () => ({ @@ -16,8 +29,6 @@ vi.mock("openclaw/plugin-sdk/webhook-ingress", () => ({ registerPluginHttpRoute, })); -const registeredRoutes: Array<() => void> = []; - function createAccount(accountId: string, webhookPath = "/webhooks/sms"): ResolvedSmsAccount { return { accountId, @@ -36,9 +47,10 @@ function createAccount(accountId: string, webhookPath = "/webhooks/sms"): Resolv }; } -describe("registerSmsWebhookRoute", () => { +describe("startSmsGatewayAccount", () => { beforeEach(() => { registerPluginHttpRoute.mockClear(); + waitUntilAbort.mockClear(); }); afterEach(() => { @@ -48,55 +60,58 @@ describe("registerSmsWebhookRoute", () => { registeredRoutes.length = 0; }); - function registerRoute(params: Parameters[0]) { - const unregister = registerSmsWebhookRoute(params); - registeredRoutes.push(unregister); - return unregister; + async function startRoute( + params: Omit[0], "abortSignal">, + ) { + return await startSmsGatewayAccount({ + ...params, + abortSignal: new AbortController().signal, + }); } - it("rejects duplicate webhook paths across SMS accounts", () => { + it("rejects duplicate webhook paths across SMS accounts", async () => { const channelRuntime = {} as SmsChannelRuntime; - registerRoute({ + await startRoute({ cfg: {}, account: createAccount("default"), channelRuntime, }); - expect(() => - registerRoute({ + await expect( + startRoute({ cfg: {}, account: createAccount("support"), channelRuntime, }), - ).toThrow(/already registered by account default/u); + ).rejects.toThrow(/already registered by account default/u); }); - it("rejects duplicate webhook paths after route normalization", () => { + it("rejects duplicate webhook paths after route normalization", async () => { const channelRuntime = {} as SmsChannelRuntime; - registerRoute({ + await startRoute({ cfg: {}, account: createAccount("default", "/webhooks/sms"), channelRuntime, }); - expect(() => - registerRoute({ + await expect( + startRoute({ cfg: {}, account: createAccount("support", "webhooks/sms"), channelRuntime, }), - ).toThrow(/already registered by account default/u); + ).rejects.toThrow(/already registered by account default/u); expect(registerPluginHttpRoute).toHaveBeenCalledTimes(1); }); - it("allows distinct webhook paths across SMS accounts", () => { + it("allows distinct webhook paths across SMS accounts", async () => { const channelRuntime = {} as SmsChannelRuntime; - registerRoute({ + await startRoute({ cfg: {}, account: createAccount("default"), channelRuntime, }); - registerRoute({ + await startRoute({ cfg: {}, account: createAccount("support", "/webhooks/sms/support"), channelRuntime, diff --git a/extensions/sms/src/gateway.ts b/extensions/sms/src/gateway.ts index 80e232d16638..53bee5c57457 100644 --- a/extensions/sms/src/gateway.ts +++ b/extensions/sms/src/gateway.ts @@ -49,7 +49,7 @@ export function collectSmsStartupWarnings(account: ResolvedSmsAccount): string[] return warnings; } -export function registerSmsWebhookRoute(params: { +function registerSmsWebhookRoute(params: { cfg: SmsWebhookHandlerParams["cfg"]; account: ResolvedSmsAccount; channelRuntime: SmsWebhookHandlerParams["channelRuntime"]; diff --git a/extensions/sms/src/twilio.test.ts b/extensions/sms/src/twilio.test.ts index 8c8df5a1ce20..047988bfb2b2 100644 --- a/extensions/sms/src/twilio.test.ts +++ b/extensions/sms/src/twilio.test.ts @@ -1,15 +1,16 @@ // Sms tests cover twilio plugin behavior. +import { createHmac } from "node:crypto"; +import type { IncomingMessage } from "node:http"; +import { Readable } from "node:stream"; import { afterEach, describe, expect, it, vi } from "vitest"; import { buildTwilioInboundMessage, - computeTwilioSignature, listTwilioIncomingPhoneNumbers, listTwilioMessages, - parseTwilioFormBody, + readTwilioWebhookForm, resolveTwilioWebhookSignatureUrl, retrieveTwilioMessagingService, sendSmsViaTwilio, - TwilioSmsApiError, verifyTwilioSignature, } from "./twilio.js"; import type { ResolvedSmsAccount } from "./types.js"; @@ -53,6 +54,26 @@ function readUrlEncodedRequestBody(init: RequestInit | undefined): URLSearchPara throw new Error("Expected Twilio request body to be URL-encoded."); } +function computeTestTwilioSignature(params: { + url: string; + authToken: string; + form: Record; +}): string { + const data = + params.url + + Object.keys(params.form) + .toSorted() + .map((key) => `${key}${params.form[key] ?? ""}`) + .join(""); + return createHmac("sha1", params.authToken).update(data).digest("base64"); +} + +async function readTestTwilioForm(body: string): Promise> { + const req = Readable.from([body]) as IncomingMessage; + req.headers = { "content-length": String(Buffer.byteLength(body)) }; + return await readTwilioWebhookForm(req); +} + function cancelTrackedTextResponse( text: string, init?: ResponseInit, @@ -80,8 +101,8 @@ describe("Twilio SMS helpers", () => { fetchWithSsrFGuardMock.mockReset(); }); - it("parses Twilio form bodies and inbound messages", () => { - const form = parseTwilioFormBody( + it("parses Twilio form bodies and inbound messages", async () => { + const form = await readTestTwilioForm( "From=%2B15551234567&To=%2B15557654321&Body=hello+there&MessageSid=SM123", ); @@ -123,7 +144,7 @@ describe("Twilio SMS helpers", () => { MessageSid: "SM123", To: "+15557654321", }; - const signature = computeTwilioSignature({ + const signature = computeTestTwilioSignature({ url: "https://gateway.example.com/webhooks/sms", authToken: "secret", form, @@ -155,11 +176,11 @@ describe("Twilio SMS helpers", () => { ).toBe(false); }); - it("preserves signed form values before signature verification", () => { - const form = parseTwilioFormBody( + it("preserves signed form values before signature verification", async () => { + const form = await readTestTwilioForm( "From=%2B15551234567&To=%2B15557654321&Body=+hello+&MessageSid=SM123&WaId=", ); - const signature = computeTwilioSignature({ + const signature = computeTestTwilioSignature({ url: "https://gateway.example.com/webhooks/sms", authToken: "secret", form, @@ -467,6 +488,7 @@ describe("Twilio SMS helpers", () => { }), ).rejects.toMatchObject({ name: "TwilioSmsApiError", + message: "Twilio SMS send failed (400): The message From/To pair violates a blacklist rule.", httpStatus: 400, twilioCode: 21610, responseText: JSON.stringify({ @@ -636,18 +658,6 @@ describe("Twilio SMS helpers", () => { expect(release).toHaveBeenCalledTimes(1); }); - it("exposes a typed Twilio SMS API error", () => { - const error = new TwilioSmsApiError( - 429, - JSON.stringify({ code: 20429, message: "Too many requests" }), - ); - - expect(error).toBeInstanceOf(TwilioSmsApiError); - expect(error.message).toBe("Twilio SMS send failed (429): Too many requests"); - expect(error.httpStatus).toBe(429); - expect(error.twilioCode).toBe(20429); - }); - it("requires successful Twilio sends to include a Message SID", async () => { const fetchImpl = vi.fn( async () => new Response(JSON.stringify({ status: "queued" }), { status: 201 }), diff --git a/extensions/sms/src/twilio.ts b/extensions/sms/src/twilio.ts index 4d1da3d4c3e6..d7ed89277907 100644 --- a/extensions/sms/src/twilio.ts +++ b/extensions/sms/src/twilio.ts @@ -159,7 +159,7 @@ export function resolveTwilioWebhookSignatureUrl(params: { return `${signatureBaseUrl}${search}`; } -export class TwilioSmsApiError extends Error { +class TwilioSmsApiError extends Error { readonly httpStatus: number; readonly responseText: string; readonly twilioCode?: number; @@ -175,7 +175,7 @@ export class TwilioSmsApiError extends Error { } } -export function parseTwilioFormBody(body: string): Record { +function parseTwilioFormBody(body: string): Record { const parsed = querystring.parse(body); const out: Record = {}; for (const [key, value] of Object.entries(parsed)) { @@ -184,7 +184,7 @@ export function parseTwilioFormBody(body: string): Record { return out; } -export function computeTwilioSignature(params: { +function computeTwilioSignature(params: { url: string; authToken: string; form: Record; diff --git a/extensions/sms/src/webhook-replay-guard.test.ts b/extensions/sms/src/webhook-replay-guard.test.ts new file mode 100644 index 000000000000..674c5d51650d --- /dev/null +++ b/extensions/sms/src/webhook-replay-guard.test.ts @@ -0,0 +1,51 @@ +import { describe, expect, it } from "vitest"; +import { createSmsWebhookReplayGuard } from "./webhook-replay-guard.js"; + +describe("createSmsWebhookReplayGuard", () => { + it("prunes only the expired insertion prefix without refreshing replays", () => { + let nowMs = 0; + const replayGuard = createSmsWebhookReplayGuard({ + ttlMs: 10, + maxKeys: 2, + now: () => nowMs, + }); + + expect(replayGuard.remember("first")).toEqual({ kind: "accepted" }); + nowMs = 2; + expect(replayGuard.remember("second")).toEqual({ kind: "accepted" }); + nowMs = 5; + expect(replayGuard.remember("first")).toEqual({ kind: "replayed" }); + expect(replayGuard.remember("overflow")).toEqual({ + kind: "saturated", + retryAfterMs: 5, + }); + + nowMs = 10; + expect(replayGuard.remember("overflow")).toEqual({ kind: "accepted" }); + expect(replayGuard.remember("second")).toEqual({ kind: "replayed" }); + }); + + it("keeps live replay keys and fails closed until capacity expires", () => { + let nowMs = 1_000; + const replayGuard = createSmsWebhookReplayGuard({ + ttlMs: 10_000, + maxKeys: 2, + now: () => nowMs, + }); + + expect(replayGuard.remember("first")).toEqual({ kind: "accepted" }); + expect(replayGuard.remember("second")).toEqual({ kind: "accepted" }); + expect(replayGuard.remember("overflow")).toEqual({ + kind: "saturated", + retryAfterMs: 10_000, + }); + expect(replayGuard.remember("overflow")).toEqual({ + kind: "saturated", + retryAfterMs: 10_000, + }); + expect(replayGuard.remember("first")).toEqual({ kind: "replayed" }); + + nowMs += 10_000; + expect(replayGuard.remember("overflow")).toEqual({ kind: "accepted" }); + }); +}); diff --git a/extensions/sms/src/webhook-replay-guard.ts b/extensions/sms/src/webhook-replay-guard.ts new file mode 100644 index 000000000000..b68b7b736581 --- /dev/null +++ b/extensions/sms/src/webhook-replay-guard.ts @@ -0,0 +1,56 @@ +import { performance } from "node:perf_hooks"; + +const REPLAY_CACHE_TTL_MS = 10 * 60_000; +const REPLAY_CACHE_MAX_KEYS = 10_000; + +type ReplayCacheDecision = + | { kind: "accepted" } + | { kind: "replayed" } + | { kind: "saturated"; retryAfterMs: number }; + +export type SmsWebhookReplayGuard = { + remember: (messageSid: string) => ReplayCacheDecision; +}; + +export function createSmsWebhookReplayGuard( + options: { + ttlMs?: number; + maxKeys?: number; + now?: () => number; + } = {}, +): SmsWebhookReplayGuard { + const ttlMs = options.ttlMs ?? REPLAY_CACHE_TTL_MS; + const maxKeys = options.maxKeys ?? REPLAY_CACHE_MAX_KEYS; + const now = options.now ?? (() => performance.now()); + const entries = new Map(); + + const pruneExpired = (nowMs: number) => { + // Fixed TTLs on a monotonic clock expire in insertion order, so only inspect + // the expired prefix. Full live caches stay O(1) instead of rescanning 10k keys. + for (const [key, expiresAt] of entries) { + if (expiresAt > nowMs) { + break; + } + entries.delete(key); + } + }; + + return { + remember: (messageSid) => { + const nowMs = now(); + pruneExpired(nowMs); + if (entries.has(messageSid)) { + return { kind: "replayed" }; + } + if (entries.size >= maxKeys) { + const oldestExpiresAt = entries.values().next().value ?? nowMs; + return { + kind: "saturated", + retryAfterMs: Math.max(0, oldestExpiresAt - nowMs), + }; + } + entries.set(messageSid, nowMs + ttlMs); + return { kind: "accepted" }; + }, + }; +} diff --git a/extensions/sms/src/webhook.test.ts b/extensions/sms/src/webhook.test.ts index 5e15859b26ef..c57eadea2dcd 100644 --- a/extensions/sms/src/webhook.test.ts +++ b/extensions/sms/src/webhook.test.ts @@ -1,13 +1,11 @@ // Sms tests cover webhook plugin behavior. +import { createHmac } from "node:crypto"; import type { IncomingMessage, ServerResponse } from "node:http"; import { Readable } from "node:stream"; import { beforeEach, describe, expect, it, vi } from "vitest"; import type { SmsChannelRuntime } from "./inbound.js"; -import { computeTwilioSignature, parseTwilioFormBody } from "./twilio.js"; import type { ResolvedSmsAccount } from "./types.js"; -import { createSmsWebhookHandler, testing } from "./webhook.js"; - -const { createSmsWebhookReplayGuard, resetSmsWebhookReplayGuardsForTest } = testing; +import { createSmsWebhookHandler } from "./webhook.js"; const dispatchSmsInboundEvent = vi.hoisted(() => vi.fn(async () => undefined)); @@ -15,9 +13,30 @@ vi.mock("./inbound.js", () => ({ dispatchSmsInboundEvent, })); +let testAccountSequence = 0; +let activeAccountId = "test-0"; + +function parseTestTwilioForm(body: string): Record { + return Object.fromEntries(new URLSearchParams(body)); +} + +function computeTestTwilioSignature(params: { + url: string; + authToken: string; + form: Record; +}): string { + const data = + params.url + + Object.keys(params.form) + .toSorted() + .map((key) => `${key}${params.form[key] ?? ""}`) + .join(""); + return createHmac("sha1", params.authToken).update(data).digest("base64"); +} + function createAccount(overrides: Partial = {}): ResolvedSmsAccount { return { - accountId: "default", + accountId: activeAccountId, enabled: true, accountSid: "AC123", authToken: "secret", @@ -45,10 +64,10 @@ function createSignedBody(params?: { `AccountSid=${encodeURIComponent(account.accountSid)}&From=%2B15551234567&To=%2B15557654321&Body=hello&MessageSid=${encodeURIComponent(params?.messageSid ?? "SM123")}`; return { body, - signature: computeTwilioSignature({ + signature: computeTestTwilioSignature({ url: account.publicWebhookUrl, authToken: account.authToken, - form: parseTwilioFormBody(body), + form: parseTestTwilioForm(body), }), }; } @@ -60,7 +79,11 @@ function createRequest( ): IncomingMessage { const req = Readable.from([body]) as IncomingMessage; req.method = "POST"; - req.headers = { "x-twilio-signature": signature, ...options?.headers }; + req.headers = { + "content-length": String(Buffer.byteLength(body)), + "x-twilio-signature": signature, + ...options?.headers, + }; Object.defineProperty(req, "socket", { value: { remoteAddress: options?.remoteAddress ?? "127.0.0.1" }, }); @@ -98,10 +121,10 @@ function createSignedSmsPayload( }).toString(); return { body, - signature: computeTwilioSignature({ + signature: computeTestTwilioSignature({ url: "https://gateway.example.com/webhooks/sms", authToken: "secret", - form: parseTwilioFormBody(body), + form: parseTestTwilioForm(body), }), }; } @@ -113,7 +136,7 @@ function createMessageSid(index: number): string { describe("createSmsWebhookHandler", () => { beforeEach(() => { dispatchSmsInboundEvent.mockClear(); - resetSmsWebhookReplayGuardsForTest(); + activeAccountId = `test-${++testAccountSequence}`; }); it("validates a fragmentless signature and preserves dedupe across handler reloads", async () => { @@ -155,7 +178,7 @@ describe("createSmsWebhookHandler", () => { channelRuntime: {} as SmsChannelRuntime, }); - expect(parseTwilioFormBody(body).From).toBe("RcS:+1 (555) 123-4567"); + expect(parseTestTwilioForm(body).From).toBe("RcS:+1 (555) 123-4567"); const res = createResponse(); await handler(createRequest(body, signature), res); @@ -174,77 +197,12 @@ describe("createSmsWebhookHandler", () => { ); }); - it("prunes only the expired insertion prefix without refreshing replays", () => { - let nowMs = 0; - const replayGuard = createSmsWebhookReplayGuard({ - ttlMs: 10, - maxKeys: 2, - now: () => nowMs, - }); - const first = createMessageSid(2); - const second = createMessageSid(3); - const overflow = createMessageSid(4); - - expect(replayGuard.remember(first)).toEqual({ kind: "accepted" }); - nowMs = 2; - expect(replayGuard.remember(second)).toEqual({ kind: "accepted" }); - nowMs = 5; - expect(replayGuard.remember(first)).toEqual({ kind: "replayed" }); - expect(replayGuard.remember(overflow)).toEqual({ kind: "saturated", retryAfterMs: 5 }); - - nowMs = 10; - expect(replayGuard.remember(overflow)).toEqual({ kind: "accepted" }); - expect(replayGuard.remember(second)).toEqual({ kind: "replayed" }); - }); - - it("keeps live replay keys and fails closed until capacity expires", async () => { - let nowMs = 1_000; - const webhookReplayGuard = createSmsWebhookReplayGuard({ - ttlMs: 10_000, - maxKeys: 2, - now: () => nowMs, - }); - const handler = createSmsWebhookHandler( - { - cfg: {}, - account: createAccount(), - channelRuntime: {} as SmsChannelRuntime, - }, - webhookReplayGuard, - ); - const first = createSignedSmsPayload(createMessageSid(5)); - const second = createSignedSmsPayload(createMessageSid(6)); - const overflow = createSignedSmsPayload(createMessageSid(7)); - - await handler(createRequest(first.body, first.signature), createResponse()); - await handler(createRequest(second.body, second.signature), createResponse()); - const overflowRes = createResponse(); - await handler(createRequest(overflow.body, overflow.signature), overflowRes); - const repeatedOverflowRes = createResponse(); - await handler(createRequest(overflow.body, overflow.signature), repeatedOverflowRes); - const firstReplayRes = createResponse(); - await handler(createRequest(first.body, first.signature), firstReplayRes); - - expect(overflowRes.statusCode).toBe(429); - expect(repeatedOverflowRes.statusCode).toBe(429); - expect(overflowRes.setHeaderMock).toHaveBeenCalledWith("Retry-After", "10"); - expect(firstReplayRes.statusCode).toBe(200); - expect(dispatchSmsInboundEvent).toHaveBeenCalledTimes(2); - - nowMs += 10_000; - const afterExpiryRes = createResponse(); - await handler(createRequest(overflow.body, overflow.signature), afterExpiryRes); - - expect(afterExpiryRes.statusCode).toBe(200); - expect(dispatchSmsInboundEvent).toHaveBeenCalledTimes(3); - }); - it("rejects signed webhooks for a different Twilio account", async () => { const body = `AccountSid=AC-other&From=%2B15551234567&To=%2B15557654321&Body=hello&SmsMessageSid=${createMessageSid(8)}`; - const signature = computeTwilioSignature({ + const signature = computeTestTwilioSignature({ url: "https://gateway.example.com/webhooks/sms", authToken: "secret", - form: parseTwilioFormBody(body), + form: parseTestTwilioForm(body), }); const handler = createSmsWebhookHandler({ cfg: {}, diff --git a/extensions/sms/src/webhook.ts b/extensions/sms/src/webhook.ts index f62f215ee795..f9cb790fc316 100644 --- a/extensions/sms/src/webhook.ts +++ b/extensions/sms/src/webhook.ts @@ -1,6 +1,5 @@ // Sms plugin module implements webhook behavior. import type { IncomingMessage, ServerResponse } from "node:http"; -import { performance } from "node:perf_hooks"; import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts"; import { createFixedWindowRateLimiter, @@ -15,6 +14,7 @@ import { verifyTwilioSignature, } from "./twilio.js"; import type { ResolvedSmsAccount } from "./types.js"; +import { createSmsWebhookReplayGuard, type SmsWebhookReplayGuard } from "./webhook-replay-guard.js"; const INVALID_REQUEST_MAX_REQUESTS = 300; const CALLBACK_DISPATCH_MAX_REQUESTS = 30; @@ -32,63 +32,8 @@ const callbackDispatchRateLimiter = createFixedWindowRateLimiter({ windowMs: 60_000, maxTrackedKeys: 5_000, }); -const REPLAY_CACHE_TTL_MS = 10 * 60_000; -const REPLAY_CACHE_MAX_KEYS = 10_000; - -type ReplayCacheDecision = - | { kind: "accepted" } - | { kind: "replayed" } - | { kind: "saturated"; retryAfterMs: number }; - -type SmsWebhookReplayGuard = { - remember: (messageSid: string) => ReplayCacheDecision; -}; - const replayGuardsByAccount = new Map(); -function createSmsWebhookReplayGuard( - options: { - ttlMs?: number; - maxKeys?: number; - now?: () => number; - } = {}, -): SmsWebhookReplayGuard { - const ttlMs = options.ttlMs ?? REPLAY_CACHE_TTL_MS; - const maxKeys = options.maxKeys ?? REPLAY_CACHE_MAX_KEYS; - const now = options.now ?? (() => performance.now()); - const entries = new Map(); - - const pruneExpired = (nowMs: number) => { - // Fixed TTLs on a monotonic clock expire in insertion order, so only inspect - // the expired prefix. Full live caches stay O(1) instead of rescanning 10k keys. - for (const [key, expiresAt] of entries) { - if (expiresAt > nowMs) { - break; - } - entries.delete(key); - } - }; - - return { - remember: (messageSid) => { - const nowMs = now(); - pruneExpired(nowMs); - if (entries.has(messageSid)) { - return { kind: "replayed" }; - } - if (entries.size >= maxKeys) { - const oldestExpiresAt = entries.values().next().value ?? nowMs; - return { - kind: "saturated", - retryAfterMs: Math.max(0, oldestExpiresAt - nowMs), - }; - } - entries.set(messageSid, nowMs + ttlMs); - return { kind: "accepted" }; - }, - }; -} - function resolveSmsWebhookReplayGuard(account: ResolvedSmsAccount): SmsWebhookReplayGuard { // Config reloads replace route handlers. Keep the guard with the Twilio account // identity so retries cannot cross that lifecycle boundary or block sibling accounts. @@ -102,12 +47,6 @@ function resolveSmsWebhookReplayGuard(account: ResolvedSmsAccount): SmsWebhookRe return created; } -function resetSmsWebhookReplayGuardsForTest(): void { - replayGuardsByAccount.clear(); - invalidRequestRateLimiter.clear(); - callbackDispatchRateLimiter.clear(); -} - type SmsWebhookLog = { info?: (message: string) => void; warn?: (message: string) => void; @@ -154,17 +93,9 @@ function rejectInvalidRequestRateLimit(params: { return true; } -/** Test-only hooks for webhook state that is otherwise private. */ -export const testing = { - createSmsWebhookReplayGuard, - resetSmsWebhookReplayGuardsForTest, -}; - // Each account route owns its guard so one saturated account cannot block sibling accounts. -export function createSmsWebhookHandler( - params: SmsWebhookHandlerParams, - webhookReplayGuard: SmsWebhookReplayGuard = resolveSmsWebhookReplayGuard(params.account), -) { +export function createSmsWebhookHandler(params: SmsWebhookHandlerParams) { + const webhookReplayGuard = resolveSmsWebhookReplayGuard(params.account); return async (req: IncomingMessage, res: ServerResponse) => { if (req.method !== "POST") { respondTwiml(res, 405, "Method not allowed"); diff --git a/extensions/twitch/src/actions.test.ts b/extensions/twitch/src/actions.test.ts index 421077bfa185..0e221e2c2543 100644 --- a/extensions/twitch/src/actions.test.ts +++ b/extensions/twitch/src/actions.test.ts @@ -1,10 +1,11 @@ // Twitch tests cover actions plugin behavior. import { describe, expect, it, vi, beforeEach } from "vitest"; import { twitchMessageActions } from "./actions.js"; -import type { ResolvedTwitchAccountContext } from "./config.js"; import { resolveTwitchAccountContext } from "./config.js"; import { twitchOutbound } from "./outbound.js"; +type ResolvedTwitchAccountContext = ReturnType; + vi.mock("./config.js", () => ({ DEFAULT_ACCOUNT_ID: "default", resolveTwitchAccountContext: vi.fn(), diff --git a/extensions/twitch/src/client-manager-registry.test.ts b/extensions/twitch/src/client-manager-registry.test.ts index 721267ea7715..fd6a95e75a5a 100644 --- a/extensions/twitch/src/client-manager-registry.test.ts +++ b/extensions/twitch/src/client-manager-registry.test.ts @@ -1,7 +1,6 @@ // Twitch tests cover client manager registry plugin behavior. import { afterEach, describe, expect, it, vi } from "vitest"; import { - clearRegistryForTest, getClientManager, getOrCreateClientManager, removeClientManager, @@ -19,21 +18,7 @@ function makeLogger(): ChannelLogSink { describe("client manager registry", () => { afterEach(async () => { - await clearRegistryForTest(); - }); - - it("clears cached managers for hot module test isolation", async () => { - const firstManager = getOrCreateClientManager("default", makeLogger()); - const disconnectAll = vi.spyOn(firstManager, "disconnectAll"); - - expect(getClientManager("default")).toBe(firstManager); - expect(getOrCreateClientManager("default", makeLogger())).toBe(firstManager); - - await clearRegistryForTest(); - - expect(disconnectAll).toHaveBeenCalledOnce(); - expect(getClientManager("default")).toBeUndefined(); - expect(getOrCreateClientManager("default", makeLogger())).not.toBe(firstManager); + await removeClientManager("default"); }); it("removes cached managers even when disconnectAll rejects", async () => { diff --git a/extensions/twitch/src/client-manager-registry.ts b/extensions/twitch/src/client-manager-registry.ts index 77a3180abf8c..26424354da15 100644 --- a/extensions/twitch/src/client-manager-registry.ts +++ b/extensions/twitch/src/client-manager-registry.ts @@ -85,25 +85,3 @@ export async function removeClientManager(accountId: string): Promise { entry.logger.info(`Unregistered client manager for account: ${accountId}`); } } - -/** - * Test-only: clear the module-level registry of all client manager entries. - * - * Mirrors the `clearForTest` escape hatch on `TwitchClientManager`. Without - * this, the module-level `registry` Map survives across tests when vitest - * is run with `--isolate=false` (or any harness that does not tear the - * module graph down between cases), and a stale entry from one test will - * shadow `getOrCreateClientManager` calls in subsequent tests, silently - * handing back another test's mocked logger/manager. See #83887. - * - * Production code MUST NOT call this. It disconnects cached managers before - * clearing the registry so tests do not leave handlers or clients behind. - */ -export async function clearRegistryForTest(): Promise { - const entries = [...registry.values()]; - try { - await Promise.all(entries.map((entry) => entry.manager.disconnectAll())); - } finally { - registry.clear(); - } -} diff --git a/extensions/twitch/src/config.ts b/extensions/twitch/src/config.ts index d160123d73d5..1bc1a669ce83 100644 --- a/extensions/twitch/src/config.ts +++ b/extensions/twitch/src/config.ts @@ -14,7 +14,7 @@ import { isAccountConfigured } from "./utils/twitch.js"; */ export const DEFAULT_ACCOUNT_ID = "default"; -export type ResolvedTwitchAccountContext = { +type ResolvedTwitchAccountContext = { accountId: string; account: TwitchAccountConfig | null; tokenResolution: TwitchTokenResolution; diff --git a/extensions/twitch/src/monitor.test.ts b/extensions/twitch/src/monitor.test.ts index 5a4580f6daa7..0fa9de40bdc5 100644 --- a/extensions/twitch/src/monitor.test.ts +++ b/extensions/twitch/src/monitor.test.ts @@ -1,42 +1,130 @@ import { beforeEach, describe, expect, it, vi } from "vitest"; import { BASE_TWITCH_TEST_ACCOUNT } from "./test-fixtures.js"; +import type { TwitchChatMessage } from "./types.js"; const mocks = vi.hoisted(() => ({ + checkAccess: vi.fn(async () => ({ allowed: true })), + getClient: vi.fn(async () => ({})), + getRuntime: vi.fn(), + onMessage: vi.fn(), + runInbound: vi.fn(), sendMessage: vi.fn(), + unregister: vi.fn(), +})); + +vi.mock("./access-control.js", () => ({ + checkTwitchAccessControl: mocks.checkAccess, })); vi.mock("./client-manager-registry.js", () => ({ - getOrCreateClientManager: () => ({ sendMessage: mocks.sendMessage }), + getOrCreateClientManager: () => ({ + getClient: mocks.getClient, + onMessage: mocks.onMessage, + sendMessage: mocks.sendMessage, + }), })); -import { testing } from "./monitor.js"; +vi.mock("./runtime.js", () => ({ + getTwitchRuntime: mocks.getRuntime, +})); -describe("deliverTwitchReply", () => { +import { monitorTwitchProvider } from "./monitor.js"; + +type InboundRunInput = { + raw: TwitchChatMessage; + adapter: { + ingest: (message: TwitchChatMessage) => unknown; + resolveTurn: (input: unknown) => Promise<{ + delivery: { + deliver: (payload: { text: string }) => Promise; + }; + }>; + }; +}; + +describe("monitorTwitchProvider", () => { beforeEach(() => { - mocks.sendMessage.mockReset(); + vi.clearAllMocks(); + mocks.getClient.mockResolvedValue({}); mocks.sendMessage.mockResolvedValue({ ok: true, messageId: "message-id" }); + mocks.runInbound.mockImplementation(async (input: InboundRunInput) => { + const ingested = input.adapter.ingest(input.raw); + const turn = await input.adapter.resolveTurn(ingested); + await turn.delivery.deliver({ text: "**Hello** Twitch" }); + }); + mocks.getRuntime.mockReturnValue({ + logging: { + getChildLogger: () => ({ + info: vi.fn(), + warn: vi.fn(), + error: vi.fn(), + debug: vi.fn(), + }), + shouldLogVerbose: () => false, + }, + channel: { + inbound: { + run: mocks.runInbound, + buildContext: vi.fn(() => ({})), + }, + routing: { + resolveAgentRoute: vi.fn(() => ({ + agentId: "main", + accountId: "default", + sessionKey: "agent:main:twitch:group:testchannel", + })), + }, + reply: { + formatAgentEnvelope: vi.fn(({ body }: { body: string }) => body), + resolveEnvelopeFormatOptions: vi.fn(() => ({})), + dispatchReplyWithBufferedBlockDispatcher: vi.fn(), + }, + session: { + resolveStorePath: vi.fn(() => "/tmp/sessions.json"), + recordInboundSession: vi.fn(), + }, + text: { + resolveMarkdownTableMode: vi.fn(() => "off"), + }, + }, + }); }); - it("routes fallback replies through the UTF-16-safe transport sender", async () => { + it("delivers fallback replies through the monitor boundary", async () => { + let onMessage: ((message: TwitchChatMessage) => void) | undefined; + mocks.onMessage.mockImplementation( + (_account: unknown, handler: (message: TwitchChatMessage) => void) => { + onMessage = handler; + return mocks.unregister; + }, + ); const account = { ...BASE_TWITCH_TEST_ACCOUNT, accessToken: "oauth:test-token" }; - - const result = await testing.deliverTwitchReply({ - payload: { text: "**Hello** Twitch" }, - channel: "testchannel", + const monitor = await monitorTwitchProvider({ account, accountId: "default", config: {}, - tableMode: "off", runtime: {}, + abortSignal: new AbortController().signal, }); - expect(result).toEqual({ visibleReplySent: true }); - expect(mocks.sendMessage).toHaveBeenCalledWith( - account, - "testchannel", - "Hello Twitch", - {}, - "default", - ); + onMessage?.({ + username: "viewer", + userId: "viewer-1", + message: "hello bot", + channel: "testchannel", + }); + + await vi.waitFor(() => { + expect(mocks.sendMessage).toHaveBeenCalledWith( + account, + "testchannel", + "Hello Twitch", + {}, + "default", + ); + }); + + monitor.stop(); + expect(mocks.unregister).toHaveBeenCalledOnce(); }); }); diff --git a/extensions/twitch/src/monitor.ts b/extensions/twitch/src/monitor.ts index 9f12314d3bb0..edad62524474 100644 --- a/extensions/twitch/src/monitor.ts +++ b/extensions/twitch/src/monitor.ts @@ -15,12 +15,12 @@ import { getTwitchRuntime } from "./runtime.js"; import type { TwitchAccountConfig, TwitchChatMessage } from "./types.js"; import { stripMarkdownForTwitch } from "./utils/markdown.js"; -export type TwitchRuntimeEnv = { +type TwitchRuntimeEnv = { log?: (message: string) => void; error?: (message: string) => void; }; -export type TwitchMonitorOptions = { +type TwitchMonitorOptions = { account: TwitchAccountConfig; accountId: string; config: unknown; // OpenClawConfig @@ -29,7 +29,7 @@ export type TwitchMonitorOptions = { statusSink?: (patch: { lastInboundAt?: number; lastOutboundAt?: number }) => void; }; -export type TwitchMonitorResult = { +type TwitchMonitorResult = { stop: () => void; }; @@ -302,5 +302,3 @@ export async function monitorTwitchProvider( return { stop }; } - -export const testing = { deliverTwitchReply }; diff --git a/extensions/twitch/src/send.ts b/extensions/twitch/src/send.ts index 0e1681e0d710..d1a6526d795e 100644 --- a/extensions/twitch/src/send.ts +++ b/extensions/twitch/src/send.ts @@ -19,7 +19,7 @@ import { generateMessageId, normalizeTwitchChannel } from "./utils/twitch.js"; /** * Result from sending a message to Twitch. */ -export interface SendMessageResult { +interface SendMessageResult { /** Whether the send was successful */ ok: boolean; /** The message ID (generated for tracking) */ diff --git a/extensions/twitch/src/token.test.ts b/extensions/twitch/src/token.test.ts index 79e71bf345cd..12d6d4864e9f 100644 --- a/extensions/twitch/src/token.test.ts +++ b/extensions/twitch/src/token.test.ts @@ -10,7 +10,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import type { OpenClawConfig } from "../api.js"; -import { resolveTwitchToken, type TwitchTokenSource } from "./token.js"; +import { resolveTwitchToken } from "./token.js"; describe("token", () => { const originalAccessToken = process.env.OPENCLAW_TWITCH_ACCESS_TOKEN; @@ -185,14 +185,4 @@ describe("token", () => { expect(result.source).toBe("none"); }); }); - - describe("TwitchTokenSource type", () => { - it("should have correct values", () => { - const sources: TwitchTokenSource[] = ["env", "config", "none"]; - - expect(sources).toContain("env"); - expect(sources).toContain("config"); - expect(sources).toContain("none"); - }); - }); }); diff --git a/extensions/twitch/src/token.ts b/extensions/twitch/src/token.ts index e254f530ef73..67593e1048ce 100644 --- a/extensions/twitch/src/token.ts +++ b/extensions/twitch/src/token.ts @@ -16,7 +16,7 @@ import { } from "openclaw/plugin-sdk/account-resolution"; import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts"; -export type TwitchTokenSource = "env" | "config" | "none"; +type TwitchTokenSource = "env" | "config" | "none"; export type TwitchTokenResolution = { token: string; diff --git a/scripts/deadcode-exports.baseline.mjs b/scripts/deadcode-exports.baseline.mjs index 9ecaec310181..cbede095d2dc 100644 --- a/scripts/deadcode-exports.baseline.mjs +++ b/scripts/deadcode-exports.baseline.mjs @@ -107,13 +107,6 @@ export const KNIP_UNUSED_EXPORT_BASELINE = [ "extensions/qa-matrix/src/substrate/e2ee-client.ts: testing", "extensions/qa-matrix/src/substrate/harness.runtime.ts: testing", "extensions/signal/src/reply-authors.ts: clearSignalReplyAuthorsForTest", - "extensions/sms/src/channel.ts: resolveSmsTextChunkLimit", - "extensions/sms/src/config-schema.ts: SmsConfigSchema", - "extensions/sms/src/gateway.ts: registerSmsWebhookRoute", - "extensions/sms/src/twilio.ts: computeTwilioSignature", - "extensions/sms/src/twilio.ts: parseTwilioFormBody", - "extensions/sms/src/twilio.ts: TwilioSmsApiError", - "extensions/sms/src/webhook.ts: testing", "extensions/synology-chat/src/channel.ts: createSynologyChatPlugin", "extensions/synology-chat/src/client.ts: fetchChatUsers (synologyClient)", "extensions/synology-chat/src/webhook-handler.ts: clearSynologyWebhookRateLimiterStateForTest", @@ -140,14 +133,6 @@ export const KNIP_UNUSED_EXPORT_BASELINE = [ "extensions/telegram/src/topic-name-cache.ts: resetTopicNameCacheForTest", "extensions/telegram/src/topic-name-cache.ts: setTelegramTopicNameStoreFactoryForTest", "extensions/telegram/src/update-offset-store.ts: setTelegramUpdateOffsetStoreForTest", - "extensions/twitch/src/client-manager-registry.ts: clearRegistryForTest", - "extensions/twitch/src/config.ts: ResolvedTwitchAccountContext", - "extensions/twitch/src/monitor.ts: testing", - "extensions/twitch/src/monitor.ts: TwitchMonitorOptions", - "extensions/twitch/src/monitor.ts: TwitchMonitorResult", - "extensions/twitch/src/monitor.ts: TwitchRuntimeEnv", - "extensions/twitch/src/send.ts: SendMessageResult", - "extensions/twitch/src/token.ts: TwitchTokenSource", "extensions/vault/src/cli.ts: testing", "extensions/voice-call/src/cli.ts: testing", "extensions/voice-call/src/runtime-state.ts: clearVoiceCallStateRuntime",