diff --git a/docs/connectors.md b/docs/connectors.md index ea2f427..322af21 100644 --- a/docs/connectors.md +++ b/docs/connectors.md @@ -92,4 +92,4 @@ im:message.group_msg im:message:send_as_bot ``` -Invite the bot to any group it should watch. A loop fires on an @mention in those groups and replies in the thread. Feishu does not list p2p chats for a bot, so DMs are not polled. +Invite the bot to any group it should watch. A loop fires on an @mention in those groups and answers in a thread under it, so the group itself stays readable. An @mention inside that thread is answered there too, with what was said earlier in the thread; threads go unread after a week without a message. Feishu does not list p2p chats for a bot, so DMs are not polled. diff --git a/src/connectors/feishu/api.test.ts b/src/connectors/feishu/api.test.ts index 3207896..c8d0721 100644 --- a/src/connectors/feishu/api.test.ts +++ b/src/connectors/feishu/api.test.ts @@ -177,6 +177,7 @@ describe("replyMessage", () => { expect(JSON.parse(String(init.body))).toEqual({ msg_type: "text", content: "{\"text\":\"Cause: redis timeout\"}", + reply_in_thread: true, }); return ok({ code: 0, diff --git a/src/connectors/feishu/api.ts b/src/connectors/feishu/api.ts index aa7572c..9863a72 100644 --- a/src/connectors/feishu/api.ts +++ b/src/connectors/feishu/api.ts @@ -31,6 +31,8 @@ export type FeishuMessage = { senderId?: string; senderType?: string; mentionIds: string[]; + /** The thread it is in, or the thread it starts. */ + threadId?: string; }; type Envelope = { @@ -178,17 +180,44 @@ function mentionsOf(raw: unknown): string[] { .filter(Boolean); } +/** + * A group's messages newer than `oldestMs`, oldest first. In an ordinary group + * that is only the first message of each thread: what is said inside one is + * read with `threadHistory`. + */ export async function history( credential: FeishuCredential, chatId: string, oldestMs: string, ): Promise { + return await messagesIn(credential, { type: "chat", id: chatId, chatId }, oldestMs); +} + +/** What was said in one thread after `oldestMs`, oldest first. */ +export async function threadHistory( + credential: FeishuCredential, + thread: { threadId: string; chatId: string }, + oldestMs: string, +): Promise { + return await messagesIn( + credential, + { type: "thread", id: thread.threadId, chatId: thread.chatId }, + oldestMs, + ); +} + +async function messagesIn( + credential: FeishuCredential, + container: { type: "chat" | "thread"; id: string; chatId: string }, + oldestMs: string, +): Promise { + const chatId = container.chatId; const parsed: FeishuMessage[] = []; let pageToken = ""; do { const query: Record = { - container_id_type: "chat", - container_id: chatId, + container_id_type: container.type, + container_id: container.id, sort_type: "ByCreateTimeAsc", page_size: "50", start_time: String(Math.floor(Number(oldestMs) / 1000)), @@ -209,6 +238,7 @@ export async function history( const message = raw as { message_id?: string; chat_id?: string; + thread_id?: string; create_time?: string; msg_type?: string; deleted?: boolean; @@ -225,6 +255,7 @@ export async function history( ...(message.sender?.id ? { senderId: message.sender.id } : {}), ...(message.sender?.sender_type ? { senderType: message.sender.sender_type } : {}), mentionIds: mentionsOf(message.mentions), + ...(message.thread_id ? { threadId: message.thread_id } : {}), }); } pageToken = data.has_more === true && typeof data.page_token === "string" ? data.page_token : ""; @@ -232,6 +263,23 @@ export async function history( return parsed.sort((a, b) => Number(a.createTime) - Number(b.createTime)); } +/** + * The thread a message is in or starts, or null while it has none. A message + * the bot was asked in only gets one once the answer is posted. + */ +export async function threadOf(credential: FeishuCredential, messageId: string): Promise { + const data = dataOf( + await call(credential.host, `/im/v1/messages/${messageId}`, { token: credential.tenantAccessToken }), + ); + const items = Array.isArray(data.items) ? data.items : []; + const message = (items[0] ?? {}) as { thread_id?: unknown }; + return typeof message.thread_id === "string" && message.thread_id ? message.thread_id : null; +} + +/** + * Answers in a thread under the message, or in the thread it is already in, + * so a conversation with the bot does not fill the group. + */ export async function replyMessage( credential: FeishuCredential, messageId: string, @@ -240,7 +288,7 @@ export async function replyMessage( const body = await call(credential.host, `/im/v1/messages/${messageId}/reply`, { token: credential.tenantAccessToken, method: "POST", - body: { msg_type: "text", content: JSON.stringify({ text }) }, + body: { msg_type: "text", content: JSON.stringify({ text }), reply_in_thread: true }, }); const data = dataOf(body); const id = typeof data.message_id === "string" ? data.message_id : ""; diff --git a/src/connectors/feishu/runtime.test.ts b/src/connectors/feishu/runtime.test.ts index 0e15f4f..fb50cbf 100644 --- a/src/connectors/feishu/runtime.test.ts +++ b/src/connectors/feishu/runtime.test.ts @@ -25,6 +25,8 @@ function givenFeishu(handlers: { chats?: Array<{ chat_id: string; name?: string }>; history?: Record>>; reply?: { message_id: string; chat_id: string }; + /** Messages looked up one at a time, by id. */ + messages?: Record>; send?: { message_id: string; chat_id: string }; }) { const sent: Call[] = []; @@ -57,6 +59,12 @@ function givenFeishu(handlers: { const chat = new URL(url).searchParams.get("container_id") ?? ""; return json({ code: 0, data: { items: handlers.history?.[chat] ?? [], has_more: false } }); } + const one = /^\/im\/v1\/messages\/([^/]+)$/.exec(path)?.[1]; + if (one && init?.method !== "POST") { + const message = handlers.messages?.[one]; + if (!message) return json({ code: 230011, msg: "message not found" }); + return json({ code: 0, data: { items: [message] } }); + } if (path.endsWith("/reply")) { return json({ code: 0, data: handlers.reply ?? { message_id: "om_reply", chat_id: "oc_eng" } }); } @@ -65,7 +73,10 @@ function givenFeishu(handlers: { return sent; } -afterEach(() => vi.unstubAllGlobals()); +afterEach(() => { + vi.unstubAllGlobals(); + vi.useRealTimers(); +}); describe("connectWithFields", () => { it("refuses a missing app id", async () => { @@ -145,7 +156,11 @@ describe("feishu poll", () => { const { signals, cursor } = await poll(null); expect(signals.map((signal) => [signal.key, signal.at])).toEqual([["oc_eng:om_ask", 1690000030000]]); - expect(cursor).toEqual({ oc_eng: "1690000030000", oc_quiet: "1690000000000" }); + expect(cursor).toEqual({ + oc_eng: "1690000030000", + oc_quiet: "1690000000000", + threads: { om_ask: { chatId: "oc_eng", messageId: "om_ask", last: "1690000030000" } }, + }); }); it("keeps a group mention of the bot and skips chat and its own replies", async () => { @@ -199,7 +214,112 @@ describe("feishu poll", () => { at: 1700000002000, }, ]); - expect(cursor).toEqual({ oc_eng: "1700000003000" }); + expect(cursor).toEqual({ + oc_eng: "1700000003000", + threads: { om_ask: { chatId: "oc_eng", messageId: "om_ask", last: "1700000002000" } }, + }); + }); +}); + +describe("feishu threads", () => { + const poll = (threads: Record) => + feishuRuntime.poll!({ + workflowId: "feishu.ask", + settings: {}, + credential: account, + cursor: { oc_eng: "1700000000000", threads } as never, + since: 1690000000000, + }); + + const said = (id: string, at: string, text: string, by: "ou_maya" | "bot", mention = false) => ({ + message_id: id, + create_time: at, + chat_id: "oc_eng", + thread_id: "omt_1", + msg_type: "text", + sender: by === "bot" ? { id: "cli_app", sender_type: "app" } : { id: by, sender_type: "user" }, + body: { content: JSON.stringify({ text }) }, + ...(mention ? { mentions: [{ id: "ou_bot" }] } : {}), + }); + + it("answers a mention inside the thread its answer opened, with what came before", async () => { + vi.useFakeTimers({ now: 1700000100000, toFake: ["Date"] }); + givenFeishu({ + chats: [{ chat_id: "oc_eng", name: "eng" }], + messages: { om_ask: { message_id: "om_ask", thread_id: "omt_1" } }, + history: { + omt_1: [ + said("om_ask", "1700000002000", "@_user_1 checkout 5xx", "ou_maya", true), + said("om_answer", "1700000003000", "Cause: redis timeout", "bot"), + said("om_more", "1700000004000", "@_user_1 since when?", "ou_maya", true), + said("om_chat", "1700000005000", "thanks all", "ou_maya"), + ], + }, + }); + + const { signals, cursor } = await poll({ + om_ask: { chatId: "oc_eng", messageId: "om_ask", last: "1700000002000" }, + }); + + expect(signals.map((signal) => signal.key)).toEqual(["oc_eng:om_more"]); + expect(signals[0]!.payload).toMatchObject({ + messageId: "om_more", + earlier: [ + { by: "person", text: "@_user_1 checkout 5xx" }, + { by: "bot", text: "Cause: redis timeout" }, + ], + }); + expect((cursor as { threads: unknown }).threads).toEqual({ + omt_1: { chatId: "oc_eng", messageId: "om_ask", threadId: "omt_1", last: "1700000005000" }, + }); + + const item = await feishuRuntime.resolveWorkItem!({ url: "", payload: signals[0]!.payload ?? null, credential: account }); + expect(item.context[0]!.body).toBe( + [ + "# Earlier in this thread", + "", + "**Them:** @_user_1 checkout 5xx", + "", + "**You:** Cause: redis timeout", + "", + "# What you were asked", + "", + "@_user_1 since when?", + "", + ].join("\n"), + ); + }); + + it("waits for a thread to open, and lets go of one that went quiet", async () => { + const day = 24 * 60 * 60 * 1000; + const now = 1700000100000; + vi.useFakeTimers({ now, toFake: ["Date"] }); + givenFeishu({ + chats: [], + messages: { om_waiting: { message_id: "om_waiting" } }, + }); + + const { signals, cursor } = await poll({ + om_waiting: { chatId: "oc_eng", messageId: "om_waiting", last: String(now - 60_000) }, + om_unanswered: { chatId: "oc_eng", messageId: "om_unanswered", last: String(now - 2 * day) }, + omt_old: { chatId: "oc_eng", messageId: "om_old", threadId: "omt_old", last: String(now - 8 * day) }, + }); + + expect(signals).toEqual([]); + expect((cursor as { threads: unknown }).threads).toEqual({ + om_waiting: { chatId: "oc_eng", messageId: "om_waiting", last: String(now - 60_000) }, + }); + }); + + it("stops reading a thread that is gone, without failing the look", async () => { + vi.useFakeTimers({ now: 1700000100000, toFake: ["Date"] }); + givenFeishu({ chats: [] }); + + const { cursor } = await poll({ + om_deleted: { chatId: "oc_eng", messageId: "om_deleted", last: "1700000000000" }, + }); + + expect((cursor as { threads: unknown }).threads).toEqual({}); }); }); @@ -225,6 +345,7 @@ describe("feishu reply", () => { expect(sent.find((call) => call.url.endsWith("/reply"))?.body).toEqual({ msg_type: "text", content: "{\"text\":\"Cause: redis timeout\"}", + reply_in_thread: true, }); expect(outcome.url).toBe("https://applink.feishu.cn/client/chat/open?openChatId=oc_eng"); }); diff --git a/src/connectors/feishu/runtime.ts b/src/connectors/feishu/runtime.ts index b48c5fd..22c4bd5 100644 --- a/src/connectors/feishu/runtime.ts +++ b/src/connectors/feishu/runtime.ts @@ -1,4 +1,5 @@ import type { JsonValue } from "#/lib/domain.ts"; +import { isTransient } from "../errors.ts"; import type { ActionSource, ConnectorRuntime, Signal, WorkItem } from "../types.ts"; import { botInfo, @@ -9,16 +10,39 @@ import { replyMessage, sendMessage, tenantAccessToken, + threadHistory, + threadOf, type FeishuCredential, + type FeishuMessage, } from "./api.ts"; +/** What was said in the thread before the bot was asked, oldest first. */ +type Said = { by: "bot" | "person"; text: string }; + type Asked = { chatId: string; messageId: string; user?: string; text: string; + earlier?: Said[]; }; +/** + * A thread the bot answers in, and how far into it this loop has read. Until + * the first answer is posted there is no thread yet, only the message it will + * hang from. + */ +type Watch = { chatId: string; messageId: string; threadId?: string; last: string }; + +/** A thread nobody has written in for this long is no longer read. */ +const WATCH_MS = 7 * 24 * 60 * 60 * 1000; +/** An ask that never got a thread was answered somewhere else, or not at all. */ +const PENDING_MS = 24 * 60 * 60 * 1000; +/** Every thread costs a request on every look, so only the liveliest are kept. */ +const WATCH_LIMIT = 50; +const EARLIER_LIMIT = 20; +const SAID_LIMIT = 2000; + function asAsked(value: JsonValue | null): Asked { const record = value as Partial | null; if (!record || typeof record.chatId !== "string" || typeof record.messageId !== "string") { @@ -29,9 +53,29 @@ function asAsked(value: JsonValue | null): Asked { messageId: record.messageId, user: typeof record.user === "string" ? record.user : undefined, text: typeof record.text === "string" ? record.text : "", + ...(Array.isArray(record.earlier) ? { earlier: record.earlier } : {}), }; } +function asWatches(value: JsonValue | null): Record { + const record = value && typeof value === "object" && !Array.isArray(value) ? value.threads : null; + if (!record || typeof record !== "object" || Array.isArray(record)) return {}; + const out: Record = {}; + for (const [key, entry] of Object.entries(record)) { + const watch = entry as Partial | null; + if (!watch || typeof watch.chatId !== "string" || typeof watch.messageId !== "string") continue; + if (typeof watch.last !== "string") continue; + out[key] = { + chatId: watch.chatId, + messageId: watch.messageId, + last: watch.last, + ...(typeof watch.threadId === "string" ? { threadId: watch.threadId } : {}), + }; + } + return out; +} + +/** Chats by id to the newest message read in each. Threads are kept apart. */ function asCursor(value: JsonValue | null): Record { if (!value || typeof value !== "object" || Array.isArray(value)) return {}; const out: Record = {}; @@ -57,6 +101,103 @@ function isAsk(input: { return input.mentionIds.includes(input.botOpenId); } +function signalFor(account: FeishuCredential, message: FeishuMessage, earlier?: Said[]): Signal { + const asked: Asked = { + chatId: message.chatId, + messageId: message.id, + user: message.senderId, + text: message.text, + ...(earlier?.length ? { earlier } : {}), + }; + return { + key: `${message.chatId}:${message.id}`, + kind: "message", + ref: message.id, + title: summarise(message.text), + url: permalink(account.host, message.chatId), + payload: asked as unknown as JsonValue, + at: Number(message.createTime), + }; +} + +function saidBy(account: FeishuCredential, message: FeishuMessage): Said { + const bot = message.senderType === "app" || message.senderId === account.botOpenId; + const text = message.text.length > SAID_LIMIT ? `${message.text.slice(0, SAID_LIMIT - 1)}…` : message.text; + return { by: bot ? "bot" : "person", text }; +} + +function asksIn(account: FeishuCredential, messages: FeishuMessage[], after: string): FeishuMessage[] { + return messages.filter( + (message) => + Number(message.createTime) > Number(after) && + isAsk({ + senderId: message.senderId, + senderType: message.senderType, + mentionIds: message.mentionIds, + botOpenId: account.botOpenId, + }), + ); +} + +/** + * Reads the threads the bot answers in, for the mentions an ordinary group's + * history leaves out. Returns the threads still worth reading. + */ +async function readThreads( + account: FeishuCredential, + watches: Record, + signals: Signal[], +): Promise> { + const now = Date.now(); + const kept: Record = {}; + for (const watch of Object.values(watches)) { + const idle = now - Number(watch.last); + if (idle > WATCH_MS || (!watch.threadId && idle > PENDING_MS)) continue; + try { + const threadId = watch.threadId ?? (await threadOf(account, watch.messageId)); + if (!threadId) { + kept[watch.messageId] = watch; + continue; + } + const thread = { threadId, chatId: watch.chatId }; + const messages = await threadHistory(account, thread, watch.last); + const asks = asksIn(account, messages, watch.last); + // Someone asking in a thread is carrying on a conversation, so the agent + // gets what came before it. Read once, only when there is an ask. + const whole = asks.length > 0 ? await threadHistory(account, thread, "0") : []; + for (const ask of asks) { + const before = whole.filter((message) => Number(message.createTime) < Number(ask.createTime)); + signals.push( + signalFor( + account, + ask, + before.slice(-EARLIER_LIMIT).map((message) => saidBy(account, message)), + ), + ); + } + const newest = messages.reduce( + (latest, message) => (Number(message.createTime) > Number(latest) ? message.createTime : latest), + watch.last, + ); + const existing = kept[threadId]; + kept[threadId] = { + ...watch, + threadId, + last: existing && Number(existing.last) > Number(newest) ? existing.last : newest, + }; + } catch (error) { + if (isTransient(error)) throw error; + // Deleted, or the bot was taken out of the group: nothing left to read. + } + } + return kept; +} + +function newest(watches: Record): Record { + const entries = Object.entries(watches).sort(([, a], [, b]) => Number(b.last) - Number(a.last)); + return Object.fromEntries(entries.slice(0, WATCH_LIMIT)); +} + function destination( target: Record, source: ActionSource, @@ -114,6 +255,9 @@ export const feishuRuntime: ConnectorRuntime = { const chats = await listChats(account); const next: Record = { ...previous }; const signals: Signal[] = []; + // Answers go in threads, and a group's history does not show what is + // said in them, so the threads are read on their own. + const watches = await readThreads(account, asWatches(cursor), signals); for (const chat of chats) { // A group seen for the first time is read from when the loop last @@ -125,37 +269,21 @@ export const feishuRuntime: ConnectorRuntime = { continue; } next[chat.id] = messages[messages.length - 1]!.createTime; - for (const message of messages) { - if (Number(message.createTime) <= Number(oldest)) continue; - if ( - !isAsk({ - senderId: message.senderId, - senderType: message.senderType, - mentionIds: message.mentionIds, - botOpenId: account.botOpenId, - }) - ) { - continue; - } - const asked: Asked = { + for (const message of asksIn(account, messages, oldest)) { + signals.push(signalFor(account, message)); + // The answer will open a thread under it, which is read from here on. + const key = message.threadId ?? message.id; + if (watches[key]) continue; + watches[key] = { chatId: message.chatId, messageId: message.id, - user: message.senderId, - text: message.text, + last: message.createTime, + ...(message.threadId ? { threadId: message.threadId } : {}), }; - signals.push({ - key: `${message.chatId}:${message.id}`, - kind: "message", - ref: message.id, - title: summarise(message.text), - url: permalink(account.host, message.chatId), - payload: asked as unknown as JsonValue, - at: Number(message.createTime), - }); } } - return { signals, cursor: next }; + return { signals, cursor: { ...next, threads: newest(watches) } as unknown as JsonValue }; }, async resolveWorkItem({ payload }): Promise { @@ -169,7 +297,22 @@ export const feishuRuntime: ConnectorRuntime = { context: [ { name: "MESSAGE.md", - body: [`# What you were asked`, ``, asked.text, ``].join("\n"), + body: [ + ...(asked.earlier?.length + ? [ + `# Earlier in this thread`, + ``, + ...asked.earlier.flatMap((said) => [ + `**${said.by === "bot" ? "You" : "Them"}:** ${said.text}`, + ``, + ]), + ] + : []), + `# What you were asked`, + ``, + asked.text, + ``, + ].join("\n"), }, ], };