feat(feishu): add long connection mode
This commit is contained in:
Johnny
parent
229736e6ec
commit
1f7d41183d
@ -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"
|
||||
],
|
||||
|
||||
@ -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: {
|
||||
|
||||
@ -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<ResolvedFeishuAccount> = {
|
||||
}
|
||||
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<ResolvedFeishuAccount> = {
|
||||
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<ResolvedFeishuAccount> = {
|
||||
|
||||
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<ResolvedFeishuAccount> = {
|
||||
const configPatch = {
|
||||
appId,
|
||||
appSecret,
|
||||
...(normalizedMode ? { mode: normalizedMode } : {}),
|
||||
...(verificationToken ? { verificationToken } : {}),
|
||||
...(encryptKey ? { encryptKey } : {}),
|
||||
...(webhookPath ? { webhookPath } : {}),
|
||||
@ -463,6 +476,7 @@ export const feishuPlugin: ChannelPlugin<ResolvedFeishuAccount> = {
|
||||
lastStartAt: null,
|
||||
lastStopAt: null,
|
||||
lastError: null,
|
||||
mode: null,
|
||||
webhookPath: null,
|
||||
lastInboundAt: null,
|
||||
lastOutboundAt: null,
|
||||
@ -470,6 +484,7 @@ export const feishuPlugin: ChannelPlugin<ResolvedFeishuAccount> = {
|
||||
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<ResolvedFeishuAccount> = {
|
||||
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<ResolvedFeishuAccount> = {
|
||||
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<ResolvedFeishuAccount> = {
|
||||
webhookUrl: account.config.webhookUrl,
|
||||
statusSink: (patch) => ctx.setStatus({ accountId: account.accountId, ...patch }),
|
||||
});
|
||||
return () => {
|
||||
unregister?.();
|
||||
ctx.setStatus({
|
||||
accountId: account.accountId,
|
||||
running: false,
|
||||
lastStopAt: Date.now(),
|
||||
});
|
||||
};
|
||||
},
|
||||
},
|
||||
};
|
||||
|
||||
41
extensions/feishu/src/config-mode.test.ts
Normal file
41
extensions/feishu/src/config-mode.test.ts
Normal file
@ -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");
|
||||
});
|
||||
});
|
||||
@ -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"],
|
||||
});
|
||||
}
|
||||
|
||||
448
extensions/feishu/src/inbound.ts
Normal file
448
extensions/feishu/src/inbound.ts
Normal file
@ -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<string, unknown> {
|
||||
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<Record<string, unknown>> {
|
||||
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<Record<string, unknown>>;
|
||||
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<Record<string, unknown>>;
|
||||
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)}`);
|
||||
},
|
||||
},
|
||||
});
|
||||
}
|
||||
80
extensions/feishu/src/monitor.lifecycle.test.ts
Normal file
80
extensions/feishu/src/monitor.lifecycle.test.ts
Normal file
@ -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<typeof createServer>[0],
|
||||
fn: (baseUrl: string) => Promise<void>,
|
||||
) {
|
||||
const server = createServer(handler);
|
||||
await new Promise<void>((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<void>((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();
|
||||
});
|
||||
});
|
||||
@ -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<typeof getFeishuRuntime>;
|
||||
@ -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<string, WebhookTarget[]>();
|
||||
@ -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<Record<string, unknown>> {
|
||||
if (!Array.isArray(value)) return [];
|
||||
return value.filter(isRecord);
|
||||
}
|
||||
|
||||
function wasBotMentioned(params: {
|
||||
mentions: Array<Record<string, unknown>>;
|
||||
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<Record<string, unknown>>;
|
||||
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<void> {
|
||||
if (signal.aborted) return Promise.resolve();
|
||||
return new Promise((resolve) => {
|
||||
signal.addEventListener("abort", () => resolve(), { once: true });
|
||||
});
|
||||
}
|
||||
|
||||
export async function startFeishuMonitor(options: FeishuMonitorOptions): Promise<void> {
|
||||
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 {
|
||||
|
||||
@ -43,9 +43,11 @@ async function noteFeishuAppHelp(prompter: WizardPrompter): Promise<void> {
|
||||
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<string, unknown> = { 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<string, unknown> = { 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({
|
||||
|
||||
@ -21,6 +21,7 @@ export type FeishuAccountConfig = {
|
||||
enabled?: boolean;
|
||||
appId?: string;
|
||||
appSecret?: string;
|
||||
mode?: "http" | "ws";
|
||||
verificationToken?: string;
|
||||
encryptKey?: string;
|
||||
webhookPath?: string;
|
||||
|
||||
53
extensions/feishu/src/ws.frame.test.ts
Normal file
53
extensions/feishu/src/ws.frame.test.ts
Normal file
@ -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);
|
||||
});
|
||||
});
|
||||
824
extensions/feishu/src/ws.ts
Normal file
824
extensions/feishu/src/ws.ts
Normal file
@ -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<string, unknown> {
|
||||
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<void> {
|
||||
const delay = Math.max(0, ms);
|
||||
if (delay === 0 || signal.aborted) return;
|
||||
await new Promise<void>((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<FeishuWsConnectConfig> {
|
||||
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<WebSocket> {
|
||||
if (params.abortSignal.aborted) {
|
||||
throw new Error("aborted");
|
||||
}
|
||||
const ws = new WebSocket(params.url);
|
||||
ws.binaryType = "arraybuffer";
|
||||
await new Promise<void>((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<string, string> {
|
||||
const map: Record<string, string> = {};
|
||||
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<Uint8Array | undefined>;
|
||||
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<Uint8Array | null> {
|
||||
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<string, unknown>;
|
||||
encryptKey?: string;
|
||||
}): Record<string, unknown> {
|
||||
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<Record<string, unknown>>;
|
||||
senderOpenId: string;
|
||||
senderUserId?: string;
|
||||
senderType?: string;
|
||||
createdAt?: number;
|
||||
};
|
||||
|
||||
function extractFeishuMessageFromEnvelope(
|
||||
envelope: Record<string, unknown>,
|
||||
): 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<void> {
|
||||
const { ws, connectConfig, options, core } = params;
|
||||
const decoder = new TextDecoder();
|
||||
const chunkCache = new Map<string, ChunkBuffer>();
|
||||
const pingState = { intervalMs: connectConfig.pingIntervalMs };
|
||||
let pingTimer: ReturnType<typeof setTimeout> | 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<void>((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<void> {
|
||||
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);
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue
Block a user