mirror of
https://github.com/openclaw/openclaw.git
synced 2026-05-25 19:03:32 +00:00
refactor(feishu): split monitor startup and transport concerns
This commit is contained in:
Peter Steinberger
286
extensions/feishu/src/monitor.account.ts
Normal file
286
extensions/feishu/src/monitor.account.ts
Normal file
@@ -0,0 +1,286 @@
|
||||
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 });
|
||||
}
|
||||
Reference in New Issue
Block a user