diff --git a/extensions/feishu/package.json b/extensions/feishu/package.json index af6ad0d1f..d8548a441 100644 --- a/extensions/feishu/package.json +++ b/extensions/feishu/package.json @@ -22,7 +22,7 @@ "selectionLabel": "Feishu (Bot API)", "docsPath": "/channels/feishu", "docsLabel": "feishu", - "blurb": "Feishu bot with HTTP webhook events.", + "blurb": "Feishu bot events via webhook or long connection.", "aliases": [ "fs" ], diff --git a/extensions/feishu/src/accounts.ts b/extensions/feishu/src/accounts.ts index bce65c019..87db013a6 100644 --- a/extensions/feishu/src/accounts.ts +++ b/extensions/feishu/src/accounts.ts @@ -52,9 +52,12 @@ function mergeFeishuAccountConfig(cfg: ClawdbotConfig, accountId: string): Feish function hasCredentials(cfg: FeishuAccountConfig): boolean { const appId = cfg.appId?.trim(); const appSecret = cfg.appSecret?.trim(); + if (!appId || !appSecret) return false; + const mode = cfg.mode ?? "http"; + if (mode === "ws") return true; const token = cfg.verificationToken?.trim(); const encryptKey = cfg.encryptKey?.trim(); - return Boolean(appId && appSecret && (token || encryptKey)); + return Boolean(token || encryptKey); } export function resolveFeishuAccount(params: { diff --git a/extensions/feishu/src/channel.ts b/extensions/feishu/src/channel.ts index 280d2378d..e205d689b 100644 --- a/extensions/feishu/src/channel.ts +++ b/extensions/feishu/src/channel.ts @@ -41,7 +41,7 @@ const meta = { selectionLabel: "Feishu (Bot API)", docsPath: "/channels/feishu", docsLabel: "feishu", - blurb: "Feishu bot with HTTP webhook events.", + blurb: "Feishu bot events via webhook or long connection.", aliases: ["fs"], order: 85, quickstartAllowFrom: true, @@ -271,6 +271,12 @@ export const feishuPlugin: ChannelPlugin = { } const appId = (input as { appId?: unknown }).appId; const appSecret = (input as { appSecret?: unknown }).appSecret; + const modeRaw = (input as { mode?: unknown }).mode; + const mode = + typeof modeRaw === "string" && modeRaw.trim() ? modeRaw.trim().toLowerCase() : undefined; + if (mode && mode !== "http" && mode !== "ws") { + return 'Feishu --mode must be "http" or "ws".'; + } const verificationToken = (input as { verificationToken?: unknown }).verificationToken; const encryptKey = (input as { encryptKey?: unknown }).encryptKey; if (typeof appId !== "string" || !appId.trim()) { @@ -279,11 +285,13 @@ export const feishuPlugin: ChannelPlugin = { if (typeof appSecret !== "string" || !appSecret.trim()) { return "Feishu requires --app-secret."; } - if ( - (typeof verificationToken !== "string" || !verificationToken.trim()) && - (typeof encryptKey !== "string" || !encryptKey.trim()) - ) { - return "Feishu requires --verification-token or --encrypt-key."; + if (mode !== "ws") { + if ( + (typeof verificationToken !== "string" || !verificationToken.trim()) && + (typeof encryptKey !== "string" || !encryptKey.trim()) + ) { + return 'Feishu requires --verification-token or --encrypt-key when --mode="http".'; + } } return null; }, @@ -304,6 +312,10 @@ export const feishuPlugin: ChannelPlugin = { const appId = String((input as { appId?: unknown }).appId ?? "").trim(); const appSecret = String((input as { appSecret?: unknown }).appSecret ?? "").trim(); + const modeRaw = (input as { mode?: unknown }).mode; + const mode = + typeof modeRaw === "string" && modeRaw.trim() ? modeRaw.trim().toLowerCase() : undefined; + const normalizedMode = mode === "http" || mode === "ws" ? mode : undefined; const verificationTokenRaw = (input as { verificationToken?: unknown }).verificationToken; const encryptKeyRaw = (input as { encryptKey?: unknown }).encryptKey; const verificationToken = @@ -320,6 +332,7 @@ export const feishuPlugin: ChannelPlugin = { const configPatch = { appId, appSecret, + ...(normalizedMode ? { mode: normalizedMode } : {}), ...(verificationToken ? { verificationToken } : {}), ...(encryptKey ? { encryptKey } : {}), ...(webhookPath ? { webhookPath } : {}), @@ -463,6 +476,7 @@ export const feishuPlugin: ChannelPlugin = { lastStartAt: null, lastStopAt: null, lastError: null, + mode: null, webhookPath: null, lastInboundAt: null, lastOutboundAt: null, @@ -470,6 +484,7 @@ export const feishuPlugin: ChannelPlugin = { buildChannelSummary: ({ snapshot }) => ({ configured: snapshot.configured ?? false, running: snapshot.running ?? false, + mode: snapshot.mode ?? null, webhookPath: snapshot.webhookPath ?? null, lastInboundAt: snapshot.lastInboundAt ?? null, lastOutboundAt: snapshot.lastOutboundAt ?? null, @@ -486,6 +501,7 @@ export const feishuPlugin: ChannelPlugin = { enabled: account.enabled, configured: account.credentialSource !== "none", credentialSource: account.credentialSource, + mode: runtime?.mode ?? account.config.mode ?? "http", webhookPath: account.config.webhookPath, webhookUrl: account.config.webhookUrl, running: runtime?.running ?? false, @@ -500,14 +516,18 @@ export const feishuPlugin: ChannelPlugin = { gateway: { startAccount: async (ctx) => { const account = ctx.account; - ctx.log?.info(`[${account.accountId}] starting Feishu webhook`); + const mode = account.config.mode ?? "http"; + ctx.log?.info( + `[${account.accountId}] starting Feishu ${mode === "ws" ? "long connection" : "webhook"}`, + ); ctx.setStatus({ accountId: account.accountId, running: true, lastStartAt: Date.now(), - webhookPath: resolveFeishuWebhookPath({ account }), + mode, + webhookPath: mode === "http" ? resolveFeishuWebhookPath({ account }) : null, }); - const unregister = await startFeishuMonitor({ + await startFeishuMonitor({ account, config: ctx.cfg as ClawdbotConfig, runtime: ctx.runtime, @@ -516,14 +536,6 @@ export const feishuPlugin: ChannelPlugin = { webhookUrl: account.config.webhookUrl, statusSink: (patch) => ctx.setStatus({ accountId: account.accountId, ...patch }), }); - return () => { - unregister?.(); - ctx.setStatus({ - accountId: account.accountId, - running: false, - lastStopAt: Date.now(), - }); - }; }, }, }; diff --git a/extensions/feishu/src/config-mode.test.ts b/extensions/feishu/src/config-mode.test.ts new file mode 100644 index 000000000..fb7afe53d --- /dev/null +++ b/extensions/feishu/src/config-mode.test.ts @@ -0,0 +1,41 @@ +import { describe, expect, it } from "vitest"; + +import type { ClawdbotConfig } from "clawdbot/plugin-sdk"; + +import { resolveFeishuAccount } from "./accounts.js"; +import { FeishuConfigSchema } from "./config-schema.js"; + +describe("Feishu config mode", () => { + it("allows ws mode without webhook validation secrets", () => { + const parsed = FeishuConfigSchema.parse({ + appId: "cli_test", + appSecret: "secret_test", + mode: "ws", + }); + expect(parsed.mode).toBe("ws"); + }); + + it("requires webhook validation secrets in http mode", () => { + expect(() => + FeishuConfigSchema.parse({ + appId: "cli_test", + appSecret: "secret_test", + mode: "http", + }), + ).toThrow(); + }); + + it("treats ws mode as configured with app credentials", () => { + const cfg = { + channels: { + feishu: { + appId: "cli_test", + appSecret: "secret_test", + mode: "ws", + }, + }, + } as unknown as ClawdbotConfig; + const resolved = resolveFeishuAccount({ cfg, accountId: "default" }); + expect(resolved.credentialSource).toBe("config"); + }); +}); diff --git a/extensions/feishu/src/config-schema.ts b/extensions/feishu/src/config-schema.ts index 0d94bd480..0b87d124f 100644 --- a/extensions/feishu/src/config-schema.ts +++ b/extensions/feishu/src/config-schema.ts @@ -37,6 +37,7 @@ export const FeishuAccountSchema = z enabled: z.boolean().optional(), appId: z.string().optional(), appSecret: z.string().optional(), + mode: z.enum(["http", "ws"]).optional(), verificationToken: z.string().optional(), encryptKey: z.string().optional(), webhookPath: z.string().optional(), @@ -68,12 +69,13 @@ export const FeishuAccountSchema = z path: ["appId"], }); } + const mode = value.mode ?? "http"; const verificationToken = value.verificationToken?.trim(); const encryptKey = value.encryptKey?.trim(); - if ((appId || appSecret) && !verificationToken && !encryptKey) { + if (mode === "http" && (appId || appSecret) && !verificationToken && !encryptKey) { ctx.addIssue({ code: z.ZodIssueCode.custom, - message: "verificationToken or encryptKey is required for webhook validation", + message: 'verificationToken or encryptKey is required when mode="http"', path: ["verificationToken"], }); } diff --git a/extensions/feishu/src/inbound.ts b/extensions/feishu/src/inbound.ts new file mode 100644 index 000000000..be74590bf --- /dev/null +++ b/extensions/feishu/src/inbound.ts @@ -0,0 +1,448 @@ +import type { ClawdbotConfig, PluginRuntime } from "clawdbot/plugin-sdk"; +import { resolveMentionGatingWithBypass } from "clawdbot/plugin-sdk"; + +import type { ResolvedFeishuAccount } from "./accounts.js"; +import { getBotIdentity, sendFeishuTextMessage } from "./api.js"; +import type { FeishuMessagingTarget } from "./targets.js"; + +export type FeishuRuntimeEnv = { + log?: (message: string) => void; + error?: (message: string) => void; +}; + +export type FeishuStatusSink = (patch: { lastInboundAt?: number; lastOutboundAt?: number }) => void; + +function isRecord(value: unknown): value is Record { + return Boolean(value && typeof value === "object" && !Array.isArray(value)); +} + +function readString(value: unknown): string | undefined { + return typeof value === "string" && value.trim() ? value.trim() : undefined; +} + +function parseJson(raw: string): { ok: true; value: unknown } | { ok: false; error: string } { + try { + return { ok: true, value: JSON.parse(raw) as unknown }; + } catch (err) { + return { ok: false, error: err instanceof Error ? err.message : String(err) }; + } +} + +export function parseFeishuTimestampMs(value: string | undefined): number | undefined { + const raw = value?.trim(); + if (!raw) return undefined; + const num = Number.parseInt(raw, 10); + if (!Number.isFinite(num)) return undefined; + if (raw.length >= 13) return num; + return num * 1000; +} + +export function extractMentions(value: unknown): Array> { + if (!Array.isArray(value)) return []; + return value.filter(isRecord); +} + +function logVerbose(core: PluginRuntime, runtime: FeishuRuntimeEnv, message: string): void { + if (core.logging.shouldLogVerbose()) { + runtime.log?.(`[feishu] ${message}`); + } +} + +function normalizeAllowEntry(raw: string): string { + return raw + .trim() + .replace(/^feishu:/i, "") + .replace(/^fs:/i, "") + .replace(/^user:/i, "") + .replace(/^open_id:/i, "") + .replace(/^openid:/i, "") + .toLowerCase(); +} + +function isSenderAllowed( + senderOpenId: string, + senderUserId: string | undefined, + allowFrom: string[], +): boolean { + if (allowFrom.includes("*")) return true; + const openId = normalizeAllowEntry(senderOpenId); + const userId = senderUserId?.trim().toLowerCase() ?? ""; + return allowFrom.some((entry) => { + const normalized = normalizeAllowEntry(String(entry)); + if (!normalized) return false; + if (normalized === openId) return true; + if (userId && normalized === userId) return true; + return false; + }); +} + +function resolveGroupEntry(account: ResolvedFeishuAccount, chatId: string) { + const groups = account.config.groups ?? {}; + const keys = Object.keys(groups).filter(Boolean); + if (keys.length === 0) return { entry: undefined, allowlistConfigured: false }; + const entry = groups[chatId] ?? groups["*"]; + return { entry, allowlistConfigured: true }; +} + +function extractTextFromMessageContent(content: string): string { + const trimmed = content.trim(); + if (!trimmed) return ""; + const parsed = parseJson(trimmed); + if (parsed.ok && isRecord(parsed.value)) { + const text = readString(parsed.value.text); + if (text) return text; + } + return trimmed; +} + +function wasBotMentioned(params: { + mentions: Array>; + bot: { openId?: string; userId?: string }; +}): boolean { + const botOpenId = params.bot.openId?.trim(); + const botUserId = params.bot.userId?.trim(); + if (!botOpenId && !botUserId) return false; + return params.mentions.some((mention) => { + const id = isRecord(mention.id) ? mention.id : null; + const openId = readString((id ?? mention).open_id) ?? readString((id ?? mention).openId); + const userId = readString((id ?? mention).user_id) ?? readString((id ?? mention).userId); + if (botOpenId && openId && botOpenId === openId) return true; + if (botUserId && userId && botUserId === userId) return true; + return false; + }); +} + +async function deliverFeishuReply(params: { + payload: { text?: string; mediaUrls?: string[]; mediaUrl?: string; replyToId?: string }; + account: ResolvedFeishuAccount; + target: FeishuMessagingTarget; + runtime: FeishuRuntimeEnv; + core: PluginRuntime; + cfg: ClawdbotConfig; + statusSink?: FeishuStatusSink; +}) { + const { payload, account, target, runtime, core, cfg, statusSink } = params; + const mediaList = payload.mediaUrls?.length + ? payload.mediaUrls + : payload.mediaUrl + ? [payload.mediaUrl] + : []; + const replyToMessageId = payload.replyToId?.trim() || undefined; + + let text = payload.text ?? ""; + if (mediaList.length > 0) { + const suffix = mediaList.join("\n"); + text = text.trim() ? `${text.trim()}\n${suffix}` : suffix; + } + + if (!text.trim()) return; + + const chunkLimit = account.config.textChunkLimit ?? 4000; + const chunkMode = core.channel.text.resolveChunkMode(cfg, "feishu", account.accountId); + const chunks = core.channel.text.chunkMarkdownTextWithMode(text, chunkLimit, chunkMode); + for (const chunk of chunks) { + try { + await sendFeishuTextMessage({ + account, + target, + text: chunk, + replyToMessageId, + }); + statusSink?.({ lastOutboundAt: Date.now() }); + } catch (err) { + runtime.error?.(`[${account.accountId}] Feishu send failed: ${String(err)}`); + } + } +} + +export async function processFeishuMessage(params: { + account: ResolvedFeishuAccount; + config: ClawdbotConfig; + runtime: FeishuRuntimeEnv; + core: PluginRuntime; + statusSink?: FeishuStatusSink; + messageId: string; + chatId: string; + chatType: string; + messageType: string; + content: string; + mentions: Array>; + senderOpenId: string; + senderUserId?: string; + senderType?: string; + createdAt?: number; +}) { + const { + account, + config, + runtime, + core, + statusSink, + messageId, + chatId, + chatType, + messageType, + content, + mentions, + senderOpenId, + senderUserId, + senderType, + createdAt, + } = params; + + const isGroup = chatType.toLowerCase() !== "p2p"; + + if (senderType && senderType.toLowerCase() !== "user") { + logVerbose(core, runtime, `skip non-user sender (type=${senderType})`); + return; + } + + if (messageType.toLowerCase() !== "text") { + logVerbose(core, runtime, `skip non-text message (type=${messageType})`); + return; + } + + const rawBody = extractTextFromMessageContent(content).trim(); + if (!rawBody) return; + + const defaultGroupPolicy = config.channels?.defaults?.groupPolicy; + const groupPolicy = account.config.groupPolicy ?? defaultGroupPolicy ?? "allowlist"; + const groupResolved = isGroup ? resolveGroupEntry(account, chatId) : null; + const groupEntry = groupResolved?.entry; + + const groupUsers = isGroup + ? (groupEntry?.users ?? account.config.groupAllowFrom ?? []).map((v) => String(v)) + : []; + + if (isGroup) { + if (groupPolicy === "disabled") { + logVerbose(core, runtime, `drop group message (groupPolicy=disabled, chat=${chatId})`); + return; + } + const allowlistConfigured = groupResolved?.allowlistConfigured ?? false; + const allowlisted = Boolean(groupEntry) || Boolean((account.config.groups ?? {})["*"]); + if (groupPolicy === "allowlist") { + if (!allowlistConfigured) { + logVerbose( + core, + runtime, + `drop group message (groupPolicy=allowlist, no allowlist, chat=${chatId})`, + ); + return; + } + if (!allowlisted) { + logVerbose(core, runtime, `drop group message (not allowlisted, chat=${chatId})`); + return; + } + } + if (groupEntry?.enabled === false || groupEntry?.allow === false) { + logVerbose(core, runtime, `drop group message (group disabled, chat=${chatId})`); + return; + } + + if (groupUsers.length > 0) { + const ok = isSenderAllowed(senderOpenId, senderUserId, groupUsers); + if (!ok) { + logVerbose(core, runtime, `drop group message (sender not allowed, user=${senderOpenId})`); + return; + } + } + } + + const dmPolicy = account.config.dm?.policy ?? "pairing"; + const configAllowFrom = (account.config.dm?.allowFrom ?? []).map((v) => String(v)); + const shouldComputeAuth = core.channel.commands.shouldComputeCommandAuthorized(rawBody, config); + const storeAllowFrom = + !isGroup && (dmPolicy !== "open" || shouldComputeAuth) + ? await core.channel.pairing.readAllowFromStore("feishu").catch(() => []) + : []; + const effectiveAllowFrom = [...configAllowFrom, ...storeAllowFrom]; + const commandAllowFrom = isGroup ? groupUsers : effectiveAllowFrom; + const useAccessGroups = config.commands?.useAccessGroups !== false; + const senderAllowedForCommands = isSenderAllowed(senderOpenId, senderUserId, commandAllowFrom); + const commandAuthorized = shouldComputeAuth + ? core.channel.commands.resolveCommandAuthorizedFromAuthorizers({ + useAccessGroups, + authorizers: [ + { configured: commandAllowFrom.length > 0, allowed: senderAllowedForCommands }, + ], + }) + : undefined; + + if (isGroup) { + const requireMention = groupEntry?.requireMention ?? account.config.requireMention ?? true; + const allowTextCommands = core.channel.commands.shouldHandleTextCommands({ + cfg: config, + surface: "feishu", + }); + let botMentioned = false; + if (requireMention) { + try { + const bot = await getBotIdentity(account).catch(() => null); + if (bot?.openId || bot?.userId) { + botMentioned = wasBotMentioned({ + mentions, + bot: { openId: bot.openId, userId: bot.userId }, + }); + } + } catch (err) { + logVerbose(core, runtime, `bot identity lookup failed: ${String(err)}`); + } + } + + const mentionGate = resolveMentionGatingWithBypass({ + isGroup: true, + requireMention, + canDetectMention: true, + wasMentioned: botMentioned, + implicitMention: false, + hasAnyMention: mentions.length > 0, + allowTextCommands, + hasControlCommand: core.channel.text.hasControlCommand(rawBody, config), + commandAuthorized: commandAuthorized === true, + }); + if (mentionGate.shouldSkip) { + logVerbose(core, runtime, `drop group message (mention required, chat=${chatId})`); + return; + } + } + + if (!isGroup) { + if (dmPolicy === "disabled" || account.config.dm?.enabled === false) { + logVerbose(core, runtime, `blocked Feishu DM from ${senderOpenId} (dmPolicy=disabled)`); + return; + } + + if (dmPolicy !== "open") { + if (!senderAllowedForCommands) { + if (dmPolicy === "pairing") { + const { code, created } = await core.channel.pairing.upsertPairingRequest({ + channel: "feishu", + id: senderOpenId, + meta: { userId: senderUserId }, + }); + if (created) { + logVerbose(core, runtime, `feishu pairing request sender=${senderOpenId}`); + try { + await sendFeishuTextMessage({ + account, + target: { kind: "user", openId: senderOpenId }, + text: core.channel.pairing.buildPairingReply({ + channel: "feishu", + idLine: `Your Feishu open id: ${senderOpenId}`, + code, + }), + }); + statusSink?.({ lastOutboundAt: Date.now() }); + } catch (err) { + logVerbose(core, runtime, `pairing reply failed for ${senderOpenId}: ${String(err)}`); + } + } + } else { + logVerbose( + core, + runtime, + `blocked unauthorized Feishu sender ${senderOpenId} (dmPolicy=${dmPolicy})`, + ); + } + return; + } + } + } + + if ( + isGroup && + core.channel.commands.isControlCommandMessage(rawBody, config) && + commandAuthorized !== true + ) { + logVerbose(core, runtime, `feishu: drop control command from ${senderOpenId}`); + return; + } + + const peerId = isGroup ? chatId : senderOpenId; + const route = core.channel.routing.resolveAgentRoute({ + cfg: config, + channel: "feishu", + accountId: account.accountId, + peer: { kind: isGroup ? "group" : "dm", id: peerId }, + }); + + const fromLabel = isGroup ? `chat:${chatId}` : `user:${senderOpenId}`; + const storePath = core.channel.session.resolveStorePath(config.session?.store, { + agentId: route.agentId, + }); + const envelopeOptions = core.channel.reply.resolveEnvelopeFormatOptions(config); + const previousTimestamp = core.channel.session.readSessionUpdatedAt({ + storePath, + sessionKey: route.sessionKey, + }); + const body = core.channel.reply.formatAgentEnvelope({ + channel: "Feishu", + from: fromLabel, + timestamp: createdAt, + previousTimestamp, + envelope: envelopeOptions, + body: rawBody, + }); + + const groupSystemPrompt = isGroup ? groupEntry?.systemPrompt?.trim() || undefined : undefined; + + const ctxPayload = core.channel.reply.finalizeInboundContext({ + Body: body, + RawBody: rawBody, + CommandBody: rawBody, + From: `feishu:${senderOpenId}`, + To: isGroup ? `feishu:${chatId}` : `feishu:${senderOpenId}`, + SessionKey: route.sessionKey, + AccountId: route.accountId, + ChatType: isGroup ? "channel" : "direct", + ConversationLabel: fromLabel, + SenderId: senderOpenId, + SenderUsername: senderUserId, + CommandAuthorized: commandAuthorized, + Provider: "feishu", + Surface: "feishu", + MessageSid: messageId, + MessageSidFull: messageId, + ReplyToId: messageId, + ReplyToIdFull: messageId, + GroupSpace: isGroup ? chatId : undefined, + GroupSystemPrompt: groupSystemPrompt, + OriginatingChannel: "feishu", + OriginatingTo: isGroup ? `feishu:${chatId}` : `feishu:${senderOpenId}`, + }); + + void core.channel.session + .recordSessionMetaFromInbound({ + storePath, + sessionKey: ctxPayload.SessionKey ?? route.sessionKey, + ctx: ctxPayload, + }) + .catch((err) => { + runtime.error?.(`feishu: failed updating session meta: ${String(err)}`); + }); + + const target: FeishuMessagingTarget = isGroup + ? { kind: "chat", chatId } + : { kind: "user", openId: senderOpenId }; + await core.channel.reply.dispatchReplyWithBufferedBlockDispatcher({ + ctx: ctxPayload, + cfg: config, + dispatcherOptions: { + deliver: async (payload) => { + await deliverFeishuReply({ + payload, + account, + target, + runtime, + core, + cfg: config, + statusSink, + }); + }, + onError: (err, info) => { + runtime.error?.(`[${account.accountId}] Feishu ${info.kind} reply failed: ${String(err)}`); + }, + }, + }); +} diff --git a/extensions/feishu/src/monitor.lifecycle.test.ts b/extensions/feishu/src/monitor.lifecycle.test.ts new file mode 100644 index 000000000..81d2b9613 --- /dev/null +++ b/extensions/feishu/src/monitor.lifecycle.test.ts @@ -0,0 +1,80 @@ +import { createServer } from "node:http"; +import type { AddressInfo } from "node:net"; + +import { describe, expect, it, vi } from "vitest"; + +import type { ClawdbotConfig, PluginRuntime } from "clawdbot/plugin-sdk"; + +import type { ResolvedFeishuAccount } from "./accounts.js"; +import { startFeishuMonitor, handleFeishuWebhookRequest } from "./monitor.js"; +import { setFeishuRuntime } from "./runtime.js"; + +async function withServer( + handler: Parameters[0], + fn: (baseUrl: string) => Promise, +) { + const server = createServer(handler); + await new Promise((resolve) => { + server.listen(0, "127.0.0.1", () => resolve()); + }); + const address = server.address() as AddressInfo | null; + if (!address) throw new Error("missing server address"); + try { + await fn(`http://127.0.0.1:${address.port}`); + } finally { + await new Promise((resolve) => server.close(() => resolve())); + } +} + +describe("Feishu monitor lifecycle", () => { + it("unregisters webhook targets on abort", async () => { + const core = { logging: { shouldLogVerbose: () => false } } as unknown as PluginRuntime; + setFeishuRuntime(core); + + const account: ResolvedFeishuAccount = { + accountId: "default", + enabled: true, + config: { mode: "http", verificationToken: "vtok" }, + credentialSource: "config", + }; + const abort = new AbortController(); + const error = vi.fn(); + + const monitorTask = startFeishuMonitor({ + account, + config: {} as ClawdbotConfig, + runtime: { error }, + abortSignal: abort.signal, + webhookPath: "/hook", + }); + + await withServer( + async (req, res) => { + const handled = await handleFeishuWebhookRequest(req, res); + if (!handled) { + res.statusCode = 404; + res.end("not found"); + } + }, + async (baseUrl) => { + const payload = { token: "vtok", challenge: "abc", type: "url_verification" }; + const response = await fetch(`${baseUrl}/hook`, { + method: "POST", + body: JSON.stringify(payload), + }); + expect(response.status).toBe(200); + + abort.abort(); + await monitorTask; + + const response2 = await fetch(`${baseUrl}/hook`, { + method: "POST", + body: JSON.stringify(payload), + }); + expect(response2.status).toBe(404); + }, + ); + + expect(error).not.toHaveBeenCalled(); + }); +}); diff --git a/extensions/feishu/src/monitor.ts b/extensions/feishu/src/monitor.ts index 01b0952d2..e1d3be236 100644 --- a/extensions/feishu/src/monitor.ts +++ b/extensions/feishu/src/monitor.ts @@ -1,18 +1,13 @@ import type { IncomingMessage, ServerResponse } from "node:http"; import type { ClawdbotConfig } from "clawdbot/plugin-sdk"; -import { resolveMentionGatingWithBypass } from "clawdbot/plugin-sdk"; import type { ResolvedFeishuAccount } from "./accounts.js"; -import { getBotIdentity, sendFeishuTextMessage } from "./api.js"; import { decryptFeishuEncrypt, verifyFeishuSignature } from "./auth.js"; import { getFeishuRuntime } from "./runtime.js"; -import type { FeishuMessagingTarget } from "./targets.js"; - -export type FeishuRuntimeEnv = { - log?: (message: string) => void; - error?: (message: string) => void; -}; +import { processFeishuMessage, parseFeishuTimestampMs, extractMentions } from "./inbound.js"; +import type { FeishuRuntimeEnv, FeishuStatusSink } from "./inbound.js"; +import { monitorFeishuLongConnection } from "./ws.js"; export type FeishuMonitorOptions = { account: ResolvedFeishuAccount; @@ -21,7 +16,7 @@ export type FeishuMonitorOptions = { abortSignal: AbortSignal; webhookPath?: string; webhookUrl?: string; - statusSink?: (patch: { lastInboundAt?: number; lastOutboundAt?: number }) => void; + statusSink?: FeishuStatusSink; }; type FeishuCoreRuntime = ReturnType; @@ -32,7 +27,7 @@ type WebhookTarget = { runtime: FeishuRuntimeEnv; core: FeishuCoreRuntime; path: string; - statusSink?: (patch: { lastInboundAt?: number; lastOutboundAt?: number }) => void; + statusSink?: FeishuStatusSink; }; const webhookTargets = new Map(); @@ -45,12 +40,6 @@ function readString(value: unknown): string | undefined { return typeof value === "string" && value.trim() ? value.trim() : undefined; } -function logVerbose(core: FeishuCoreRuntime, runtime: FeishuRuntimeEnv, message: string): void { - if (core.logging.shouldLogVerbose()) { - runtime.log?.(`[feishu] ${message}`); - } -} - function normalizeWebhookPath(raw: string): string { const trimmed = raw.trim(); if (!trimmed) return "/"; @@ -224,419 +213,6 @@ function readHeaderValues(req: IncomingMessage) { return { timestamp, nonce, signature }; } -function parseFeishuTimestampMs(value: string | undefined): number | undefined { - const raw = value?.trim(); - if (!raw) return undefined; - const num = Number.parseInt(raw, 10); - if (!Number.isFinite(num)) return undefined; - if (raw.length >= 13) return num; - return num * 1000; -} - -function normalizeAllowEntry(raw: string): string { - return raw - .trim() - .replace(/^feishu:/i, "") - .replace(/^fs:/i, "") - .replace(/^user:/i, "") - .replace(/^open_id:/i, "") - .replace(/^openid:/i, "") - .toLowerCase(); -} - -function isSenderAllowed( - senderOpenId: string, - senderUserId: string | undefined, - allowFrom: string[], -): boolean { - if (allowFrom.includes("*")) return true; - const openId = normalizeAllowEntry(senderOpenId); - const userId = senderUserId?.trim().toLowerCase() ?? ""; - return allowFrom.some((entry) => { - const normalized = normalizeAllowEntry(String(entry)); - if (!normalized) return false; - if (normalized === openId) return true; - if (userId && normalized === userId) return true; - return false; - }); -} - -function resolveGroupEntry(account: ResolvedFeishuAccount, chatId: string) { - const groups = account.config.groups ?? {}; - const keys = Object.keys(groups).filter(Boolean); - if (keys.length === 0) return { entry: undefined, allowlistConfigured: false }; - const entry = groups[chatId] ?? groups["*"]; - return { entry, allowlistConfigured: true }; -} - -function extractTextFromMessageContent(content: string): string { - const trimmed = content.trim(); - if (!trimmed) return ""; - const parsed = parseJson(trimmed); - if (parsed.ok && isRecord(parsed.value)) { - const text = readString(parsed.value.text); - if (text) return text; - } - return trimmed; -} - -function extractMentions(value: unknown): Array> { - if (!Array.isArray(value)) return []; - return value.filter(isRecord); -} - -function wasBotMentioned(params: { - mentions: Array>; - bot: { openId?: string; userId?: string }; -}): boolean { - const botOpenId = params.bot.openId?.trim(); - const botUserId = params.bot.userId?.trim(); - if (!botOpenId && !botUserId) return false; - return params.mentions.some((mention) => { - const id = isRecord(mention.id) ? mention.id : null; - const openId = readString((id ?? mention).open_id) ?? readString((id ?? mention).openId); - const userId = readString((id ?? mention).user_id) ?? readString((id ?? mention).userId); - if (botOpenId && openId && botOpenId === openId) return true; - if (botUserId && userId && botUserId === userId) return true; - return false; - }); -} - -async function deliverFeishuReply(params: { - payload: { text?: string; mediaUrls?: string[]; mediaUrl?: string; replyToId?: string }; - account: ResolvedFeishuAccount; - target: FeishuMessagingTarget; - runtime: FeishuRuntimeEnv; - core: FeishuCoreRuntime; - cfg: ClawdbotConfig; - statusSink?: (patch: { lastInboundAt?: number; lastOutboundAt?: number }) => void; -}) { - const { payload, account, target, runtime, core, cfg, statusSink } = params; - const mediaList = payload.mediaUrls?.length - ? payload.mediaUrls - : payload.mediaUrl - ? [payload.mediaUrl] - : []; - const replyToMessageId = payload.replyToId?.trim() || undefined; - - let text = payload.text ?? ""; - if (mediaList.length > 0) { - const suffix = mediaList.join("\n"); - text = text.trim() ? `${text.trim()}\n${suffix}` : suffix; - } - - if (!text.trim()) return; - - const chunkLimit = account.config.textChunkLimit ?? 4000; - const chunkMode = core.channel.text.resolveChunkMode(cfg, "feishu", account.accountId); - const chunks = core.channel.text.chunkMarkdownTextWithMode(text, chunkLimit, chunkMode); - for (const chunk of chunks) { - try { - await sendFeishuTextMessage({ - account, - target, - text: chunk, - replyToMessageId, - }); - statusSink?.({ lastOutboundAt: Date.now() }); - } catch (err) { - runtime.error?.(`[${account.accountId}] Feishu send failed: ${String(err)}`); - } - } -} - -async function processFeishuMessage(params: { - account: ResolvedFeishuAccount; - config: ClawdbotConfig; - runtime: FeishuRuntimeEnv; - core: FeishuCoreRuntime; - statusSink?: (patch: { lastInboundAt?: number; lastOutboundAt?: number }) => void; - messageId: string; - chatId: string; - chatType: string; - messageType: string; - content: string; - mentions: Array>; - senderOpenId: string; - senderUserId?: string; - senderType?: string; - createdAt?: number; -}) { - const { - account, - config, - runtime, - core, - statusSink, - messageId, - chatId, - chatType, - messageType, - content, - mentions, - senderOpenId, - senderUserId, - senderType, - createdAt, - } = params; - - const isGroup = chatType.toLowerCase() !== "p2p"; - - if (senderType && senderType.toLowerCase() !== "user") { - logVerbose(core, runtime, `skip non-user sender (type=${senderType})`); - return; - } - - if (messageType.toLowerCase() !== "text") { - logVerbose(core, runtime, `skip non-text message (type=${messageType})`); - return; - } - - const rawBody = extractTextFromMessageContent(content).trim(); - if (!rawBody) return; - - const defaultGroupPolicy = config.channels?.defaults?.groupPolicy; - const groupPolicy = account.config.groupPolicy ?? defaultGroupPolicy ?? "allowlist"; - const groupResolved = isGroup ? resolveGroupEntry(account, chatId) : null; - const groupEntry = groupResolved?.entry; - - const groupUsers = isGroup - ? (groupEntry?.users ?? account.config.groupAllowFrom ?? []).map((v) => String(v)) - : []; - - if (isGroup) { - if (groupPolicy === "disabled") { - logVerbose(core, runtime, `drop group message (groupPolicy=disabled, chat=${chatId})`); - return; - } - const allowlistConfigured = groupResolved?.allowlistConfigured ?? false; - const allowlisted = Boolean(groupEntry) || Boolean((account.config.groups ?? {})["*"]); - if (groupPolicy === "allowlist") { - if (!allowlistConfigured) { - logVerbose( - core, - runtime, - `drop group message (groupPolicy=allowlist, no allowlist, chat=${chatId})`, - ); - return; - } - if (!allowlisted) { - logVerbose(core, runtime, `drop group message (not allowlisted, chat=${chatId})`); - return; - } - } - if (groupEntry?.enabled === false || groupEntry?.allow === false) { - logVerbose(core, runtime, `drop group message (group disabled, chat=${chatId})`); - return; - } - - if (groupUsers.length > 0) { - const ok = isSenderAllowed(senderOpenId, senderUserId, groupUsers); - if (!ok) { - logVerbose(core, runtime, `drop group message (sender not allowed, user=${senderOpenId})`); - return; - } - } - } - - const dmPolicy = account.config.dm?.policy ?? "pairing"; - const configAllowFrom = (account.config.dm?.allowFrom ?? []).map((v) => String(v)); - const shouldComputeAuth = core.channel.commands.shouldComputeCommandAuthorized(rawBody, config); - const storeAllowFrom = - !isGroup && (dmPolicy !== "open" || shouldComputeAuth) - ? await core.channel.pairing.readAllowFromStore("feishu").catch(() => []) - : []; - const effectiveAllowFrom = [...configAllowFrom, ...storeAllowFrom]; - const commandAllowFrom = isGroup ? groupUsers : effectiveAllowFrom; - const useAccessGroups = config.commands?.useAccessGroups !== false; - const senderAllowedForCommands = isSenderAllowed(senderOpenId, senderUserId, commandAllowFrom); - const commandAuthorized = shouldComputeAuth - ? core.channel.commands.resolveCommandAuthorizedFromAuthorizers({ - useAccessGroups, - authorizers: [ - { configured: commandAllowFrom.length > 0, allowed: senderAllowedForCommands }, - ], - }) - : undefined; - - if (isGroup) { - const requireMention = groupEntry?.requireMention ?? account.config.requireMention ?? true; - const allowTextCommands = core.channel.commands.shouldHandleTextCommands({ - cfg: config, - surface: "feishu", - }); - let botMentioned = false; - if (requireMention) { - try { - const bot = await getBotIdentity(account).catch(() => null); - if (bot?.openId || bot?.userId) { - botMentioned = wasBotMentioned({ - mentions, - bot: { openId: bot.openId, userId: bot.userId }, - }); - } - } catch (err) { - logVerbose(core, runtime, `bot identity lookup failed: ${String(err)}`); - } - } - - const mentionGate = resolveMentionGatingWithBypass({ - isGroup: true, - requireMention, - canDetectMention: true, - wasMentioned: botMentioned, - implicitMention: false, - hasAnyMention: mentions.length > 0, - allowTextCommands, - hasControlCommand: core.channel.text.hasControlCommand(rawBody, config), - commandAuthorized: commandAuthorized === true, - }); - if (mentionGate.shouldSkip) { - logVerbose(core, runtime, `drop group message (mention required, chat=${chatId})`); - return; - } - } - - if (!isGroup) { - if (dmPolicy === "disabled" || account.config.dm?.enabled === false) { - logVerbose(core, runtime, `blocked Feishu DM from ${senderOpenId} (dmPolicy=disabled)`); - return; - } - - if (dmPolicy !== "open") { - if (!senderAllowedForCommands) { - if (dmPolicy === "pairing") { - const { code, created } = await core.channel.pairing.upsertPairingRequest({ - channel: "feishu", - id: senderOpenId, - meta: { userId: senderUserId }, - }); - if (created) { - logVerbose(core, runtime, `feishu pairing request sender=${senderOpenId}`); - try { - await sendFeishuTextMessage({ - account, - target: { kind: "user", openId: senderOpenId }, - text: core.channel.pairing.buildPairingReply({ - channel: "feishu", - idLine: `Your Feishu open id: ${senderOpenId}`, - code, - }), - }); - statusSink?.({ lastOutboundAt: Date.now() }); - } catch (err) { - logVerbose(core, runtime, `pairing reply failed for ${senderOpenId}: ${String(err)}`); - } - } - } else { - logVerbose( - core, - runtime, - `blocked unauthorized Feishu sender ${senderOpenId} (dmPolicy=${dmPolicy})`, - ); - } - return; - } - } - } - - if ( - isGroup && - core.channel.commands.isControlCommandMessage(rawBody, config) && - commandAuthorized !== true - ) { - logVerbose(core, runtime, `feishu: drop control command from ${senderOpenId}`); - return; - } - - const peerId = isGroup ? chatId : senderOpenId; - const route = core.channel.routing.resolveAgentRoute({ - cfg: config, - channel: "feishu", - accountId: account.accountId, - peer: { kind: isGroup ? "group" : "dm", id: peerId }, - }); - - const fromLabel = isGroup ? `chat:${chatId}` : `user:${senderOpenId}`; - const storePath = core.channel.session.resolveStorePath(config.session?.store, { - agentId: route.agentId, - }); - const envelopeOptions = core.channel.reply.resolveEnvelopeFormatOptions(config); - const previousTimestamp = core.channel.session.readSessionUpdatedAt({ - storePath, - sessionKey: route.sessionKey, - }); - const body = core.channel.reply.formatAgentEnvelope({ - channel: "Feishu", - from: fromLabel, - timestamp: createdAt, - previousTimestamp, - envelope: envelopeOptions, - body: rawBody, - }); - - const groupSystemPrompt = isGroup ? groupEntry?.systemPrompt?.trim() || undefined : undefined; - - const ctxPayload = core.channel.reply.finalizeInboundContext({ - Body: body, - RawBody: rawBody, - CommandBody: rawBody, - From: `feishu:${senderOpenId}`, - To: isGroup ? `feishu:${chatId}` : `feishu:${senderOpenId}`, - SessionKey: route.sessionKey, - AccountId: route.accountId, - ChatType: isGroup ? "channel" : "direct", - ConversationLabel: fromLabel, - SenderId: senderOpenId, - SenderUsername: senderUserId, - CommandAuthorized: commandAuthorized, - Provider: "feishu", - Surface: "feishu", - MessageSid: messageId, - MessageSidFull: messageId, - ReplyToId: messageId, - ReplyToIdFull: messageId, - GroupSpace: isGroup ? chatId : undefined, - GroupSystemPrompt: groupSystemPrompt, - OriginatingChannel: "feishu", - OriginatingTo: isGroup ? `feishu:${chatId}` : `feishu:${senderOpenId}`, - }); - - void core.channel.session - .recordSessionMetaFromInbound({ - storePath, - sessionKey: ctxPayload.SessionKey ?? route.sessionKey, - ctx: ctxPayload, - }) - .catch((err) => { - runtime.error?.(`feishu: failed updating session meta: ${String(err)}`); - }); - - const target: FeishuMessagingTarget = isGroup - ? { kind: "chat", chatId } - : { kind: "user", openId: senderOpenId }; - await core.channel.reply.dispatchReplyWithBufferedBlockDispatcher({ - ctx: ctxPayload, - cfg: config, - dispatcherOptions: { - deliver: async (payload) => { - await deliverFeishuReply({ - payload, - account, - target, - runtime, - core, - cfg: config, - statusSink, - }); - }, - onError: (err, info) => { - runtime.error?.(`[${account.accountId}] Feishu ${info.kind} reply failed: ${String(err)}`); - }, - }, - }); -} - export async function handleFeishuWebhookRequest( req: IncomingMessage, res: ServerResponse, @@ -771,12 +347,33 @@ export async function handleFeishuWebhookRequest( return true; } -export async function startFeishuMonitor(options: FeishuMonitorOptions): Promise<() => void> { +function waitForAbort(signal: AbortSignal): Promise { + if (signal.aborted) return Promise.resolve(); + return new Promise((resolve) => { + signal.addEventListener("abort", () => resolve(), { once: true }); + }); +} + +export async function startFeishuMonitor(options: FeishuMonitorOptions): Promise { const core = getFeishuRuntime(); + const mode = options.account.config.mode ?? "http"; + + if (mode === "ws") { + await monitorFeishuLongConnection({ + account: options.account, + config: options.config, + runtime: options.runtime, + abortSignal: options.abortSignal, + statusSink: options.statusSink, + }); + return; + } + const webhookPath = resolveWebhookPath(options.webhookPath, options.webhookUrl); if (!webhookPath) { options.runtime.error?.(`[${options.account.accountId}] invalid webhook path`); - return () => {}; + await waitForAbort(options.abortSignal); + return; } const unregister = registerFeishuWebhookTarget({ @@ -788,7 +385,11 @@ export async function startFeishuMonitor(options: FeishuMonitorOptions): Promise statusSink: options.statusSink, }); - return unregister; + try { + await waitForAbort(options.abortSignal); + } finally { + unregister(); + } } export function resolveFeishuWebhookPath(params: { account: ResolvedFeishuAccount }): string { diff --git a/extensions/feishu/src/onboarding.ts b/extensions/feishu/src/onboarding.ts index 5462000d7..421d2f589 100644 --- a/extensions/feishu/src/onboarding.ts +++ b/extensions/feishu/src/onboarding.ts @@ -43,9 +43,11 @@ async function noteFeishuAppHelp(prompter: WizardPrompter): Promise { await prompter.note( [ "1) Create a Feishu app and enable the bot capability", - "2) Configure event subscription callback: https://gateway.example.com/feishu", - "3) Subscribe to the im.message.receive_v1 event", - "4) Copy appId/appSecret + verification token (or encrypt key) into Clawdbot", + "2) Subscribe to the im.message.receive_v1 event", + "3) Choose one delivery mode:", + " - HTTP callback: configure request URL (e.g. https://gateway.example.com/feishu) and copy the verification token (or encrypt key)", + " - Long connection: enable long connection mode (no public URL required)", + "4) Copy appId/appSecret into Clawdbot", "Docs: https://docs.clawd.bot/channels/feishu", ].join("\n"), "Feishu bot app", @@ -212,32 +214,43 @@ export const feishuOnboardingAdapter: ChannelOnboardingAdapter = { }), ).trim(); - const validationMode = await prompter.select({ - message: "Webhook validation", + const mode = await prompter.select({ + message: "Receive events via", options: [ - { value: "token", label: "Verification token", hint: "Simple shared secret check" }, - { value: "encrypt", label: "Encrypt key", hint: "Signature + encrypted payload" }, + { value: "http", label: "HTTP callback", hint: "Public webhook URL required" }, + { value: "ws", label: "Long connection", hint: "WebSocket; no public URL required" }, ], - initialValue: "token", + initialValue: resolved.config.mode ?? "http", }); - const patch: Record = { appId, appSecret }; - if (validationMode === "encrypt") { - const encryptKey = String( - await prompter.text({ - message: "Feishu encrypt key", - validate: (value) => (String(value ?? "").trim() ? undefined : "Required"), - }), - ).trim(); - patch.encryptKey = encryptKey; - } else { - const verificationToken = String( - await prompter.text({ - message: "Feishu verification token", - validate: (value) => (String(value ?? "").trim() ? undefined : "Required"), - }), - ).trim(); - patch.verificationToken = verificationToken; + const patch: Record = { appId, appSecret, mode }; + if (mode === "http") { + const validationMode = await prompter.select({ + message: "Webhook validation", + options: [ + { value: "token", label: "Verification token", hint: "Simple shared secret check" }, + { value: "encrypt", label: "Encrypt key", hint: "Signature + encrypted payload" }, + ], + initialValue: "token", + }); + + if (validationMode === "encrypt") { + const encryptKey = String( + await prompter.text({ + message: "Feishu encrypt key", + validate: (value) => (String(value ?? "").trim() ? undefined : "Required"), + }), + ).trim(); + patch.encryptKey = encryptKey; + } else { + const verificationToken = String( + await prompter.text({ + message: "Feishu verification token", + validate: (value) => (String(value ?? "").trim() ? undefined : "Required"), + }), + ).trim(); + patch.verificationToken = verificationToken; + } } const wantsPath = await prompter.confirm({ diff --git a/extensions/feishu/src/types.config.ts b/extensions/feishu/src/types.config.ts index e8df8bb6f..ec96de7d8 100644 --- a/extensions/feishu/src/types.config.ts +++ b/extensions/feishu/src/types.config.ts @@ -21,6 +21,7 @@ export type FeishuAccountConfig = { enabled?: boolean; appId?: string; appSecret?: string; + mode?: "http" | "ws"; verificationToken?: string; encryptKey?: string; webhookPath?: string; diff --git a/extensions/feishu/src/ws.frame.test.ts b/extensions/feishu/src/ws.frame.test.ts new file mode 100644 index 000000000..e22e4c156 --- /dev/null +++ b/extensions/feishu/src/ws.frame.test.ts @@ -0,0 +1,53 @@ +import { describe, expect, it } from "vitest"; + +import { decodeFeishuLongConnectionFrame, encodeFeishuLongConnectionFrame } from "./ws.js"; + +describe("Feishu long connection frame codec", () => { + it("roundtrips a ping frame", () => { + const frame = { + SeqID: 0n, + LogID: 0n, + service: 123, + method: 0, + headers: [{ key: "type", value: "ping" }], + }; + const encoded = encodeFeishuLongConnectionFrame(frame); + expect(Array.from(encoded.slice(0, 8))).toEqual([8, 0, 16, 0, 24, 123, 32, 0]); + const decoded = decodeFeishuLongConnectionFrame(encoded); + expect(decoded.SeqID).toBe(frame.SeqID); + expect(decoded.LogID).toBe(frame.LogID); + expect(decoded.service).toBe(frame.service); + expect(decoded.method).toBe(frame.method); + expect(decoded.headers).toEqual(frame.headers); + expect(decoded.payload).toBeUndefined(); + }); + + it("roundtrips frames with payload and metadata", () => { + const payload = new TextEncoder().encode(JSON.stringify({ code: 200 })); + const frame = { + SeqID: 12n, + LogID: 34n, + service: 9, + method: 1, + headers: [ + { key: "type", value: "event" }, + { key: "message_id", value: "m_1" }, + ], + payloadEncoding: "raw", + payloadType: "json", + LogIDNew: "log_1", + payload, + }; + const encoded = encodeFeishuLongConnectionFrame(frame); + const decoded = decodeFeishuLongConnectionFrame(encoded); + expect(decoded.SeqID).toBe(frame.SeqID); + expect(decoded.LogID).toBe(frame.LogID); + expect(decoded.service).toBe(frame.service); + expect(decoded.method).toBe(frame.method); + expect(decoded.headers).toEqual(frame.headers); + expect(decoded.payloadEncoding).toBe("raw"); + expect(decoded.payloadType).toBe("json"); + expect(decoded.LogIDNew).toBe("log_1"); + expect(decoded.payload).toEqual(payload); + }); +}); diff --git a/extensions/feishu/src/ws.ts b/extensions/feishu/src/ws.ts new file mode 100644 index 000000000..b68fcf8b6 --- /dev/null +++ b/extensions/feishu/src/ws.ts @@ -0,0 +1,824 @@ +import type { ClawdbotConfig, PluginRuntime } from "clawdbot/plugin-sdk"; + +import type { ResolvedFeishuAccount } from "./accounts.js"; +import { decryptFeishuEncrypt } from "./auth.js"; +import { getFeishuRuntime } from "./runtime.js"; +import { + extractMentions, + parseFeishuTimestampMs, + processFeishuMessage, + type FeishuRuntimeEnv, + type FeishuStatusSink, +} from "./inbound.js"; + +type FeishuLongConnectionOptions = { + account: ResolvedFeishuAccount; + config: ClawdbotConfig; + runtime: FeishuRuntimeEnv; + abortSignal: AbortSignal; + statusSink?: FeishuStatusSink; +}; + +type FeishuWsConnectConfig = { + connectUrl: string; + serviceId: number; + pingIntervalMs: number; + reconnectCount: number; + reconnectIntervalMs: number; + reconnectNonceMs: number; +}; + +type FrameHeader = { key: string; value: string }; + +type FrameMessage = { + SeqID: bigint; + LogID: bigint; + service: number; + method: number; + headers: FrameHeader[]; + payloadEncoding?: string; + payloadType?: string; + payload?: Uint8Array; + LogIDNew?: string; +}; + +const FEISHU_WS_CONFIG_URL = "https://open.feishu.cn/callback/ws/endpoint"; + +const FRAME_TYPE_CONTROL = 0; +const FRAME_TYPE_DATA = 1; + +const MESSAGE_TYPE_EVENT = "event"; +const MESSAGE_TYPE_PING = "ping"; +const MESSAGE_TYPE_PONG = "pong"; + +function isRecord(value: unknown): value is Record { + return Boolean(value && typeof value === "object" && !Array.isArray(value)); +} + +function readString(value: unknown): string | undefined { + return typeof value === "string" && value.trim() ? value.trim() : undefined; +} + +function parseJson(raw: string): { ok: true; value: unknown } | { ok: false; error: string } { + try { + return { ok: true, value: JSON.parse(raw) as unknown }; + } catch (err) { + return { ok: false, error: err instanceof Error ? err.message : String(err) }; + } +} + +async function sleepMs(ms: number, signal: AbortSignal): Promise { + const delay = Math.max(0, ms); + if (delay === 0 || signal.aborted) return; + await new Promise((resolve) => { + const id = setTimeout(() => { + signal.removeEventListener("abort", onAbort); + resolve(); + }, delay); + const onAbort = () => { + clearTimeout(id); + resolve(); + }; + signal.addEventListener("abort", onAbort, { once: true }); + }); +} + +async function fetchConnectConfig( + options: FeishuLongConnectionOptions, +): Promise { + const appId = options.account.config.appId?.trim() ?? ""; + const appSecret = options.account.config.appSecret?.trim() ?? ""; + if (!appId || !appSecret) { + throw new Error("Feishu long connection requires appId/appSecret"); + } + + const controller = new AbortController(); + const onAbort = () => controller.abort(); + options.abortSignal.addEventListener("abort", onAbort, { once: true }); + const timeoutId = setTimeout(() => controller.abort(), 15_000); + try { + const res = await fetch(FEISHU_WS_CONFIG_URL, { + method: "POST", + headers: { + "Content-Type": "application/json", + locale: "zh", + }, + body: JSON.stringify({ AppID: appId, AppSecret: appSecret }), + signal: controller.signal, + }); + const text = await res.text(); + const payload = text.trim() ? parseJson(text) : { ok: true as const, value: null }; + if (!payload.ok) { + throw new Error(`Feishu ws config returned invalid JSON: ${payload.error}`); + } + if (!isRecord(payload.value)) { + throw new Error(`Feishu ws config returned invalid payload (status=${res.status})`); + } + const code = typeof payload.value.code === "number" ? payload.value.code : -1; + const msg = readString(payload.value.msg) ?? "request failed"; + if (!res.ok || code !== 0) { + throw new Error(`Feishu ws config failed: code=${code} msg=${msg}`); + } + const data = isRecord(payload.value.data) ? payload.value.data : null; + const connectUrl = data ? readString(data.URL) : undefined; + const clientConfig = data && isRecord(data.ClientConfig) ? data.ClientConfig : null; + if (!connectUrl || !clientConfig) { + throw new Error("Feishu ws config missing URL/ClientConfig"); + } + + let serviceId = 0; + try { + const url = new URL(connectUrl); + const raw = url.searchParams.get("service_id")?.trim() ?? ""; + serviceId = Number.parseInt(raw, 10); + } catch { + serviceId = 0; + } + if (!Number.isFinite(serviceId) || serviceId <= 0) { + throw new Error("Feishu ws config missing service_id"); + } + + const pingIntervalSec = + typeof clientConfig.PingInterval === "number" ? clientConfig.PingInterval : 0; + const reconnectCount = + typeof clientConfig.ReconnectCount === "number" ? clientConfig.ReconnectCount : -1; + const reconnectIntervalSec = + typeof clientConfig.ReconnectInterval === "number" ? clientConfig.ReconnectInterval : 0; + const reconnectNonceSec = + typeof clientConfig.ReconnectNonce === "number" ? clientConfig.ReconnectNonce : 0; + const pingIntervalMs = Math.max(5_000, Math.floor(pingIntervalSec * 1000)); + const reconnectIntervalMs = Math.max(1_000, Math.floor(reconnectIntervalSec * 1000)); + const reconnectNonceMs = Math.max(0, Math.floor(reconnectNonceSec * 1000)); + + return { + connectUrl, + serviceId, + pingIntervalMs, + reconnectCount, + reconnectIntervalMs, + reconnectNonceMs, + }; + } finally { + clearTimeout(timeoutId); + options.abortSignal.removeEventListener("abort", onAbort); + } +} + +async function connectWebSocket(params: { + url: string; + abortSignal: AbortSignal; +}): Promise { + if (params.abortSignal.aborted) { + throw new Error("aborted"); + } + const ws = new WebSocket(params.url); + ws.binaryType = "arraybuffer"; + await new Promise((resolve, reject) => { + const onAbort = () => { + cleanup(); + try { + ws.close(); + } finally { + reject(new Error("aborted")); + } + }; + const onOpen = () => { + cleanup(); + resolve(); + }; + const onError = () => { + cleanup(); + reject(new Error("websocket error")); + }; + const cleanup = () => { + ws.removeEventListener("open", onOpen); + ws.removeEventListener("error", onError); + params.abortSignal.removeEventListener("abort", onAbort); + }; + ws.addEventListener("open", onOpen); + ws.addEventListener("error", onError); + params.abortSignal.addEventListener("abort", onAbort, { once: true }); + }); + return ws; +} + +function encodeVarint(value: bigint): number[] { + const bytes: number[] = []; + let remaining = value; + while (remaining >= 0x80n) { + bytes.push(Number(remaining & 0x7fn) | 0x80); + remaining >>= 7n; + } + bytes.push(Number(remaining)); + return bytes; +} + +function decodeVarint(data: Uint8Array, offset: number): { value: bigint; offset: number } { + let value = 0n; + let shift = 0n; + let pos = offset; + while (pos < data.length) { + const byte = data[pos++]; + value |= BigInt(byte & 0x7f) << shift; + if ((byte & 0x80) === 0) { + return { value, offset: pos }; + } + shift += 7n; + if (shift > 70n) { + throw new Error("varint overflow"); + } + } + throw new Error("unexpected EOF"); +} + +function encodeStringField(tag: number, value: string): number[] { + const encoder = new TextEncoder(); + const bytes = encoder.encode(value); + return [tag, ...encodeVarint(BigInt(bytes.length)), ...Array.from(bytes)]; +} + +function encodeBytesField(tag: number, value: Uint8Array): number[] { + return [tag, ...encodeVarint(BigInt(value.length)), ...Array.from(value)]; +} + +function decodeString(data: Uint8Array): string { + return new TextDecoder().decode(data); +} + +function decodeLengthDelimited( + data: Uint8Array, + offset: number, +): { bytes: Uint8Array; offset: number } { + const len = decodeVarint(data, offset); + const length = Number(len.value); + if (!Number.isFinite(length) || length < 0) { + throw new Error("invalid length-delimited field"); + } + const start = len.offset; + const end = start + length; + if (end > data.length) { + throw new Error("unexpected EOF"); + } + return { bytes: data.subarray(start, end), offset: end }; +} + +function encodeHeader(header: FrameHeader): number[] { + const keyField = encodeStringField(10, header.key); + const valueField = encodeStringField(18, header.value); + const body = [...keyField, ...valueField]; + return [42, ...encodeVarint(BigInt(body.length)), ...body]; +} + +function decodeHeader(data: Uint8Array): FrameHeader { + let offset = 0; + let key = ""; + let value = ""; + while (offset < data.length) { + const tag = decodeVarint(data, offset); + offset = tag.offset; + const fieldNumber = Number(tag.value >> 3n); + const wireType = Number(tag.value & 7n); + if (wireType !== 2) { + throw new Error("invalid header encoding"); + } + const field = decodeLengthDelimited(data, offset); + offset = field.offset; + const decoded = decodeString(field.bytes); + if (fieldNumber === 1) key = decoded; + if (fieldNumber === 2) value = decoded; + } + return { key, value }; +} + +export function encodeFeishuLongConnectionFrame(frame: FrameMessage): Uint8Array { + const out: number[] = []; + out.push(8, ...encodeVarint(frame.SeqID)); + out.push(16, ...encodeVarint(frame.LogID)); + out.push(24, ...encodeVarint(BigInt(frame.service))); + out.push(32, ...encodeVarint(BigInt(frame.method))); + for (const header of frame.headers) { + out.push(...encodeHeader(header)); + } + if (frame.payloadEncoding) { + out.push(...encodeStringField(50, frame.payloadEncoding)); + } + if (frame.payloadType) { + out.push(...encodeStringField(58, frame.payloadType)); + } + if (frame.payload) { + out.push(...encodeBytesField(66, frame.payload)); + } + if (frame.LogIDNew) { + out.push(...encodeStringField(74, frame.LogIDNew)); + } + return Uint8Array.from(out); +} + +export function decodeFeishuLongConnectionFrame(data: Uint8Array): FrameMessage { + let offset = 0; + let SeqID: bigint | null = null; + let LogID: bigint | null = null; + let service: number | null = null; + let method: number | null = null; + const headers: FrameHeader[] = []; + let payloadEncoding: string | undefined; + let payloadType: string | undefined; + let payload: Uint8Array | undefined; + let LogIDNew: string | undefined; + + while (offset < data.length) { + const tag = decodeVarint(data, offset); + offset = tag.offset; + const fieldNumber = Number(tag.value >> 3n); + const wireType = Number(tag.value & 7n); + + if (fieldNumber === 1 && wireType === 0) { + const v = decodeVarint(data, offset); + SeqID = v.value; + offset = v.offset; + continue; + } + if (fieldNumber === 2 && wireType === 0) { + const v = decodeVarint(data, offset); + LogID = v.value; + offset = v.offset; + continue; + } + if (fieldNumber === 3 && wireType === 0) { + const v = decodeVarint(data, offset); + service = Number(v.value); + offset = v.offset; + continue; + } + if (fieldNumber === 4 && wireType === 0) { + const v = decodeVarint(data, offset); + method = Number(v.value); + offset = v.offset; + continue; + } + if (fieldNumber === 5 && wireType === 2) { + const field = decodeLengthDelimited(data, offset); + offset = field.offset; + headers.push(decodeHeader(field.bytes)); + continue; + } + if (fieldNumber === 6 && wireType === 2) { + const field = decodeLengthDelimited(data, offset); + offset = field.offset; + payloadEncoding = decodeString(field.bytes); + continue; + } + if (fieldNumber === 7 && wireType === 2) { + const field = decodeLengthDelimited(data, offset); + offset = field.offset; + payloadType = decodeString(field.bytes); + continue; + } + if (fieldNumber === 8 && wireType === 2) { + const field = decodeLengthDelimited(data, offset); + offset = field.offset; + payload = field.bytes; + continue; + } + if (fieldNumber === 9 && wireType === 2) { + const field = decodeLengthDelimited(data, offset); + offset = field.offset; + LogIDNew = decodeString(field.bytes); + continue; + } + + if (wireType === 0) { + const v = decodeVarint(data, offset); + offset = v.offset; + continue; + } + if (wireType === 1) { + offset += 8; + continue; + } + if (wireType === 2) { + const field = decodeLengthDelimited(data, offset); + offset = field.offset; + continue; + } + if (wireType === 5) { + offset += 4; + continue; + } + throw new Error(`unsupported wireType=${wireType}`); + } + + if (SeqID == null || LogID == null || service == null || method == null) { + throw new Error("invalid frame missing required fields"); + } + + return { + SeqID, + LogID, + service, + method, + headers, + payloadEncoding, + payloadType, + payload, + LogIDNew, + }; +} + +function resolveHeaderMap(headers: FrameHeader[]): Record { + const map: Record = {}; + for (const header of headers) { + const key = header.key.trim(); + if (!key) continue; + map[key] = String(header.value ?? ""); + } + return map; +} + +function buildControlPingFrame(params: { serviceId: number }): FrameMessage { + return { + SeqID: 0n, + LogID: 0n, + service: params.serviceId, + method: FRAME_TYPE_CONTROL, + headers: [{ key: "type", value: MESSAGE_TYPE_PING }], + }; +} + +function buildAckFrame(params: { request: FrameMessage; code: number }): FrameMessage { + const payload = new TextEncoder().encode(JSON.stringify({ code: params.code })); + return { + SeqID: params.request.SeqID, + LogID: params.request.LogID, + service: params.request.service, + method: params.request.method, + headers: [...params.request.headers, { key: "biz_rt", value: "0" }], + payloadEncoding: params.request.payloadEncoding, + payloadType: params.request.payloadType, + LogIDNew: params.request.LogIDNew, + payload, + }; +} + +type ChunkBuffer = { + parts: Array; + received: number; + createdAt: number; +}; + +function mergeChunks(buffer: ChunkBuffer): Uint8Array | null { + if (!buffer.parts.every(Boolean)) return null; + const combinedLength = buffer.parts.reduce((acc, cur) => acc + (cur?.length ?? 0), 0); + const combined = new Uint8Array(combinedLength); + let offset = 0; + for (const part of buffer.parts) { + if (!part) continue; + combined.set(part, offset); + offset += part.length; + } + return combined; +} + +async function normalizeWebSocketData(data: unknown): Promise { + if (typeof data === "string") { + return new TextEncoder().encode(data); + } + if (data instanceof ArrayBuffer) { + return new Uint8Array(data); + } + if (ArrayBuffer.isView(data)) { + return new Uint8Array(data.buffer, data.byteOffset, data.byteLength); + } + if (typeof Blob !== "undefined" && data instanceof Blob) { + const buf = await data.arrayBuffer(); + return new Uint8Array(buf); + } + return null; +} + +function decryptEnvelopeIfNeeded(params: { + envelope: Record; + encryptKey?: string; +}): Record { + const encrypt = readString(params.envelope.encrypt); + if (!encrypt) return params.envelope; + const encryptKey = params.encryptKey?.trim(); + if (!encryptKey) return params.envelope; + + const { encrypt: _ignored, ...rest } = params.envelope; + try { + const decrypted = decryptFeishuEncrypt({ encrypt, encryptKey }); + const parsed = parseJson(decrypted); + if (!parsed.ok || !isRecord(parsed.value)) return params.envelope; + return { ...parsed.value, ...rest }; + } catch { + return params.envelope; + } +} + +type FeishuExtractedMessage = { + messageId: string; + chatId: string; + chatType: string; + messageType: string; + content: string; + mentions: Array>; + senderOpenId: string; + senderUserId?: string; + senderType?: string; + createdAt?: number; +}; + +function extractFeishuMessageFromEnvelope( + envelope: Record, +): FeishuExtractedMessage | null { + const eventType = readString(isRecord(envelope.header) ? envelope.header.event_type : undefined); + if (eventType !== "im.message.receive_v1") return null; + const event = isRecord(envelope.event) ? envelope.event : null; + const message = event && isRecord(event.message) ? event.message : null; + const sender = event && isRecord(event.sender) ? event.sender : null; + if (!message || !sender) return null; + + const messageId = + readString(message.message_id) ?? readString(message.messageId) ?? readString(message.id) ?? ""; + const chatId = readString(message.chat_id) ?? readString(message.chatId) ?? ""; + const chatType = readString(message.chat_type) ?? readString(message.chatType) ?? ""; + const messageType = readString(message.message_type) ?? readString(message.messageType) ?? ""; + const content = readString(message.content) ?? ""; + if (!messageId || !chatId || !chatType || !messageType || !content) return null; + + const senderId = isRecord(sender.sender_id) ? sender.sender_id : null; + const senderOpenId = + readString((senderId ?? sender).open_id) ?? readString((senderId ?? sender).openId) ?? ""; + const senderUserId = + readString((senderId ?? sender).user_id) ?? readString((senderId ?? sender).userId); + const senderType = readString(sender.sender_type) ?? readString(sender.senderType); + if (!senderOpenId) return null; + + const mentions = extractMentions(message.mentions); + const createdAt = parseFeishuTimestampMs( + readString(isRecord(envelope.header) ? envelope.header.create_time : undefined), + ); + + return { + messageId, + chatId, + chatType, + messageType, + content, + mentions, + senderOpenId, + senderUserId, + senderType, + createdAt, + }; +} + +async function handleEventPayload(params: { + core: PluginRuntime; + account: ResolvedFeishuAccount; + config: ClawdbotConfig; + runtime: FeishuRuntimeEnv; + statusSink?: FeishuStatusSink; + event: unknown; +}) { + if (!isRecord(params.event)) return; + const normalized = decryptEnvelopeIfNeeded({ + envelope: params.event, + encryptKey: params.account.config.encryptKey, + }); + const token = + readString(normalized.token) ?? + readString(isRecord(normalized.header) ? normalized.header.token : undefined); + const configuredToken = params.account.config.verificationToken?.trim(); + if (configuredToken && token && configuredToken !== token) { + return; + } + const extracted = extractFeishuMessageFromEnvelope(normalized); + if (!extracted) return; + + params.statusSink?.({ lastInboundAt: Date.now() }); + void processFeishuMessage({ + account: params.account, + config: params.config, + runtime: params.runtime, + core: params.core, + statusSink: params.statusSink, + ...extracted, + }).catch((err) => { + params.runtime.error?.( + `[${params.account.accountId}] Feishu long connection handler failed: ${String(err)}`, + ); + }); +} + +async function monitorSingleSession(params: { + ws: WebSocket; + connectConfig: FeishuWsConnectConfig; + options: FeishuLongConnectionOptions; + core: PluginRuntime; +}): Promise { + const { ws, connectConfig, options, core } = params; + const decoder = new TextDecoder(); + const chunkCache = new Map(); + const pingState = { intervalMs: connectConfig.pingIntervalMs }; + let pingTimer: ReturnType | null = null; + + const schedulePing = () => { + if (options.abortSignal.aborted) return; + if (pingTimer) clearTimeout(pingTimer); + pingTimer = setTimeout(() => { + if (ws.readyState === WebSocket.OPEN) { + try { + const ping = buildControlPingFrame({ serviceId: connectConfig.serviceId }); + ws.send(encodeFeishuLongConnectionFrame(ping)); + } catch (err) { + options.runtime.error?.( + `[${options.account.accountId}] Feishu ping failed: ${String(err)}`, + ); + } + } + schedulePing(); + }, pingState.intervalMs); + }; + + const clearCache = () => { + chunkCache.clear(); + }; + + const handleControlFrame = (frame: FrameMessage) => { + const headerMap = resolveHeaderMap(frame.headers); + const type = headerMap.type?.trim().toLowerCase(); + if (type === MESSAGE_TYPE_PONG && frame.payload) { + const payload = decoder.decode(frame.payload); + const parsed = parseJson(payload); + if (parsed.ok && isRecord(parsed.value)) { + const PingInterval = + typeof parsed.value.PingInterval === "number" ? parsed.value.PingInterval : undefined; + if (typeof PingInterval === "number" && PingInterval > 0) { + pingState.intervalMs = Math.max(5_000, Math.floor(PingInterval * 1000)); + schedulePing(); + } + } + } + }; + + const handleDataFrame = async (frame: FrameMessage) => { + const headerMap = resolveHeaderMap(frame.headers); + const type = headerMap.type?.trim().toLowerCase(); + if (type !== MESSAGE_TYPE_EVENT) return; + + const messageId = headerMap.message_id?.trim() ?? ""; + const sum = Number.parseInt(headerMap.sum ?? "", 10); + const seq = Number.parseInt(headerMap.seq ?? "", 10); + if (!messageId || !Number.isFinite(sum) || !Number.isFinite(seq) || sum <= 0 || seq < 0) { + return; + } + + const payload = frame.payload; + if (!payload || payload.length === 0) return; + + const existing = chunkCache.get(messageId); + const buffer = + existing ?? + ({ + parts: new Array(sum).fill(undefined), + received: 0, + createdAt: Date.now(), + } satisfies ChunkBuffer); + if (!existing) chunkCache.set(messageId, buffer); + + if (!buffer.parts[seq]) { + buffer.parts[seq] = payload; + buffer.received += 1; + } + + const merged = mergeChunks(buffer); + if (!merged) return; + chunkCache.delete(messageId); + + const raw = decoder.decode(merged); + const parsed = parseJson(raw); + if (!parsed.ok) { + if (core.logging.shouldLogVerbose()) { + options.runtime.error?.( + `[${options.account.accountId}] Feishu long connection invalid JSON: ${parsed.error}`, + ); + } + ws.send(encodeFeishuLongConnectionFrame(buildAckFrame({ request: frame, code: 500 }))); + return; + } + + ws.send(encodeFeishuLongConnectionFrame(buildAckFrame({ request: frame, code: 200 }))); + await handleEventPayload({ + core, + account: options.account, + config: options.config, + runtime: options.runtime, + statusSink: options.statusSink, + event: parsed.value, + }); + }; + + const closePromise = new Promise((resolve) => { + const onAbort = () => { + cleanup(); + try { + if (ws.readyState === WebSocket.OPEN || ws.readyState === WebSocket.CONNECTING) { + ws.close(); + } + } finally { + resolve(); + } + }; + const onClose = () => { + cleanup(); + resolve(); + }; + const onMessage = (event: { data: unknown }) => { + void (async () => { + const bytes = await normalizeWebSocketData(event.data); + if (!bytes) return; + const frame = decodeFeishuLongConnectionFrame(bytes); + if (frame.method === FRAME_TYPE_CONTROL) { + handleControlFrame(frame); + return; + } + if (frame.method === FRAME_TYPE_DATA) { + await handleDataFrame(frame); + } + })().catch((err) => { + options.runtime.error?.( + `[${options.account.accountId}] Feishu long connection message failed: ${String(err)}`, + ); + }); + }; + const onError = () => { + // The close event triggers the reconnect loop. + }; + const cleanup = () => { + ws.removeEventListener("close", onClose); + ws.removeEventListener("message", onMessage); + ws.removeEventListener("error", onError); + options.abortSignal.removeEventListener("abort", onAbort); + if (pingTimer) clearTimeout(pingTimer); + pingTimer = null; + clearCache(); + }; + ws.addEventListener("close", onClose); + ws.addEventListener("message", onMessage); + ws.addEventListener("error", onError); + options.abortSignal.addEventListener("abort", onAbort, { once: true }); + }); + + schedulePing(); + await closePromise; +} + +export async function monitorFeishuLongConnection( + options: FeishuLongConnectionOptions, +): Promise { + const core = getFeishuRuntime(); + + let attempts = 0; + while (!options.abortSignal.aborted) { + let connectConfig: FeishuWsConnectConfig; + try { + connectConfig = await fetchConnectConfig(options); + } catch (err) { + if (options.abortSignal.aborted) return; + options.runtime.error?.( + `[${options.account.accountId}] Feishu long connection config failed: ${String(err)}`, + ); + await sleepMs(5_000, options.abortSignal); + continue; + } + + try { + const ws = await connectWebSocket({ + url: connectConfig.connectUrl, + abortSignal: options.abortSignal, + }); + attempts = 0; + await monitorSingleSession({ ws, connectConfig, options, core }); + } catch (err) { + if (options.abortSignal.aborted) return; + options.runtime.error?.( + `[${options.account.accountId}] Feishu long connection failed: ${String(err)}`, + ); + } + + if (options.abortSignal.aborted) return; + attempts += 1; + const maxAttempts = connectConfig.reconnectCount; + if (maxAttempts >= 0 && attempts > maxAttempts) { + options.runtime.error?.( + `[${options.account.accountId}] Feishu long connection exceeded reconnect attempts`, + ); + return; + } + const jitter = + connectConfig.reconnectNonceMs > 0 ? Math.random() * connectConfig.reconnectNonceMs : 0; + await sleepMs(connectConfig.reconnectIntervalMs + jitter, options.abortSignal); + } +}