diff --git a/extensions/twitch/index.ts b/extensions/twitch/index.ts index 7f2aa6960..25adc4705 100644 --- a/extensions/twitch/index.ts +++ b/extensions/twitch/index.ts @@ -4,6 +4,8 @@ import { emptyPluginConfigSchema } from "clawdbot/plugin-sdk"; import { twitchPlugin } from "./src/plugin.js"; import { setTwitchRuntime } from "./src/runtime.js"; +export { monitorTwitchProvider } from "./src/monitor.js"; + const plugin = { id: "twitch", name: "Twitch", @@ -11,7 +13,7 @@ const plugin = { configSchema: emptyPluginConfigSchema(), register(api: ClawdbotPluginApi) { setTwitchRuntime(api.runtime); - api.registerChannel({ plugin: twitchPlugin }); + api.registerChannel({ plugin: twitchPlugin as any }); }, }; diff --git a/extensions/twitch/src/monitor.ts b/extensions/twitch/src/monitor.ts new file mode 100644 index 000000000..a905cdece --- /dev/null +++ b/extensions/twitch/src/monitor.ts @@ -0,0 +1,251 @@ +/** + * Twitch message monitor - processes incoming messages and routes to agents. + * + * This monitor connects to the Twitch client manager, processes incoming messages, + * resolves agent routes, and handles replies. + */ + +import type { ReplyPayload } from "clawdbot/plugin-sdk"; +import type { TwitchAccountConfig, TwitchChatMessage } from "./types.js"; +import { checkTwitchAccessControl } from "./access-control.js"; +import { getTwitchRuntime } from "./runtime.js"; +import { getOrCreateClientManager } from "./client-manager-registry.js"; + +export type TwitchRuntimeEnv = { + log?: (message: string) => void; + error?: (message: string) => void; +}; + +export type TwitchMonitorOptions = { + account: TwitchAccountConfig; + accountId: string; + config: unknown; // ClawdbotConfig + runtime: TwitchRuntimeEnv; + abortSignal: AbortSignal; + statusSink?: (patch: { lastInboundAt?: number; lastOutboundAt?: number }) => void; +}; + +export type TwitchMonitorResult = { + stop: () => void; +}; + +type TwitchCoreRuntime = ReturnType; + +/** + * Process an incoming Twitch message and dispatch to agent. + */ +async function processTwitchMessage(params: { + message: TwitchChatMessage; + account: TwitchAccountConfig; + accountId: string; + config: unknown; + runtime: TwitchRuntimeEnv; + core: TwitchCoreRuntime; + statusSink?: (patch: { lastInboundAt?: number; lastOutboundAt?: number }) => void; +}): Promise { + const { message, account, accountId, config, runtime, core, statusSink } = params; + + // Resolve route for this message + const route = core.channel.routing.resolveAgentRoute({ + cfg: config as Parameters[0]["cfg"], + channel: "twitch", + accountId, + peer: { + kind: "group", // Twitch chat is always group-like + id: message.channel, + }, + }); + + // Build message body + const rawBody = message.message; + const body = core.channel.reply.formatAgentEnvelope({ + channel: "Twitch", + from: message.displayName ?? message.username, + timestamp: message.timestamp?.getTime(), + envelope: core.channel.reply.resolveEnvelopeFormatOptions(config as Parameters[0]), + body: rawBody, + }); + + // Build context payload + const ctxPayload = core.channel.reply.finalizeInboundContext({ + Body: body, + RawBody: rawBody, + CommandBody: rawBody, + From: `twitch:user:${message.userId}`, + To: `twitch:channel:${message.channel}`, + SessionKey: route.sessionKey, + AccountId: route.accountId, + ChatType: "group", + ConversationLabel: message.channel, + SenderName: message.displayName ?? message.username, + SenderId: message.userId, + SenderUsername: message.username, + Provider: "twitch", + Surface: "twitch", + MessageSid: message.id, + OriginatingChannel: "twitch", + OriginatingTo: `twitch:channel:${message.channel}`, + }); + + // Record session + const storePath = core.channel.session.resolveStorePath( + (config as Parameters[0]["cfg"])?.session?.store, + { agentId: route.agentId }, + ); + await core.channel.session.recordInboundSession({ + storePath, + sessionKey: ctxPayload.SessionKey ?? route.sessionKey, + ctx: ctxPayload, + onRecordError: (err) => { + runtime.error?.(`[twitch] Failed updating session meta: ${String(err)}`); + }, + }); + + // Resolve markdown table mode + const tableMode = core.channel.text.resolveMarkdownTableMode({ + cfg: config as Parameters[0]["cfg"], + channel: "twitch", + accountId, + }); + + // Dispatch reply + await core.channel.reply.dispatchReplyWithBufferedBlockDispatcher({ + ctx: ctxPayload, + cfg: config as Parameters[0]["cfg"], + dispatcherOptions: { + deliver: async (payload) => { + await deliverTwitchReply({ + payload, + channel: message.channel, + account, + runtime, + statusSink, + }); + }, + }, + runtime, + tableMode, + }); +} + +/** + * Deliver a reply to Twitch chat. + */ +async function deliverTwitchReply(params: { + payload: ReplyPayload; + channel: string; + account: TwitchAccountConfig; + runtime: TwitchRuntimeEnv; + statusSink?: (patch: { lastInboundAt?: number; lastOutboundAt?: number }) => void; +}): Promise { + const { payload, channel, account, runtime, statusSink } = params; + + try { + const clientManager = getOrCreateClientManager(account.accountId ?? "default", { + info: (msg) => runtime.log?.(`[twitch] ${msg}`), + warn: (msg) => runtime.log?.(`[twitch] ${msg}`), + error: (msg) => runtime.error?.(`[twitch] ${msg}`), + debug: (msg) => runtime.log?.(`[twitch] ${msg}`), + }); + + const client = await clientManager.getClient(account, null, account.accountId ?? "default"); + if (!client) { + runtime.error?.(`[twitch] No client available for sending reply`); + return; + } + + // Send the reply + await client.say(channel, payload.text); + statusSink?.({ lastOutboundAt: Date.now() }); + } catch (err) { + runtime.error?.(`[twitch] Failed to send reply: ${String(err)}`); + } +} + +/** + * Main monitor provider for Twitch. + * + * Sets up message handlers and processes incoming messages. + */ +export async function monitorTwitchProvider( + options: TwitchMonitorOptions, +): Promise { + const { account, accountId, config, runtime, abortSignal, statusSink } = options; + + const core = getTwitchRuntime(); + let stopped = false; + + // Create logger for client manager + const logger = { + info: (msg: string) => runtime.log?.(`[twitch] ${msg}`), + warn: (msg: string) => runtime.log?.(`[twitch] ${msg}`), + error: (msg: string) => runtime.error?.(`[twitch] ${msg}`), + debug: (msg: string) => runtime.log?.(`[twitch] ${msg}`), + }; + + // Get or create client manager + const clientManager = getOrCreateClientManager(accountId, logger); + + // Establish connection + try { + await clientManager.getClient(account, config as Parameters[1], accountId); + runtime.log?.(`[twitch] Connected to Twitch as ${account.username}`); + } catch (error) { + const errorMsg = error instanceof Error ? error.message : String(error); + runtime.error?.(`[twitch] Failed to connect: ${errorMsg}`); + throw error; + } + + // Register message handler + const unregisterHandler = clientManager.onMessage(account, async (message) => { + if (stopped) return; + + // Access control check + const botUsername = account.username.toLowerCase(); + if (message.username.toLowerCase() === botUsername) { + return; // Ignore own messages + } + + const access = checkTwitchAccessControl({ + message, + account, + botUsername, + }); + + if (!access.allowed) { + runtime.log?.( + `[twitch] Ignored message from ${message.username}: ${access.reason ?? "blocked"}`, + ); + return; + } + + statusSink?.({ lastInboundAt: Date.now() }); + + try { + await processTwitchMessage({ + message, + account, + accountId, + config, + runtime, + core, + statusSink, + }); + } catch (err) { + runtime.error?.(`[twitch] Message processing failed: ${String(err)}`); + } + }); + + // Stop function + const stop = () => { + stopped = true; + unregisterHandler(); + }; + + // Handle abort signal + abortSignal.addEventListener("abort", stop, { once: true }); + + runtime.log?.(`[twitch] Monitor started for account ${accountId}`); + + return { stop }; +} diff --git a/extensions/twitch/src/plugin.ts b/extensions/twitch/src/plugin.ts index f3338159c..eca3e62b9 100644 --- a/extensions/twitch/src/plugin.ts +++ b/extensions/twitch/src/plugin.ts @@ -5,8 +5,6 @@ * This is the primary entry point for the Twitch channel integration. */ -import { checkTwitchAccessControl } from "./access-control.js"; -import { resolveAgentRoute } from "../../../src/routing/resolve-route.js"; import type { ClawdbotConfig } from "clawdbot/plugin-sdk"; import { buildChannelConfigSchema } from "clawdbot/plugin-sdk"; import { twitchMessageActions } from "./actions.js"; @@ -21,10 +19,7 @@ import { twitchOutbound } from "./outbound.js"; import { probeTwitch } from "./probe.js"; import { resolveTwitchTargets } from "./resolver.js"; import { collectTwitchStatusIssues } from "./status.js"; -import { - getOrCreateClientManager, - removeClientManager, -} from "./client-manager-registry.js"; +import { removeClientManager } from "./client-manager-registry.js"; import type { ChannelAccountSnapshot, ChannelCapabilities, @@ -245,61 +240,6 @@ export const twitchPlugin: ChannelPlugin = { const account = ctx.account as TwitchAccountConfig; const accountId = ctx.accountId; - // Create logger for client manager - const logger: ChannelLogSink = { - info: (msg) => ctx.log?.info(`[twitch] ${msg}`), - warn: (msg) => ctx.log?.warn(`[twitch] ${msg}`), - error: (msg) => ctx.log?.error(`[twitch] ${msg}`), - debug: (msg) => ctx.log?.debug?.(`[twitch] ${msg}`), - }; - - // Get or create client manager from registry - const clientManager = getOrCreateClientManager(accountId, logger); - - // Set up message handler - clientManager.onMessage(account, (message) => { - // Access control check - const botUsername = account.username.toLowerCase(); - if (message.username.toLowerCase() === botUsername) { - return; // Ignore own messages - } - - const access = checkTwitchAccessControl({ - message, - account, - botUsername, - }); - - if (!access.allowed) { - ctx.log?.info( - `[twitch] Ignored message from ${message.username}: ${access.reason ?? "blocked"}`, - ); - return; - } - - // Resolve route for this message - resolveAgentRoute({ - cfg: ctx.cfg, - channel: "twitch", - accountId, - peer: { - kind: "group", // Twitch chat is always group-like - id: message.channel, - }, - }); - - // Build message preview - const preview = message.message.replace(/\s+/g, " ").slice(0, 160); - - // Log message receipt - // Note: Message dispatch to agent system will be handled by a separate monitor/loop - // For now, we just log that we received and validated the message - ctx.log?.info( - `[twitch] Received message from ${message.displayName ?? message.username}: ${preview}`, - ); - }); - - // Update status ctx.setStatus?.({ accountId, running: true, @@ -311,19 +251,15 @@ export const twitchPlugin: ChannelPlugin = { `[twitch] Starting Twitch connection for ${account.username}`, ); - try { - await clientManager.getClient(account, ctx.cfg); - ctx.log?.info(`[twitch] Connected to Twitch as ${account.username}`); - } catch (error) { - const errorMsg = error instanceof Error ? error.message : String(error); - ctx.log?.error(`[twitch] Failed to connect: ${errorMsg}`); - ctx.setStatus?.({ - accountId, - running: false, - lastError: errorMsg, - }); - throw error; - } + // Lazy import: the monitor pulls the reply pipeline; avoid ESM init cycles. + const { monitorTwitchProvider } = await import("./monitor.js"); + await monitorTwitchProvider({ + account, + accountId, + config: ctx.cfg, + runtime: ctx.runtime, + abortSignal: ctx.abortSignal, + }); }, /** Stop an account connection */ diff --git a/extensions/twitch/src/send.test.ts b/extensions/twitch/src/send.test.ts index df0dd878e..20050da70 100644 --- a/extensions/twitch/src/send.test.ts +++ b/extensions/twitch/src/send.test.ts @@ -285,6 +285,7 @@ describe("send", () => { "testchannel", // normalized account channel "Hello!", mockConfig, + "default", ); }); }); diff --git a/extensions/twitch/src/send.ts b/extensions/twitch/src/send.ts index f9066f19a..3922a86b0 100644 --- a/extensions/twitch/src/send.ts +++ b/extensions/twitch/src/send.ts @@ -113,6 +113,7 @@ export async function sendMessageTwitchInternal( normalizeTwitchChannel(normalizedChannel), cleanedText, cfg, + accountId, ); if (!result.ok) { diff --git a/extensions/twitch/src/twitch-client.ts b/extensions/twitch/src/twitch-client.ts index 7745c959f..1ec4ec971 100644 --- a/extensions/twitch/src/twitch-client.ts +++ b/extensions/twitch/src/twitch-client.ts @@ -96,6 +96,7 @@ export class TwitchClientManager { async getClient( account: TwitchAccountConfig, cfg?: ClawdbotConfig, + accountId?: string, ): Promise { const key = this.getAccountKey(account); @@ -106,7 +107,7 @@ export class TwitchClientManager { // Resolve token from config or environment const tokenResolution = resolveTwitchToken(cfg, { - accountId: account.username, + accountId, }); if (!tokenResolution.token) { @@ -283,9 +284,10 @@ export class TwitchClientManager { channel: string, message: string, cfg?: ClawdbotConfig, + accountId?: string, ): Promise<{ ok: boolean; error?: string; messageId?: string }> { try { - const client = await this.getClient(account, cfg); + const client = await this.getClient(account, cfg, accountId); // Generate a message ID (Twurple's say() doesn't return the message ID, so we generate one) const messageId = `${Date.now()}-${Math.random().toString(36).substring(2, 15)}`; diff --git a/tsconfig.json b/tsconfig.json index 879b14415..8f82c611d 100644 --- a/tsconfig.json +++ b/tsconfig.json @@ -5,7 +5,6 @@ "moduleResolution": "NodeNext", "outDir": "dist", "rootDir": "src", - "declaration": true, "strict": true, "esModuleInterop": true, "forceConsistentCasingInFileNames": true,