diff --git a/extensions/twitch/src/access-control.test.ts b/extensions/twitch/src/access-control.test.ts index fe3055f139bb..2957b26cbc40 100644 --- a/extensions/twitch/src/access-control.test.ts +++ b/extensions/twitch/src/access-control.test.ts @@ -12,6 +12,7 @@ describe("checkTwitchAccessControl", () => { }; const mockMessage: TwitchChatMessage = { + id: "message-1", username: "testuser", userId: "123456", message: "hello bot", diff --git a/extensions/twitch/src/monitor.test.ts b/extensions/twitch/src/monitor.test.ts index 0fa9de40bdc5..414adab7b2e4 100644 --- a/extensions/twitch/src/monitor.test.ts +++ b/extensions/twitch/src/monitor.test.ts @@ -4,8 +4,11 @@ import type { TwitchChatMessage } from "./types.js"; const mocks = vi.hoisted(() => ({ checkAccess: vi.fn(async () => ({ allowed: true })), + createIngress: vi.fn(), getClient: vi.fn(async () => ({})), getRuntime: vi.fn(), + ingressStart: vi.fn(), + ingressStop: vi.fn(async () => undefined), onMessage: vi.fn(), runInbound: vi.fn(), sendMessage: vi.fn(), @@ -28,6 +31,10 @@ vi.mock("./runtime.js", () => ({ getTwitchRuntime: mocks.getRuntime, })); +vi.mock("./twitch-ingress.js", () => ({ + createTwitchIngress: mocks.createIngress, +})); + import { monitorTwitchProvider } from "./monitor.js"; type InboundRunInput = { @@ -47,6 +54,32 @@ describe("monitorTwitchProvider", () => { vi.clearAllMocks(); mocks.getClient.mockResolvedValue({}); mocks.sendMessage.mockResolvedValue({ ok: true, messageId: "message-id" }); + mocks.createIngress.mockImplementation( + (options: { + deliver: ( + message: TwitchChatMessage, + lifecycle: { + admission: "exclusive"; + abortSignal: AbortSignal; + onAdopted: () => Promise; + onDeferred: () => void; + onAbandoned: () => Promise; + }, + ) => Promise; + }) => ({ + accept: async (message: TwitchChatMessage) => { + await options.deliver(message, { + admission: "exclusive", + abortSignal: new AbortController().signal, + onAdopted: async () => undefined, + onDeferred: () => undefined, + onAbandoned: async () => undefined, + }); + }, + start: mocks.ingressStart, + stop: mocks.ingressStop, + }), + ); mocks.runInbound.mockImplementation(async (input: InboundRunInput) => { const ingested = input.adapter.ingest(input.raw); const turn = await input.adapter.resolveTurn(ingested); @@ -108,6 +141,7 @@ describe("monitorTwitchProvider", () => { }); onMessage?.({ + id: "message-1", username: "viewer", userId: "viewer-1", message: "hello bot", @@ -124,7 +158,8 @@ describe("monitorTwitchProvider", () => { ); }); - monitor.stop(); + await monitor.stop(); expect(mocks.unregister).toHaveBeenCalledOnce(); + expect(mocks.ingressStop).toHaveBeenCalledOnce(); }); }); diff --git a/extensions/twitch/src/monitor.ts b/extensions/twitch/src/monitor.ts index 4001d5883df2..8bae8b952bfe 100644 --- a/extensions/twitch/src/monitor.ts +++ b/extensions/twitch/src/monitor.ts @@ -13,6 +13,7 @@ import { normalizeLowercaseStringOrEmpty } from "openclaw/plugin-sdk/string-coer import { checkTwitchAccessControl } from "./access-control.js"; import { getOrCreateClientManager } from "./client-manager-registry.js"; import { getTwitchRuntime } from "./runtime.js"; +import { createTwitchIngress } from "./twitch-ingress.js"; import type { TwitchAccountConfig, TwitchChatMessage } from "./types.js"; import { stripMarkdownForTwitch } from "./utils/markdown.js"; @@ -31,10 +32,11 @@ type TwitchMonitorOptions = { }; type TwitchMonitorResult = { - stop: () => void; + stop: () => Promise; }; type TwitchCoreRuntime = ReturnType; +type TwitchIngressLifecycle = Parameters[0]["deliver"]>[1]; /** * Process an incoming Twitch message and dispatch to agent. @@ -46,19 +48,22 @@ async function processTwitchMessage(params: { config: unknown; runtime: TwitchRuntimeEnv; core: TwitchCoreRuntime; + turnAdoptionLifecycle: TwitchIngressLifecycle; statusSink?: (patch: { lastInboundAt?: number; lastOutboundAt?: number }) => void; }): Promise { - const { message, account, accountId, config, runtime, core, statusSink } = params; + const { message, account, accountId, config, runtime, core, turnAdoptionLifecycle, statusSink } = + params; const cfg = config as OpenClawConfig; await core.channel.inbound.run({ channel: "twitch", accountId, raw: message, + turnAdoptionLifecycle, adapter: { ingest: (incoming) => ({ - id: incoming.id ?? `${incoming.channel}:${incoming.timestamp?.getTime() ?? Date.now()}`, - timestamp: incoming.timestamp?.getTime(), + id: incoming.id, + timestamp: incoming.timestamp, rawText: incoming.message, textForAgent: incoming.message, textForCommands: incoming.message, @@ -220,6 +225,7 @@ export async function monitorTwitchProvider( const core = getTwitchRuntime(); let stopped = false; + let stopTask: Promise | undefined; const coreLogger = core.logging.getChildLogger({ module: "twitch" }); const logVerboseMessage = (message: string) => { @@ -249,12 +255,10 @@ export async function monitorTwitchProvider( throw error; } - const unregisterHandler = clientManager.onMessage(account, (message) => { - if (stopped) { - return; - } - - void (async () => { + const ingress = createTwitchIngress({ + accountId, + runtime, + deliver: async (message, turnAdoptionLifecycle) => { const botUsername = normalizeLowercaseStringOrEmpty(account.username); if (normalizeLowercaseStringOrEmpty(message.username) === botUsername) { return; @@ -266,7 +270,7 @@ export async function monitorTwitchProvider( botUsername, }); - if (stopped || !access.allowed) { + if (!access.allowed) { return; } @@ -279,19 +283,41 @@ export async function monitorTwitchProvider( config, runtime, core, + turnAdoptionLifecycle, statusSink, }); - })().catch((err: unknown) => { - runtime.error?.(`Message processing failed: ${String(err)}`); + }, + }); + ingress.start(); + + const unregisterHandler = clientManager.onMessage(account, (message) => { + if (stopped) { + return; + } + + void ingress.accept(message).catch((err: unknown) => { + runtime.error?.(`Message durable admission failed: ${String(err)}`); }); }); - const stop = () => { - stopped = true; - unregisterHandler(); + const stop = (): Promise => { + stopTask ??= (async () => { + stopped = true; + unregisterHandler(); + await ingress.stop(); + })(); + return stopTask; }; - abortSignal.addEventListener("abort", stop, { once: true }); + abortSignal.addEventListener( + "abort", + () => { + void stop().catch((error: unknown) => { + runtime.error?.(`Twitch ingress stop failed: ${String(error)}`); + }); + }, + { once: true }, + ); return { stop }; } diff --git a/extensions/twitch/src/plugin.live.test.ts b/extensions/twitch/src/plugin.live.test.ts index 5f670c7a3ded..52d97791a8cc 100644 --- a/extensions/twitch/src/plugin.live.test.ts +++ b/extensions/twitch/src/plugin.live.test.ts @@ -1,5 +1,5 @@ /** - * Live Twitch IRC verification for the runStoppablePassiveMonitor lifecycle + * Live Twitch IRC verification for the passive account lifecycle * pattern used by the Twitch gateway. * * This test connects to irc.chat.twitch.tv using the same twurple stack the @@ -19,7 +19,7 @@ import { StaticAuthProvider } from "@twurple/auth"; import { ChatClient } from "@twurple/chat"; -import { runStoppablePassiveMonitor } from "openclaw/plugin-sdk/extension-shared"; +import { runPassiveAccountLifecycle } from "openclaw/plugin-sdk/channel-outbound"; import { describe, expect, it } from "vitest"; const LIVE = process.env.TWITCH_LIVE_TEST === "1"; @@ -33,7 +33,7 @@ const HAS_CREDS = Boolean( const maybeDescribe = LIVE && HAS_CREDS ? describe : describe.skip; maybeDescribe("twitch live IRC lifecycle (skipped unless TWITCH_LIVE_TEST=1)", () => { - it("real twurple connection + runStoppablePassiveMonitor stays pending until abort, then stops cleanly", async () => { + it("real twurple connection stays pending until abort, then stops cleanly", async () => { const accessTokenRaw = process.env.TWITCH_ACCESS_TOKEN!.replace(/^oauth:/, ""); const clientId = process.env.TWITCH_CLIENT_ID!; const channel = process.env.TWITCH_CHANNEL!; @@ -56,7 +56,7 @@ maybeDescribe("twitch live IRC lifecycle (skipped unless TWITCH_LIVE_TEST=1)", ( let settled = false; let stopCalled = false; - const task = runStoppablePassiveMonitor({ + const task = runPassiveAccountLifecycle({ abortSignal: abort.signal, start: async () => { const chat = new ChatClient({ @@ -86,6 +86,9 @@ maybeDescribe("twitch live IRC lifecycle (skipped unless TWITCH_LIVE_TEST=1)", ( }, }; }, + stop: async (monitor) => { + monitor.stop(); + }, }) .then(() => { settled = true; diff --git a/extensions/twitch/src/plugin.ts b/extensions/twitch/src/plugin.ts index f913953fed2c..122820cd165a 100644 --- a/extensions/twitch/src/plugin.ts +++ b/extensions/twitch/src/plugin.ts @@ -12,15 +12,13 @@ import { createChatChannelPlugin, stripChannelTargetPrefix, } from "openclaw/plugin-sdk/channel-core"; +import { runPassiveAccountLifecycle } from "openclaw/plugin-sdk/channel-outbound"; import { createLoggedPairingApprovalNotifier, createPairingPrefixStripper, } from "openclaw/plugin-sdk/channel-pairing"; import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts"; -import { - buildPassiveProbedChannelStatusSummary, - runStoppablePassiveMonitor, -} from "openclaw/plugin-sdk/extension-shared"; +import { buildPassiveProbedChannelStatusSummary } from "openclaw/plugin-sdk/extension-shared"; import { createComputedAccountStatusAdapter, createDefaultChannelRuntimeState, @@ -221,7 +219,7 @@ export const twitchPlugin: ChannelPlugin = // supervisor reads the settled task as `channel exited without an // error` and triggers a restart loop. See #60071. try { - await runStoppablePassiveMonitor({ + await runPassiveAccountLifecycle({ abortSignal: ctx.abortSignal, start: async () => { // Lazy import: the monitor pulls the reply pipeline; avoid ESM init cycles. @@ -234,6 +232,9 @@ export const twitchPlugin: ChannelPlugin = abortSignal: ctx.abortSignal, }); }, + stop: async (monitor) => { + await monitor.stop(); + }, }); } catch (error) { ctx.setStatus?.({ diff --git a/extensions/twitch/src/twitch-client.test.ts b/extensions/twitch/src/twitch-client.test.ts index e2e67e352ca4..e208bbdfef8b 100644 --- a/extensions/twitch/src/twitch-client.test.ts +++ b/extensions/twitch/src/twitch-client.test.ts @@ -677,11 +677,11 @@ describe("TwitchClientManager", () => { expect(capturedMessage?.displayName).toBe("TestUser"); expect(capturedMessage?.userId).toBe("12345"); expect(capturedMessage?.message).toBe("Hello bot!"); - expect(capturedMessage?.channel).toBe("testchannel"); + expect(capturedMessage?.channel).toBe("#testchannel"); expect(capturedMessage?.chatType).toBe("group"); }); - it("should normalize channel names without # prefix", async () => { + it("should preserve channel names without a # prefix", async () => { await manager.getClient(testAccount); const onMessageCallback = expectDefined(messageHandlers[0], "Twitch message handler"); diff --git a/extensions/twitch/src/twitch-client.ts b/extensions/twitch/src/twitch-client.ts index 6b2840cc93da..91ebd362dd5b 100644 --- a/extensions/twitch/src/twitch-client.ts +++ b/extensions/twitch/src/twitch-client.ts @@ -274,11 +274,10 @@ export class TwitchClientManager { client.onMessage((channelName, _user, messageText, msg) => { const handler = this.messageHandlers.get(key); if (handler) { - const normalizedChannel = channelName.startsWith("#") ? channelName.slice(1) : channelName; const from = `twitch:${msg.userInfo.userName}`; const preview = sliceUtf16Safe(messageText, 0, 100).replace(/\n/g, "\\n"); this.logger.debug?.( - `twitch inbound: channel=${normalizedChannel} from=${from} len=${messageText.length} preview="${preview}"`, + `twitch inbound: channel=${channelName} from=${from} len=${messageText.length} preview="${preview}"`, ); handler({ @@ -286,9 +285,10 @@ export class TwitchClientManager { displayName: msg.userInfo.displayName, userId: msg.userInfo.userId, message: messageText, - channel: normalizedChannel, + // Preserve the raw callback channel; durable dispatch normalizes it. + channel: channelName, id: msg.id, - timestamp: new Date(), + timestamp: Date.now(), isMod: msg.userInfo.isMod, isOwner: msg.userInfo.isBroadcaster, isVip: msg.userInfo.isVip, diff --git a/extensions/twitch/src/twitch-ingress.test-support.ts b/extensions/twitch/src/twitch-ingress.test-support.ts new file mode 100644 index 000000000000..1b854f858455 --- /dev/null +++ b/extensions/twitch/src/twitch-ingress.test-support.ts @@ -0,0 +1,74 @@ +// Twitch tests share isolated durable-ingress state and raw chat envelopes. +import fs from "node:fs/promises"; +import path from "node:path"; +import { + closeOpenClawStateDatabaseForTest, + createChannelIngressQueueForTests, +} from "openclaw/plugin-sdk/plugin-state-test-runtime"; +import { resolvePreferredOpenClawTmpDir } from "openclaw/plugin-sdk/temp-path"; +import { expect, vi } from "vitest"; +import { createTwitchIngress } from "./twitch-ingress.js"; +import type { TwitchChatMessage } from "./types.js"; + +type TwitchIngressTestQueue = NonNullable[0]["queue"]>; +export type TwitchIngressTestPayload = Parameters[1]; + +export function createTwitchIngressTestMessage( + params: Partial = {}, +): TwitchChatMessage { + return { + id: params.id ?? "message-1", + username: params.username ?? "viewer", + userId: params.userId ?? "viewer-1", + displayName: params.displayName ?? "Viewer", + message: params.message ?? "hello bot", + channel: params.channel ?? "#TestChannel", + timestamp: params.timestamp ?? 1_721_300_000_000, + isMod: params.isMod ?? false, + isOwner: params.isOwner ?? false, + isVip: params.isVip ?? false, + isSub: params.isSub ?? false, + chatType: "group", + }; +} + +export async function withTwitchIngressTestQueue( + fn: (queue: TwitchIngressTestQueue) => Promise, +): Promise { + const createdDir = await fs.mkdtemp( + path.join(resolvePreferredOpenClawTmpDir(), "openclaw-twitch-ingress-"), + ); + const stateDir = await fs.realpath(createdDir); + const previousStateDir = process.env.OPENCLAW_STATE_DIR; + process.env.OPENCLAW_STATE_DIR = stateDir; + const queue = createChannelIngressQueueForTests({ + channelId: "twitch", + accountId: "default", + stateDir, + }); + try { + return await fn(queue); + } finally { + if (previousStateDir === undefined) { + delete process.env.OPENCLAW_STATE_DIR; + } else { + process.env.OPENCLAW_STATE_DIR = previousStateDir; + } + closeOpenClawStateDatabaseForTest(); + await fs.rm(stateDir, { recursive: true, force: true }); + } +} + +export async function waitForTwitchIngressVerdict( + queue: TwitchIngressTestQueue, + eventId: string, + expected: "completed" | "failed", +): Promise { + await vi.waitFor( + async () => { + const verdict = await queue.enqueue(eventId, { version: 1, rawEvent: "{}" }); + expect(verdict.kind).toBe(expected); + }, + { timeout: 5_000 }, + ); +} diff --git a/extensions/twitch/src/twitch-ingress.test.ts b/extensions/twitch/src/twitch-ingress.test.ts new file mode 100644 index 000000000000..ed809c4c2964 --- /dev/null +++ b/extensions/twitch/src/twitch-ingress.test.ts @@ -0,0 +1,285 @@ +// Twitch durable ingress tests cover raw admission, recovery, and tombstones. +import type { ChannelIngressQueue } from "openclaw/plugin-sdk/channel-outbound"; +import { closeOpenClawStateDatabaseForTest } from "openclaw/plugin-sdk/plugin-state-test-runtime"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { createTwitchIngress } from "./twitch-ingress.js"; +import { + createTwitchIngressTestMessage, + waitForTwitchIngressVerdict, + withTwitchIngressTestQueue, + type TwitchIngressTestPayload, +} from "./twitch-ingress.test-support.js"; + +function runtime() { + return { error: vi.fn() }; +} + +afterEach(() => { + closeOpenClawStateDatabaseForTest(); + vi.restoreAllMocks(); +}); + +describe("Twitch durable ingress", () => { + it("durably appends before dispatch", async () => { + await withTwitchIngressTestQueue(async (queue) => { + const realEnqueue = queue.enqueue.bind(queue); + let releaseAppend = () => {}; + const appendGate = new Promise((resolve) => { + releaseAppend = resolve; + }); + const enqueue: typeof queue.enqueue = vi.fn( + async (...args: Parameters) => { + await appendGate; + return await realEnqueue(...args); + }, + ); + const gatedQueue: ChannelIngressQueue = { ...queue, enqueue }; + const deliver = vi.fn(async (_message, lifecycle) => { + await lifecycle.onAdopted(); + }); + const ingress = createTwitchIngress({ + accountId: "default", + runtime: runtime(), + queue: gatedQueue, + deliver, + pollIntervalMs: 5, + }); + ingress.start(); + try { + const admission = ingress.accept(createTwitchIngressTestMessage({ id: "durable-first" })); + await vi.waitFor(() => expect(enqueue).toHaveBeenCalledOnce()); + expect(deliver).not.toHaveBeenCalled(); + releaseAppend(); + await admission; + await waitForTwitchIngressVerdict(queue, "durable-first", "completed"); + expect(deliver).toHaveBeenCalledOnce(); + } finally { + releaseAppend(); + await ingress.stop(); + } + }); + }); + + it("recovers an uncompleted event with a fresh drain and dispatches exactly once", async () => { + await withTwitchIngressTestQueue(async (queue) => { + const interrupted = createTwitchIngress({ + accountId: "default", + runtime: runtime(), + queue, + deliver: vi.fn(), + }); + await interrupted.accept(createTwitchIngressTestMessage({ id: "restart" })); + await interrupted.stop(); + + const deliver = vi.fn(async (_message, lifecycle) => { + await lifecycle.onAdopted(); + }); + const recovered = createTwitchIngress({ + accountId: "default", + runtime: runtime(), + queue, + deliver, + pollIntervalMs: 5, + }); + recovered.start(); + try { + await waitForTwitchIngressVerdict(queue, "restart", "completed"); + expect(deliver).toHaveBeenCalledOnce(); + } finally { + await recovered.stop(); + } + }); + }); + + it("keeps a completion tombstone and rejects a post-completion duplicate", async () => { + await withTwitchIngressTestQueue(async (queue) => { + const deliver = vi.fn(async (_message, lifecycle) => { + await lifecycle.onAdopted(); + }); + const ingress = createTwitchIngress({ + accountId: "default", + runtime: runtime(), + queue, + deliver, + pollIntervalMs: 5, + }); + const message = createTwitchIngressTestMessage({ id: "duplicate" }); + ingress.start(); + try { + await ingress.accept(message); + await waitForTwitchIngressVerdict(queue, "duplicate", "completed"); + await ingress.accept(message); + await new Promise((resolve) => { + setTimeout(resolve, 30); + }); + expect(deliver).toHaveBeenCalledOnce(); + } finally { + await ingress.stop(); + } + }); + }); + + it("stores the raw callback envelope and normalizes its channel only at dispatch", async () => { + await withTwitchIngressTestQueue(async (queue) => { + const message = createTwitchIngressTestMessage({ + id: "raw", + channel: "#MixedCase", + message: "before", + }); + const delivered = vi.fn(async (_message, lifecycle) => { + await lifecycle.onAdopted(); + }); + const ingress = createTwitchIngress({ + accountId: "default", + runtime: runtime(), + queue, + deliver: delivered, + pollIntervalMs: 5, + }); + await ingress.accept(message); + expect(await queue.listPending()).toEqual([ + expect.objectContaining({ + id: "raw", + laneKey: "channel:mixedcase", + payload: { version: 1, rawEvent: JSON.stringify(message) }, + }), + ]); + message.message = "after"; + + ingress.start(); + try { + await waitForTwitchIngressVerdict(queue, "raw", "completed"); + expect(delivered).toHaveBeenCalledWith( + expect.objectContaining({ channel: "mixedcase", message: "before" }), + expect.any(Object), + ); + } finally { + await ingress.stop(); + } + }); + }); + + it("dead-letters malformed persisted JSON without dispatch", async () => { + await withTwitchIngressTestQueue(async (queue) => { + await queue.enqueue( + "malformed", + { version: 1, rawEvent: "{" }, + { laneKey: "channel:testchannel" }, + ); + const deliver = vi.fn(); + const ingress = createTwitchIngress({ + accountId: "default", + runtime: runtime(), + queue, + deliver, + pollIntervalMs: 5, + }); + ingress.start(); + try { + await waitForTwitchIngressVerdict(queue, "malformed", "failed"); + expect(deliver).not.toHaveBeenCalled(); + } finally { + await ingress.stop(); + } + }); + }); + + it("waits for an in-flight durable admission before stop returns", async () => { + await withTwitchIngressTestQueue(async (queue) => { + const realEnqueue = queue.enqueue.bind(queue); + let releaseAppend = () => {}; + const appendGate = new Promise((resolve) => { + releaseAppend = resolve; + }); + const enqueue: typeof queue.enqueue = async (...args: Parameters) => { + await appendGate; + return await realEnqueue(...args); + }; + const ingress = createTwitchIngress({ + accountId: "default", + runtime: runtime(), + queue: { ...queue, enqueue }, + deliver: vi.fn(), + }); + const admission = ingress.accept(createTwitchIngressTestMessage({ id: "admitting" })); + let stopped = false; + const stopping = ingress.stop().then(() => { + stopped = true; + }); + await new Promise((resolve) => { + setTimeout(resolve, 30); + }); + expect(stopped).toBe(false); + releaseAppend(); + await admission; + await stopping; + expect(stopped).toBe(true); + }); + }); + + it("waits for an adopted active delivery before stop returns", async () => { + await withTwitchIngressTestQueue(async (queue) => { + let releaseDelivery = () => {}; + const deliveryGate = new Promise((resolve) => { + releaseDelivery = resolve; + }); + const deliver = vi.fn(async (_message, lifecycle) => { + await lifecycle.onAdopted(); + await deliveryGate; + }); + const ingress = createTwitchIngress({ + accountId: "default", + runtime: runtime(), + queue, + deliver, + pollIntervalMs: 5, + }); + ingress.start(); + await ingress.accept(createTwitchIngressTestMessage({ id: "active-stop" })); + await vi.waitFor(() => expect(deliver).toHaveBeenCalledOnce()); + + let stopped = false; + const stopping = ingress.stop().then(() => { + stopped = true; + }); + await new Promise((resolve) => { + setTimeout(resolve, 30); + }); + expect(stopped).toBe(false); + releaseDelivery(); + await stopping; + expect(stopped).toBe(true); + }); + }); + + it("releases a pre-adoption delivery for retry during shutdown", async () => { + await withTwitchIngressTestQueue(async (queue) => { + let releaseDelivery = () => {}; + const deliveryGate = new Promise((resolve) => { + releaseDelivery = resolve; + }); + const deliver = vi.fn(async () => { + await deliveryGate; + }); + const ingress = createTwitchIngress({ + accountId: "default", + runtime: runtime(), + queue, + deliver, + pollIntervalMs: 5, + }); + ingress.start(); + await ingress.accept(createTwitchIngressTestMessage({ id: "shutdown-retry" })); + await vi.waitFor(() => expect(deliver).toHaveBeenCalledOnce()); + + const stopping = ingress.stop(); + releaseDelivery(); + await stopping; + + expect(await queue.listClaims()).toHaveLength(0); + expect(await queue.listPending()).toEqual([ + expect.objectContaining({ id: "shutdown-retry", lastError: expect.any(String) }), + ]); + }); + }); +}); diff --git a/extensions/twitch/src/twitch-ingress.ts b/extensions/twitch/src/twitch-ingress.ts new file mode 100644 index 000000000000..bdc541ce4d46 --- /dev/null +++ b/extensions/twitch/src/twitch-ingress.ts @@ -0,0 +1,356 @@ +// Twitch plugin owns raw chat-envelope durable admission and replay draining. +import { HttpStatusCodeError } from "@twurple/api-call"; +import { + bindIngressLifecycleToReplyOptions, + createChannelIngressDrain, + DEFAULT_INGRESS_ADOPTION_STALL_MS, + DEFAULT_INGRESS_RETRY_DEAD_LETTER_MIN_AGE_MS, + DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS, + type ChannelIngressQueue, +} from "openclaw/plugin-sdk/channel-outbound"; +import { formatErrorMessage } from "openclaw/plugin-sdk/error-runtime"; +import { getTwitchRuntime } from "./runtime.js"; +import type { TwitchChatMessage } from "./types.js"; +import { normalizeTwitchChannel } from "./utils/twitch.js"; + +const TWITCH_INGRESS_PAYLOAD_VERSION = 1; +const TWITCH_INGRESS_DRAIN_INTERVAL_MS = 1_000; +const TWITCH_INGRESS_PRUNE_INTERVAL_MS = 60 * 60 * 1_000; +const TWITCH_INGRESS_COMPLETED_TTL_MS = 30 * 24 * 60 * 60 * 1_000; +// Twitch IRC does not replay accepted PRIVMSG lines. These tombstones are near-inert; +// the durable queue protects the local accept-to-dispatch crash window instead. +const TWITCH_INGRESS_COMPLETED_MAX_ENTRIES = 1_000; +const TWITCH_INGRESS_FAILED_TTL_MS = 30 * 24 * 60 * 60 * 1_000; +const TWITCH_INGRESS_FAILED_MAX_ENTRIES = 1_000; +const TWITCH_INGRESS_APPEND_RETRY_DELAYS_MS = [0, 100, 300] as const; + +type TwitchIngressPayload = { + version: typeof TWITCH_INGRESS_PAYLOAD_VERSION; + rawEvent: string; +}; + +type TwitchIngressLifecycle = ReturnType< + typeof bindIngressLifecycleToReplyOptions +>["turnAdoptionLifecycle"]; + +type TwitchIngress = { + accept: (message: TwitchChatMessage) => Promise; + start: () => void; + stop: () => Promise; +}; + +class TwitchIngressPermanentError extends Error { + constructor(message: string, options?: ErrorOptions) { + super(message, options); + this.name = "TwitchIngressPermanentError"; + } +} + +function nonEmptyString(value: unknown): string | null { + return typeof value === "string" && value.trim() ? value.trim() : null; +} + +function inspectTwitchIngressEvent(event: unknown): { eventId: string; laneKey: string } { + if (!event || typeof event !== "object" || Array.isArray(event)) { + throw new TwitchIngressPermanentError("Twitch ingress event must be an object."); + } + const candidate = event as { id?: unknown; channel?: unknown }; + const eventId = nonEmptyString(candidate.id); + if (!eventId) { + throw new TwitchIngressPermanentError("Twitch ingress event is missing its message id."); + } + const rawChannel = nonEmptyString(candidate.channel); + const channel = rawChannel ? normalizeTwitchChannel(rawChannel) : ""; + if (!channel) { + throw new TwitchIngressPermanentError("Twitch ingress event is missing its channel."); + } + return { eventId, laneKey: `channel:${channel}` }; +} + +function parseClaimedTwitchMessage( + payload: TwitchIngressPayload, + claimedId: string, + claimedLaneKey: string | undefined, +): TwitchChatMessage { + if (payload.version !== TWITCH_INGRESS_PAYLOAD_VERSION || typeof payload.rawEvent !== "string") { + throw new TwitchIngressPermanentError("Twitch ingress payload is invalid."); + } + let parsed: unknown; + try { + parsed = JSON.parse(payload.rawEvent); + } catch (error) { + throw new TwitchIngressPermanentError("Twitch ingress event JSON is invalid.", { + cause: error, + }); + } + const facts = inspectTwitchIngressEvent(parsed); + if (facts.eventId !== claimedId || facts.laneKey !== claimedLaneKey) { + throw new TwitchIngressPermanentError( + "Twitch ingress event identity changed after durable admission.", + ); + } + const candidate = parsed as Partial; + const username = nonEmptyString(candidate.username); + const rawChannel = nonEmptyString(candidate.channel); + if (!username || typeof candidate.message !== "string" || !rawChannel) { + throw new TwitchIngressPermanentError("Twitch ingress event shape is invalid."); + } + return { + ...candidate, + id: claimedId, + username, + message: candidate.message, + channel: normalizeTwitchChannel(rawChannel), + } as TwitchChatMessage; +} + +function isTwitchAuthenticationFailure(error: unknown): boolean { + let current: unknown = error; + for (let depth = 0; depth < 8 && current && typeof current === "object"; depth += 1) { + if ( + current instanceof HttpStatusCodeError && + (current.statusCode === 401 || current.statusCode === 403) + ) { + return true; + } + current = (current as { cause?: unknown }).cause; + } + return false; +} + +function stoppedError(): Error { + return new Error("Twitch ingress stopped before dispatch."); +} + +export function createTwitchIngress(options: { + accountId: string; + runtime: { error?: (message: string) => void }; + deliver: (message: TwitchChatMessage, lifecycle: TwitchIngressLifecycle) => Promise; + queue?: ChannelIngressQueue; + pollIntervalMs?: number; +}): TwitchIngress { + const queue = + options.queue ?? + getTwitchRuntime().state.openChannelIngressQueue({ + accountId: options.accountId, + }); + const shutdown = new AbortController(); + const activeDeliveries = new Set>(); + const deferredClaims = new Map>(); + let running = false; + let stopped = false; + let drainRequested = false; + let drainTask: Promise | undefined; + let drainTimer: ReturnType | undefined; + let lastPrunedAt = 0; + let admissionTail: Promise = Promise.resolve(); + let stopTask: Promise | undefined; + + const drain = createChannelIngressDrain({ + queue, + abortSignal: shutdown.signal, + adoptionStallTimeoutMs: DEFAULT_INGRESS_ADOPTION_STALL_MS, + retryPolicy: { + maxAttempts: DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS, + deadLetterMinAgeMs: DEFAULT_INGRESS_RETRY_DEAD_LETTER_MIN_AGE_MS, + }, + resolveNonRetryableFailure: (error) => { + if (error instanceof TwitchIngressPermanentError) { + return { reason: "invalid-event", message: error.message }; + } + if (isTwitchAuthenticationFailure(error)) { + return { reason: "authentication-failed", message: formatErrorMessage(error) }; + } + return null; + }, + onLog: (message) => options.runtime.error?.(`twitch ingress: ${message}`), + dispatchClaimedEvent: async (claimed, lifecycle) => { + if (!running || lifecycle.abortSignal.aborted) { + return { kind: "failed-retryable", error: stoppedError() }; + } + const message = parseClaimedTwitchMessage(claimed.payload, claimed.id, claimed.laneKey); + const bound = bindIngressLifecycleToReplyOptions(lifecycle).turnAdoptionLifecycle; + let handedOff = false; + let resolveDeferredClaim!: () => void; + const deferredClaim = new Promise((resolve) => { + resolveDeferredClaim = resolve; + }); + let deferredClaimSettled = false; + const settleDeferredClaim = () => { + if (deferredClaimSettled) { + return; + } + deferredClaimSettled = true; + if (deferredClaims.get(claimed.id) === deferredClaim) { + deferredClaims.delete(claimed.id); + } + resolveDeferredClaim(); + }; + const delivery = options.deliver(message, { + ...bound, + onAdopted: async () => { + handedOff = true; + try { + await bound.onAdopted(); + } finally { + settleDeferredClaim(); + } + }, + onDeferred: () => { + handedOff = true; + if (!deferredClaimSettled) { + deferredClaims.set(claimed.id, deferredClaim); + } + bound.onDeferred(); + }, + onAbandoned: async () => { + handedOff = true; + try { + await bound.onAbandoned(); + } finally { + settleDeferredClaim(); + } + }, + }); + activeDeliveries.add(delivery); + try { + await delivery; + } catch (error) { + if (!running || lifecycle.abortSignal.aborted) { + return { kind: "failed-retryable", error }; + } + throw error; + } finally { + activeDeliveries.delete(delivery); + } + if (!handedOff) { + if (!running || lifecycle.abortSignal.aborted) { + return { kind: "failed-retryable", error: stoppedError() }; + } + // Echoes and access-gated messages are terminal no-dispatch events. + await bound.onAdopted(); + } + return deferredClaims.has(claimed.id) ? { kind: "deferred" } : { kind: "completed" }; + }, + }); + + const pruneIfDue = async (): Promise => { + const now = Date.now(); + if (now - lastPrunedAt < TWITCH_INGRESS_PRUNE_INTERVAL_MS) { + return; + } + await queue.prune({ + completedTtlMs: TWITCH_INGRESS_COMPLETED_TTL_MS, + completedMaxEntries: TWITCH_INGRESS_COMPLETED_MAX_ENTRIES, + failedTtlMs: TWITCH_INGRESS_FAILED_TTL_MS, + failedMaxEntries: TWITCH_INGRESS_FAILED_MAX_ENTRIES, + now, + }); + lastPrunedAt = now; + }; + + const requestDrain = (): void => { + if (!running || stopped || shutdown.signal.aborted) { + return; + } + drainRequested = true; + if (drainTask) { + return; + } + drainTask = (async () => { + while (drainRequested) { + if (!running) { + break; + } + drainRequested = false; + await pruneIfDue(); + if (!running) { + break; + } + const { started } = await drain.drainOnce({ shouldStop: () => !running }); + if (!running || (!drainRequested && started === 0)) { + break; + } + } + })() + .catch((error: unknown) => { + options.runtime.error?.(`Twitch ingress drain failed: ${formatErrorMessage(error)}`); + }) + .finally(() => { + drainTask = undefined; + if (running && drainRequested) { + requestDrain(); + } + }); + }; + + const admitOnce = async (message: TwitchChatMessage): Promise => { + const facts = inspectTwitchIngressEvent(message); + const rawEvent = JSON.stringify(message); + const receivedAt = Date.now(); + let lastError: unknown; + for (const delayMs of TWITCH_INGRESS_APPEND_RETRY_DELAYS_MS) { + if (delayMs > 0) { + await new Promise((resolve) => { + setTimeout(resolve, delayMs); + }); + } + try { + await queue.enqueue( + facts.eventId, + { version: TWITCH_INGRESS_PAYLOAD_VERSION, rawEvent }, + { receivedAt, laneKey: facts.laneKey }, + ); + requestDrain(); + return; + } catch (error) { + lastError = error; + } + } + throw lastError; + }; + + return { + accept: (message) => { + if (stopped) { + return Promise.reject(stoppedError()); + } + // Preserve socket arrival order across append retry backoff. + const admission = admissionTail.then(() => admitOnce(message)); + admissionTail = admission.catch(() => undefined); + return admission; + }, + start: () => { + if (running || stopped) { + return; + } + running = true; + requestDrain(); + drainTimer = setInterval( + requestDrain, + options.pollIntervalMs ?? TWITCH_INGRESS_DRAIN_INTERVAL_MS, + ); + drainTimer.unref?.(); + }, + stop: () => { + stopTask ??= (async () => { + stopped = true; + running = false; + if (drainTimer) { + clearInterval(drainTimer); + drainTimer = undefined; + } + await admissionTail; + shutdown.abort(stoppedError()); + await drainTask; + await Promise.allSettled(activeDeliveries); + await Promise.allSettled(deferredClaims.values()); + await drain.waitForIdle(); + // Stop is idempotent, and drain disposal remains safe if cleanup repeats. + drain.dispose(); + drain.dispose(); + })(); + return stopTask; + }, + }; +} diff --git a/extensions/twitch/src/types.ts b/extensions/twitch/src/types.ts index e8105e2b998a..ceb7ab691792 100644 --- a/extensions/twitch/src/types.ts +++ b/extensions/twitch/src/types.ts @@ -74,9 +74,9 @@ export interface TwitchChatMessage { /** Display name (may include special characters) */ displayName?: string; /** Message ID */ - id?: string; - /** Timestamp */ - timestamp?: Date; + id: string; + /** Receive timestamp in milliseconds */ + timestamp?: number; /** Whether the sender is a moderator */ isMod?: boolean; /** Whether the sender is the channel owner/broadcaster */