openclaw/extensions/feishu/src/monitor.account.ts

287 lines
9.0 KiB
TypeScript

import * as crypto from "crypto";
import * as Lark from "@larksuiteoapi/node-sdk";
import type { ClawdbotConfig, RuntimeEnv, HistoryEntry } from "openclaw/plugin-sdk";
import { resolveFeishuAccount } from "./accounts.js";
import { raceWithTimeoutAndAbort } from "./async.js";
import { handleFeishuMessage, type FeishuMessageEvent, type FeishuBotAddedEvent } from "./bot.js";
import { handleFeishuCardAction, type FeishuCardActionEvent } from "./card-action.js";
import { createEventDispatcher } from "./client.js";
import { fetchBotOpenIdForMonitor } from "./monitor.startup.js";
import { botOpenIds } from "./monitor.state.js";
import { monitorWebhook, monitorWebSocket } from "./monitor.transport.js";
import { getMessageFeishu } from "./send.js";
import type { ResolvedFeishuAccount } from "./types.js";
const FEISHU_REACTION_VERIFY_TIMEOUT_MS = 1_500;
export type FeishuReactionCreatedEvent = {
message_id: string;
chat_id?: string;
chat_type?: "p2p" | "group";
reaction_type?: { emoji_type?: string };
operator_type?: string;
user_id?: { open_id?: string };
action_time?: string;
};
type ResolveReactionSyntheticEventParams = {
cfg: ClawdbotConfig;
accountId: string;
event: FeishuReactionCreatedEvent;
botOpenId?: string;
fetchMessage?: typeof getMessageFeishu;
verificationTimeoutMs?: number;
logger?: (message: string) => void;
uuid?: () => string;
};
export async function resolveReactionSyntheticEvent(
params: ResolveReactionSyntheticEventParams,
): Promise<FeishuMessageEvent | null> {
const {
cfg,
accountId,
event,
botOpenId,
fetchMessage = getMessageFeishu,
verificationTimeoutMs = FEISHU_REACTION_VERIFY_TIMEOUT_MS,
logger,
uuid = () => crypto.randomUUID(),
} = params;
const emoji = event.reaction_type?.emoji_type;
const messageId = event.message_id;
const senderId = event.user_id?.open_id;
if (!emoji || !messageId || !senderId) {
return null;
}
const account = resolveFeishuAccount({ cfg, accountId });
const reactionNotifications = account.config.reactionNotifications ?? "own";
if (reactionNotifications === "off") {
return null;
}
if (event.operator_type === "app" || senderId === botOpenId) {
return null;
}
if (emoji === "Typing") {
return null;
}
if (reactionNotifications === "own" && !botOpenId) {
logger?.(
`feishu[${accountId}]: bot open_id unavailable, skipping reaction ${emoji} on ${messageId}`,
);
return null;
}
const reactedMsg = await raceWithTimeoutAndAbort(fetchMessage({ cfg, messageId, accountId }), {
timeoutMs: verificationTimeoutMs,
})
.then((result) => (result.status === "resolved" ? result.value : null))
.catch(() => null);
const isBotMessage = reactedMsg?.senderType === "app" || reactedMsg?.senderOpenId === botOpenId;
if (!reactedMsg || (reactionNotifications === "own" && !isBotMessage)) {
logger?.(
`feishu[${accountId}]: ignoring reaction on non-bot/unverified message ${messageId} ` +
`(sender: ${reactedMsg?.senderOpenId ?? "unknown"})`,
);
return null;
}
const syntheticChatIdRaw = event.chat_id ?? reactedMsg.chatId;
const syntheticChatId = syntheticChatIdRaw?.trim() ? syntheticChatIdRaw : `p2p:${senderId}`;
const syntheticChatType: "p2p" | "group" = event.chat_type ?? "p2p";
return {
sender: {
sender_id: { open_id: senderId },
sender_type: "user",
},
message: {
message_id: `${messageId}:reaction:${emoji}:${uuid()}`,
chat_id: syntheticChatId,
chat_type: syntheticChatType,
message_type: "text",
content: JSON.stringify({
text: `[reacted with ${emoji} to message ${messageId}]`,
}),
},
};
}
type RegisterEventHandlersContext = {
cfg: ClawdbotConfig;
accountId: string;
runtime?: RuntimeEnv;
chatHistories: Map<string, HistoryEntry[]>;
fireAndForget?: boolean;
};
function registerEventHandlers(
eventDispatcher: Lark.EventDispatcher,
context: RegisterEventHandlersContext,
): void {
const { cfg, accountId, runtime, chatHistories, fireAndForget } = context;
const log = runtime?.log ?? console.log;
const error = runtime?.error ?? console.error;
eventDispatcher.register({
"im.message.receive_v1": async (data) => {
try {
const event = data as unknown as FeishuMessageEvent;
const promise = handleFeishuMessage({
cfg,
event,
botOpenId: botOpenIds.get(accountId),
runtime,
chatHistories,
accountId,
});
if (fireAndForget) {
promise.catch((err) => {
error(`feishu[${accountId}]: error handling message: ${String(err)}`);
});
} else {
await promise;
}
} catch (err) {
error(`feishu[${accountId}]: error handling message: ${String(err)}`);
}
},
"im.message.message_read_v1": async () => {
// Ignore read receipts
},
"im.chat.member.bot.added_v1": async (data) => {
try {
const event = data as unknown as FeishuBotAddedEvent;
log(`feishu[${accountId}]: bot added to chat ${event.chat_id}`);
} catch (err) {
error(`feishu[${accountId}]: error handling bot added event: ${String(err)}`);
}
},
"im.chat.member.bot.deleted_v1": async (data) => {
try {
const event = data as unknown as { chat_id: string };
log(`feishu[${accountId}]: bot removed from chat ${event.chat_id}`);
} catch (err) {
error(`feishu[${accountId}]: error handling bot removed event: ${String(err)}`);
}
},
"im.message.reaction.created_v1": async (data) => {
const processReaction = async () => {
const event = data as FeishuReactionCreatedEvent;
const myBotId = botOpenIds.get(accountId);
const syntheticEvent = await resolveReactionSyntheticEvent({
cfg,
accountId,
event,
botOpenId: myBotId,
logger: log,
});
if (!syntheticEvent) {
return;
}
const promise = handleFeishuMessage({
cfg,
event: syntheticEvent,
botOpenId: myBotId,
runtime,
chatHistories,
accountId,
});
if (fireAndForget) {
promise.catch((err) => {
error(`feishu[${accountId}]: error handling reaction: ${String(err)}`);
});
return;
}
await promise;
};
if (fireAndForget) {
void processReaction().catch((err) => {
error(`feishu[${accountId}]: error handling reaction event: ${String(err)}`);
});
return;
}
try {
await processReaction();
} catch (err) {
error(`feishu[${accountId}]: error handling reaction event: ${String(err)}`);
}
},
"im.message.reaction.deleted_v1": async () => {
// Ignore reaction removals
},
"card.action.trigger": async (data: unknown) => {
try {
const event = data as unknown as FeishuCardActionEvent;
const promise = handleFeishuCardAction({
cfg,
event,
botOpenId: botOpenIds.get(accountId),
runtime,
accountId,
});
if (fireAndForget) {
promise.catch((err) => {
error(`feishu[${accountId}]: error handling card action: ${String(err)}`);
});
} else {
await promise;
}
} catch (err) {
error(`feishu[${accountId}]: error handling card action: ${String(err)}`);
}
},
});
}
export type BotOpenIdSource = { kind: "prefetched"; botOpenId?: string } | { kind: "fetch" };
export type MonitorSingleAccountParams = {
cfg: ClawdbotConfig;
account: ResolvedFeishuAccount;
runtime?: RuntimeEnv;
abortSignal?: AbortSignal;
botOpenIdSource?: BotOpenIdSource;
};
export async function monitorSingleAccount(params: MonitorSingleAccountParams): Promise<void> {
const { cfg, account, runtime, abortSignal } = params;
const { accountId } = account;
const log = runtime?.log ?? console.log;
const botOpenIdSource = params.botOpenIdSource ?? { kind: "fetch" };
const botOpenId =
botOpenIdSource.kind === "prefetched"
? botOpenIdSource.botOpenId
: await fetchBotOpenIdForMonitor(account, { runtime, abortSignal });
botOpenIds.set(accountId, botOpenId ?? "");
log(`feishu[${accountId}]: bot open_id resolved: ${botOpenId ?? "unknown"}`);
const connectionMode = account.config.connectionMode ?? "websocket";
if (connectionMode === "webhook" && !account.verificationToken?.trim()) {
throw new Error(`Feishu account "${accountId}" webhook mode requires verificationToken`);
}
const eventDispatcher = createEventDispatcher(account);
const chatHistories = new Map<string, HistoryEntry[]>();
registerEventHandlers(eventDispatcher, {
cfg,
accountId,
runtime,
chatHistories,
fireAndForget: true,
});
if (connectionMode === "webhook") {
return monitorWebhook({ account, accountId, runtime, abortSignal, eventDispatcher });
}
return monitorWebSocket({ account, accountId, runtime, abortSignal, eventDispatcher });
}