From 7fec19dde3a9d85c97093ce160ed94468e3ca94b Mon Sep 17 00:00:00 2001 From: Psychomet Date: Wed, 2 Sep 2026 08:26:03 +0330 Subject: [PATCH 1/5] feat(telegram): add Business mode support Handle business_connection, business_message, and pass business_connection_id on outbound API calls. Closes #887 Signed-off-by: Psychomet Co-authored-by: Cursor --- .changeset/telegram-business-mode.md | 5 + .../content/adapters/official/telegram.mdx | 20 ++ packages/adapter-telegram/README.md | 31 ++ packages/adapter-telegram/sample-messages.md | 41 +++ packages/adapter-telegram/src/index.test.ts | 298 ++++++++++++++++++ packages/adapter-telegram/src/index.ts | 281 ++++++++++++++--- packages/adapter-telegram/src/types.ts | 33 ++ 7 files changed, 666 insertions(+), 43 deletions(-) create mode 100644 .changeset/telegram-business-mode.md diff --git a/.changeset/telegram-business-mode.md b/.changeset/telegram-business-mode.md new file mode 100644 index 00000000..85250f35 --- /dev/null +++ b/.changeset/telegram-business-mode.md @@ -0,0 +1,5 @@ +--- +"@chat-adapter/telegram": minor +--- + +Add Telegram Business mode support (`business_connection`, `business_message`, and `business_connection_id` on outbound API calls). Opt in via `businessMode: true`. diff --git a/apps/docs/content/adapters/official/telegram.mdx b/apps/docs/content/adapters/official/telegram.mdx index 17ba3e6a..b37b4a08 100644 --- a/apps/docs/content/adapters/official/telegram.mdx +++ b/apps/docs/content/adapters/official/telegram.mdx @@ -163,6 +163,12 @@ const telegram = createTelegramAdapter({ description: "Long polling tuning. Fields: `timeout`, `limit`, `allowedUpdates`, `deleteWebhook`, `dropPendingUpdates`, `retryDelayMs`.", }, + businessMode: { + type: "boolean", + default: "false", + description: + "Enable Telegram Business mode for Connected Business Bots. Handles `business_connection` and `business_message` updates and passes `business_connection_id` on outbound API calls.", + }, mentionOnReply: { type: "boolean", default: "false", @@ -196,6 +202,20 @@ const telegram = createTelegramAdapter({ `botToken` is always required. Webhook mode also requires `secretToken` unless `allowUnverifiedWebhooks` is explicitly enabled. Polling mode does not require webhook verification. +## Business mode + +Telegram [Connected Business Bots](https://core.telegram.org/api/bots/connected-business-bots) let a bot reply to customer messages on behalf of a business account. Enable it with `businessMode: true`: + +```typescript title="lib/bot.ts" lineNumbers +const telegram = createTelegramAdapter({ + businessMode: true, +}); +``` + +Business threads use the ID format `telegram:biz:{connectionId}:{chatId}`. Outbound sends automatically include `business_connection_id`. + +When configuring webhooks manually, add business update types to `allowed_updates` (or rely on the adapter's polling merge when `longPolling.allowedUpdates` is set). + ## Authentication Create a bot via [BotFather](https://t.me/BotFather): diff --git a/packages/adapter-telegram/README.md b/packages/adapter-telegram/README.md index 4c6d467c..7161b7f9 100644 --- a/packages/adapter-telegram/README.md +++ b/packages/adapter-telegram/README.md @@ -91,6 +91,36 @@ curl -X POST "https://api.telegram.org/bot$TELEGRAM_BOT_TOKEN/setWebhook" \ }' ``` +## Business mode (Connected Business Bots) + +Enable Telegram Business mode when your bot manages customer conversations on behalf of a connected business account: + +```typescript +const telegram = createTelegramAdapter({ + businessMode: true, +}); +``` + +With `businessMode: true`, the adapter: + +- Handles `business_connection`, `business_message`, and `edited_business_message` updates +- Skips messages sent by the business owner and connections without `can_reply` +- Encodes business threads as `telegram:biz:{connectionId}:{chatId}` +- Passes `business_connection_id` on outbound `sendMessage`, edits, typing, and file uploads + +When registering a webhook or starting polling with a custom `allowedUpdates` list, include the business update types (the adapter merges them automatically for polling when `businessMode` is `true`): + +```json +[ + "message", + "business_connection", + "business_message", + "edited_business_message" +] +``` + +Defaults to `false` for backward compatibility. + ## Polling (local development) When developing locally you typically can't expose a public URL for Telegram to deliver webhooks to. Polling mode uses `getUpdates` to fetch messages directly from Telegram instead — no public endpoint needed. @@ -161,6 +191,7 @@ Most options are auto-detected from environment variables when not provided. `na | `secretToken` | Webhook* | Webhook secret token. Auto-detected from `TELEGRAM_WEBHOOK_SECRET_TOKEN` | | `mode` | No | Adapter mode: `auto` (default), `webhook`, or `polling` | | `longPolling` | No | Optional long polling config for `getUpdates` (`timeout`, `limit`, `allowedUpdates`, `deleteWebhook`, `dropPendingUpdates`, `retryDelayMs`) | +| `businessMode` | No | Enable Telegram Business mode (`business_connection`, `business_message`). Defaults to `false` | | `userName` | No | Bot username used for mention detection. Auto-detected from `TELEGRAM_BOT_USERNAME` or `getMe` | | `mentionOnReply` | No | Treat a reply to one of the bot's own messages as a mention, so it routes to `onNewMention`. Defaults to `false`. Auto-detected from `TELEGRAM_MENTION_ON_REPLY=true`. Implicit forum-topic replies and the bot's own messages never count | | `nativeStreaming` | No | Stream with Telegram's native draft previews in private chats. Defaults to `false`, which uses post-and-edit in every chat type | diff --git a/packages/adapter-telegram/sample-messages.md b/packages/adapter-telegram/sample-messages.md index 4e4f824e..d8682d2a 100644 --- a/packages/adapter-telegram/sample-messages.md +++ b/packages/adapter-telegram/sample-messages.md @@ -576,3 +576,44 @@ and the story id. } } ``` + +## Business message (Connected Business Bot) + +```json +{ + "update_id": 312744900, + "business_connection": { + "id": "conn-example-abc", + "user": { + "id": 100000010, + "is_bot": false, + "first_name": "Business Owner", + "username": "owner" + }, + "user_chat_id": 100000010, + "date": 1756290700, + "is_enabled": true, + "rights": { + "can_reply": true + } + }, + "business_message": { + "message_id": 42, + "from": { + "id": 100000001, + "is_bot": false, + "first_name": "Customer", + "username": "customer" + }, + "chat": { + "id": 100000001, + "first_name": "Customer", + "username": "customer", + "type": "private" + }, + "date": 1756290701, + "business_connection_id": "conn-example-abc", + "text": "Hi, I need help with my order" + } +} +``` diff --git a/packages/adapter-telegram/src/index.test.ts b/packages/adapter-telegram/src/index.test.ts index b543d1ab..a8604296 100644 --- a/packages/adapter-telegram/src/index.test.ts +++ b/packages/adapter-telegram/src/index.test.ts @@ -656,6 +656,14 @@ threadIdContract({ decoded: { chatId: "-100123", messageThreadId: 42 }, encoded: "telegram:-100123:42", }, + // Connected Business Bot thread. + { + decoded: { + chatId: "456", + businessConnectionId: "conn-abc", + }, + encoded: "telegram:biz:conn-abc:456", + }, ], isDM: { fn: (id) => threadIdAdapter.isDM(id), @@ -6825,3 +6833,293 @@ describe("non-file content", () => { expect(await parseContent({ story: { id: 7 } })).toBe("📖 Story"); }); }); + +// ============================================================================ +// Telegram Business mode +// ============================================================================ + +const sampleBusinessConnection = { + id: "conn-abc", + user: { + id: 111, + is_bot: false, + first_name: "Owner", + username: "owner", + }, + user_chat_id: 111, + date: 1_735_689_600, + is_enabled: true, + rights: { can_reply: true }, +}; + +function sampleBusinessMessage( + overrides?: Partial +): TelegramMessage { + return sampleMessage({ + business_connection_id: "conn-abc", + chat: { id: 456, type: "private", first_name: "Customer" }, + from: { + id: 456, + is_bot: false, + first_name: "Customer", + username: "customer", + }, + text: "Need help with my order", + ...overrides, + }); +} + +describe("Telegram Business mode", () => { + const awaitWebhookTasks = async ( + tasks: Promise[] + ): Promise => { + await Promise.all(tasks); + }; + + it("routes business_message updates to processMessage when businessMode is enabled", async () => { + mockFetch + .mockResolvedValueOnce( + telegramOk({ + id: 999, + is_bot: true, + first_name: "Bot", + username: "mybot", + }) + ) + .mockResolvedValue(telegramOk(true)); + + const adapter = createTelegramAdapter({ + allowUnverifiedWebhooks: true, + botToken: "token", + businessMode: true, + mode: "webhook", + logger: mockLogger, + userName: "mybot", + }); + const chat = createMockChat(); + await adapter.initialize(chat); + + const pendingTasks: Promise[] = []; + const response = await adapter.handleWebhook( + new Request("https://example.com/webhooks/telegram", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + update_id: 1, + business_connection: sampleBusinessConnection, + business_message: sampleBusinessMessage(), + }), + }), + { + waitUntil: (task: Promise) => { + pendingTasks.push(task); + }, + } + ); + + await awaitWebhookTasks(pendingTasks); + + expect(response.status).toBe(200); + const processMessage = chat.processMessage as ReturnType; + expect(processMessage).toHaveBeenCalledTimes(1); + const [, threadId] = processMessage.mock.calls[0] as [unknown, string]; + expect(threadId).toBe("telegram:biz:conn-abc:456"); + }); + + it("skips business messages sent by the business owner", async () => { + mockFetch + .mockResolvedValueOnce( + telegramOk({ + id: 999, + is_bot: true, + first_name: "Bot", + username: "mybot", + }) + ) + .mockResolvedValue(telegramOk(true)); + + const adapter = createTelegramAdapter({ + allowUnverifiedWebhooks: true, + botToken: "token", + businessMode: true, + mode: "webhook", + logger: mockLogger, + userName: "mybot", + }); + const chat = createMockChat(); + await adapter.initialize(chat); + + const pendingTasks: Promise[] = []; + await adapter.handleWebhook( + new Request("https://example.com/webhooks/telegram", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + update_id: 1, + business_connection: sampleBusinessConnection, + business_message: sampleBusinessMessage({ + from: sampleBusinessConnection.user, + }), + }), + }), + { + waitUntil: (task: Promise) => { + pendingTasks.push(task); + }, + } + ); + + await awaitWebhookTasks(pendingTasks); + + expect(chat.processMessage).not.toHaveBeenCalled(); + }); + + it("skips business messages when can_reply is false", async () => { + mockFetch + .mockResolvedValueOnce( + telegramOk({ + id: 999, + is_bot: true, + first_name: "Bot", + username: "mybot", + }) + ) + .mockResolvedValue(telegramOk(true)); + + const adapter = createTelegramAdapter({ + allowUnverifiedWebhooks: true, + botToken: "token", + businessMode: true, + mode: "webhook", + logger: mockLogger, + userName: "mybot", + }); + const chat = createMockChat(); + await adapter.initialize(chat); + + const pendingTasks: Promise[] = []; + await adapter.handleWebhook( + new Request("https://example.com/webhooks/telegram", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + update_id: 1, + business_connection: { + ...sampleBusinessConnection, + rights: { can_reply: false }, + }, + business_message: sampleBusinessMessage(), + }), + }), + { + waitUntil: (task: Promise) => { + pendingTasks.push(task); + }, + } + ); + + await awaitWebhookTasks(pendingTasks); + + expect(chat.processMessage).not.toHaveBeenCalled(); + }); + + it("ignores business updates when businessMode is false", async () => { + mockFetch.mockResolvedValueOnce( + telegramOk({ + id: 999, + is_bot: true, + first_name: "Bot", + username: "mybot", + }) + ); + + const adapter = createTelegramAdapter({ + allowUnverifiedWebhooks: true, + botToken: "token", + mode: "webhook", + logger: mockLogger, + userName: "mybot", + }); + const chat = createMockChat(); + await adapter.initialize(chat); + + await adapter.handleWebhook( + new Request("https://example.com/webhooks/telegram", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + update_id: 1, + business_message: sampleBusinessMessage(), + }), + }) + ); + + expect(chat.processMessage).not.toHaveBeenCalled(); + }); + + it("includes business_connection_id on outbound post() for business threads", async () => { + mockFetch + .mockResolvedValueOnce( + telegramOk({ + id: 999, + is_bot: true, + first_name: "Bot", + username: "mybot", + }) + ) + .mockResolvedValueOnce( + telegramOk({ + message_id: 77, + date: 1, + chat: { id: 456, type: "private", first_name: "Customer" }, + text: "reply", + }) + ); + + const adapter = createTelegramAdapter({ + botToken: "token", + mode: "webhook", + logger: mockLogger, + userName: "mybot", + }); + await adapter.initialize(createMockChat()); + + await adapter.postMessage("telegram:biz:conn-abc:456", "reply"); + + const sendMessageBody = JSON.parse( + String((mockFetch.mock.calls[1]?.[1] as RequestInit).body) + ) as { + business_connection_id?: string; + chat_id: string; + }; + expect(sendMessageBody.chat_id).toBe("456"); + expect(sendMessageBody.business_connection_id).toBe("conn-abc"); + }); + + it("merges business update types into custom polling allowedUpdates", () => { + class PollingTestAdapter extends TelegramAdapter { + resolvePolling( + override?: Parameters[0] + ) { + return this.resolvePollingConfig(override); + } + } + + const pollingAdapter = new PollingTestAdapter({ + botToken: "token", + businessMode: true, + longPolling: { + allowedUpdates: ["message"], + }, + mode: "polling", + logger: mockLogger, + }); + + expect(pollingAdapter.resolvePolling()?.allowedUpdates).toEqual([ + "message", + "business_connection", + "business_message", + "edited_business_message", + ]); + }); +}); diff --git a/packages/adapter-telegram/src/index.ts b/packages/adapter-telegram/src/index.ts index 80442210..926ef3ce 100644 --- a/packages/adapter-telegram/src/index.ts +++ b/packages/adapter-telegram/src/index.ts @@ -65,6 +65,7 @@ import type { TelegramAdapterConfig, TelegramAdapterMode, TelegramApiResponse, + TelegramBusinessConnection, TelegramCallbackQuery, TelegramChat, TelegramFile, @@ -188,6 +189,11 @@ const TELEGRAM_MAX_POLLING_LIMIT = 100; const TELEGRAM_MIN_POLLING_LIMIT = 1; const TELEGRAM_MIN_POLLING_TIMEOUT_SECONDS = 0; const TELEGRAM_MAX_POLLING_TIMEOUT_SECONDS = 300; +const TELEGRAM_BUSINESS_ALLOWED_UPDATES = [ + "business_connection", + "business_message", + "edited_business_message", +] as const; function normalizeBotTokenProvider( botToken: string | (() => string | Promise) @@ -450,6 +456,11 @@ export class TelegramAdapter protected readonly nativeStreaming: boolean; protected readonly streamingEditIntervalMs?: number; protected readonly longPolling?: TelegramLongPollingConfig; + protected readonly businessMode: boolean; + private readonly businessConnectionCache = new Map< + string, + TelegramBusinessConnection + >(); private _runtimeMode: TelegramRuntimeMode = "webhook"; private pollingAbortController: AbortController | null = null; private pollingTask: Promise | null = null; @@ -522,6 +533,7 @@ export class TelegramAdapter Number.MAX_SAFE_INTEGER ); this.longPolling = config.longPolling; + this.businessMode = config.businessMode ?? false; if (!["auto", "webhook", "polling"].includes(this.mode)) { throw new ValidationError( @@ -916,6 +928,111 @@ export class TelegramAdapter if (update.message_reaction) { this.handleMessageReactionUpdate(update.message_reaction, options); } + + if (this.businessMode) { + if (update.business_connection) { + this.cacheBusinessConnection(update.business_connection); + } + + const businessMessage = + update.business_message ?? update.edited_business_message; + if (businessMessage) { + const task = this.handleBusinessMessageUpdate( + businessMessage, + businessMessage.business_connection_id, + options + ).catch((error) => { + this.logger.warn("Failed to process Telegram business message", { + error: String(error), + connectionId: businessMessage.business_connection_id, + }); + }); + options?.waitUntil?.(task); + } + } + } + + protected cacheBusinessConnection( + connection: TelegramBusinessConnection + ): void { + this.businessConnectionCache.set(connection.id, connection); + } + + protected async resolveBusinessConnection( + connectionId: string + ): Promise { + const cached = this.businessConnectionCache.get(connectionId); + if (cached) { + return cached; + } + + const connection = await this.telegramFetch( + "getBusinessConnection", + { + business_connection_id: connectionId, + } + ); + this.businessConnectionCache.set(connectionId, connection); + return connection; + } + + protected isBusinessConnectionUsable( + connection: TelegramBusinessConnection + ): boolean { + return ( + connection.is_enabled && (connection.rights?.can_reply ?? false) + ); + } + + protected async handleBusinessMessageUpdate( + telegramMessage: TelegramMessage, + connectionId: string | undefined, + options?: WebhookOptions + ): Promise { + if (!this.chat) { + return; + } + + const resolvedConnectionId = + connectionId ?? telegramMessage.business_connection_id; + if (!resolvedConnectionId) { + this.logger.warn("Telegram business message missing connection id", { + messageId: telegramMessage.message_id, + }); + return; + } + + const connection = + await this.resolveBusinessConnection(resolvedConnectionId); + if (!this.isBusinessConnectionUsable(connection)) { + return; + } + + if (telegramMessage.from?.id === connection.user.id) { + return; + } + + const threadId = this.encodeThreadId({ + chatId: String(telegramMessage.chat.id), + businessConnectionId: resolvedConnectionId, + }); + + if (telegramMessage.media_group_id) { + await this.processIncomingMediaGroup(telegramMessage, threadId); + return; + } + + this.startTypingForPrivateMessage(telegramMessage, threadId, options); + + const parsedMessage = this.parseTelegramMessage(telegramMessage, threadId); + this.cacheMessage(parsedMessage); + + await this.chat.processMessage( + this, + threadId, + parsedMessage, + options + ); } protected handleIncomingMessageUpdate( @@ -1410,8 +1527,7 @@ export class TelegramAdapter ? await this.withTelegramRichFallback( () => this.telegramFetch("sendRichMessage", { - chat_id: parsedThread.chatId, - message_thread_id: parsedThread.messageThreadId, + ...this.buildChatTargetParams(parsedThread), rich_message: { markdown: rich.markdown, }, @@ -1435,6 +1551,7 @@ export class TelegramAdapter chatId: String(rawMessage.chat.id), messageThreadId: rawMessage.message_thread_id ?? parsedThread.messageThreadId, + businessConnectionId: parsedThread.businessConnectionId, }), rich ? { @@ -1533,7 +1650,7 @@ export class TelegramAdapter parseMode, (resolvedParseMode, resolvedText) => this.telegramFetch("editMessageText", { - chat_id: chatId, + ...this.buildChatTargetParams(parsedThread), message_id: telegramMessageId, text: resolvedText, reply_markup: replyMarkup ?? emptyTelegramInlineKeyboard(), @@ -1552,7 +1669,7 @@ export class TelegramAdapter ? await this.withTelegramRichFallback( () => this.telegramFetch("editMessageText", { - chat_id: chatId, + ...this.buildChatTargetParams(parsedThread), message_id: telegramMessageId, rich_message: { markdown: rich.markdown, @@ -1600,6 +1717,7 @@ export class TelegramAdapter const resultingThreadId = this.encodeThreadId({ chatId: String(result.chat.id), messageThreadId: result.message_thread_id ?? parsedThread.messageThreadId, + businessConnectionId: parsedThread.businessConnectionId, }); const parsedMessage = this.parseTelegramMessage( @@ -1630,7 +1748,7 @@ export class TelegramAdapter } = this.decodeCompositeMessageId(messageId, parsedThread.chatId); await this.telegramFetch("deleteMessage", { - chat_id: chatId, + ...this.buildChatTargetParams(parsedThread), message_id: telegramMessageId, }); @@ -1647,7 +1765,7 @@ export class TelegramAdapter this.decodeCompositeMessageId(messageId, parsedThread.chatId); await this.telegramFetch("setMessageReaction", { - chat_id: chatId, + ...this.buildChatTargetParams(parsedThread), message_id: telegramMessageId, reaction: [this.toTelegramReaction(emoji)], }); @@ -1663,7 +1781,7 @@ export class TelegramAdapter this.decodeCompositeMessageId(messageId, parsedThread.chatId); await this.telegramFetch("setMessageReaction", { - chat_id: chatId, + ...this.buildChatTargetParams(parsedThread), message_id: telegramMessageId, reaction: [], }); @@ -1672,8 +1790,7 @@ export class TelegramAdapter async startTyping(threadId: string): Promise { const parsedThread = this.resolveThreadId(threadId); await this.telegramFetch("sendChatAction", { - chat_id: parsedThread.chatId, - message_thread_id: parsedThread.messageThreadId, + ...this.buildChatTargetParams(parsedThread), action: "typing", }); } @@ -1921,8 +2038,7 @@ export class TelegramAdapter if (streamUsesRich) { try { await this.telegramFetch("sendRichMessageDraft", { - chat_id: parsedThread.chatId, - message_thread_id: parsedThread.messageThreadId, + ...this.buildChatTargetParams(parsedThread), draft_id: draftId, rich_message: { markdown: text, @@ -1959,16 +2075,14 @@ export class TelegramAdapter try { if (useMarkdown) { await this.telegramFetch("sendMessageDraft", { - chat_id: parsedThread.chatId, - message_thread_id: parsedThread.messageThreadId, + ...this.buildChatTargetParams(parsedThread), draft_id: draftId, text: draftText, parse_mode: toBotApiParseMode("MarkdownV2"), }); } else { await this.telegramFetch("sendMessageDraft", { - chat_id: parsedThread.chatId, - message_thread_id: parsedThread.messageThreadId, + ...this.buildChatTargetParams(parsedThread), draft_id: draftId, text: draftText, }); @@ -1983,8 +2097,7 @@ export class TelegramAdapter try { await this.telegramFetch("sendMessageDraft", { - chat_id: parsedThread.chatId, - message_thread_id: parsedThread.messageThreadId, + ...this.buildChatTargetParams(parsedThread), draft_id: draftId, text: plainDraftText, }); @@ -2058,8 +2171,7 @@ export class TelegramAdapter const raw = await this.telegramFetch( "sendRichMessage", { - chat_id: parsedThread.chatId, - message_thread_id: parsedThread.messageThreadId, + ...this.buildChatTargetParams(parsedThread), rich_message: { markdown, }, @@ -2069,6 +2181,7 @@ export class TelegramAdapter chatId: String(raw.chat.id), messageThreadId: raw.message_thread_id ?? parsedThread.messageThreadId, + businessConnectionId: parsedThread.businessConnectionId, }); const formatted = this.formatConverter.toAst(accumulated); const message = this.parseTelegramMessage(raw, resultingThreadId, { @@ -2104,6 +2217,7 @@ export class TelegramAdapter const resultingThreadId = this.encodeThreadId({ chatId: String(raw.chat.id), messageThreadId: raw.message_thread_id ?? parsedThread.messageThreadId, + businessConnectionId: parsedThread.businessConnectionId, }); const message = this.parseTelegramMessage(raw, resultingThreadId); this.cacheMessage(message); @@ -2220,6 +2334,10 @@ export class TelegramAdapter } encodeThreadId(platformData: TelegramThreadId): string { + if (platformData.businessConnectionId) { + return `telegram:biz:${platformData.businessConnectionId}:${platformData.chatId}`; + } + if (typeof platformData.messageThreadId === "number") { return `telegram:${platformData.chatId}:${platformData.messageThreadId}`; } @@ -2229,7 +2347,37 @@ export class TelegramAdapter decodeThreadId(threadId: string): TelegramThreadId { const parts = threadId.split(":"); - if (parts[0] !== "telegram" || parts.length < 2 || parts.length > 3) { + if (parts[0] !== "telegram") { + throw new ValidationError( + "telegram", + `Invalid Telegram thread ID: ${threadId}` + ); + } + + if (parts[1] === "biz") { + if (parts.length !== 4) { + throw new ValidationError( + "telegram", + `Invalid Telegram business thread ID: ${threadId}` + ); + } + + const businessConnectionId = parts[2]; + const chatId = parts[3]; + if (!(businessConnectionId && chatId)) { + throw new ValidationError( + "telegram", + `Invalid Telegram business thread ID: ${threadId}` + ); + } + + return { + chatId, + businessConnectionId, + }; + } + + if (parts.length < 2 || parts.length > 3) { throw new ValidationError( "telegram", `Invalid Telegram thread ID: ${threadId}` @@ -2613,11 +2761,8 @@ export class TelegramAdapter replyParameters?: TelegramReplyParameters ): FormData { const formData = new FormData(); - formData.append("chat_id", thread.chatId); + this.appendChatTargetToFormData(formData, thread); this.appendReplyParameters(formData, replyParameters); - if (typeof thread.messageThreadId === "number") { - formData.append("message_thread_id", String(thread.messageThreadId)); - } if (text.trim()) { formData.append( @@ -2669,15 +2814,11 @@ export class TelegramAdapter (resolvedParseMode, resolvedText) => { if (!buffer) { const payload: Record = { - chat_id: thread.chatId, + ...this.buildChatTargetParams(thread), [upload.field]: attachment.url, reply_parameters: replyParameters, }; - if (typeof thread.messageThreadId === "number") { - payload.message_thread_id = thread.messageThreadId; - } - if (resolvedText.trim()) { payload.caption = truncateForTelegram( resolvedText, @@ -2708,10 +2849,7 @@ export class TelegramAdapter const formData = new FormData(); - formData.append("chat_id", thread.chatId); - if (typeof thread.messageThreadId === "number") { - formData.append("message_thread_id", String(thread.messageThreadId)); - } + this.appendChatTargetToFormData(formData, thread); if (resolvedText.trim()) { formData.append( @@ -2890,11 +3028,8 @@ export class TelegramAdapter replyParameters?: TelegramReplyParameters ): FormData { const formData = new FormData(); - formData.append("chat_id", thread.chatId); + this.appendChatTargetToFormData(formData, thread); this.appendReplyParameters(formData, replyParameters); - if (typeof thread.messageThreadId === "number") { - formData.append("message_thread_id", String(thread.messageThreadId)); - } // Telegram shows an album's caption from the first item that carries one; // on a later item it renders under that media instead of the whole group. @@ -3101,6 +3236,50 @@ export class TelegramAdapter return { chatId: value }; } + /** + * Bot API chat-target fields (`chat_id`, `message_thread_id`, + * `business_connection_id`) derived from a decoded thread. + */ + protected buildChatTargetParams(thread: TelegramThreadId): { + business_connection_id?: string; + chat_id: string; + message_thread_id?: number; + } { + const params: { + business_connection_id?: string; + chat_id: string; + message_thread_id?: number; + } = { + chat_id: thread.chatId, + }; + + if (typeof thread.messageThreadId === "number") { + params.message_thread_id = thread.messageThreadId; + } + + if (thread.businessConnectionId) { + params.business_connection_id = thread.businessConnectionId; + } + + return params; + } + + protected appendChatTargetToFormData( + formData: FormData, + thread: TelegramThreadId + ): void { + formData.append("chat_id", thread.chatId); + if (typeof thread.messageThreadId === "number") { + formData.append("message_thread_id", String(thread.messageThreadId)); + } + if (thread.businessConnectionId) { + formData.append( + "business_connection_id", + thread.businessConnectionId + ); + } + } + protected encodeMessageId(chatId: string, messageId: number): string { return `${chatId}:${messageId}`; } @@ -3427,8 +3606,7 @@ export class TelegramAdapter parseMode, (resolvedParseMode, resolvedText) => this.telegramFetch("sendMessage", { - chat_id: thread.chatId, - message_thread_id: thread.messageThreadId, + ...this.buildChatTargetParams(thread), text: resolvedText, reply_markup: replyMarkup, parse_mode: toBotApiParseMode(resolvedParseMode), @@ -3628,10 +3806,7 @@ export class TelegramAdapter }; return { - allowedUpdates: - merged.allowedUpdates && merged.allowedUpdates.length > 0 - ? [...merged.allowedUpdates] - : undefined, + allowedUpdates: this.resolveAllowedUpdates(merged.allowedUpdates), deleteWebhook: merged.deleteWebhook ?? true, dropPendingUpdates: merged.dropPendingUpdates ?? false, limit: this.clampInteger( @@ -3655,6 +3830,24 @@ export class TelegramAdapter }; } + protected resolveAllowedUpdates( + allowedUpdates?: string[] + ): string[] | undefined { + if (!this.businessMode) { + return allowedUpdates && allowedUpdates.length > 0 + ? [...allowedUpdates] + : undefined; + } + + if (!allowedUpdates || allowedUpdates.length === 0) { + return undefined; + } + + return [ + ...new Set([...allowedUpdates, ...TELEGRAM_BUSINESS_ALLOWED_UPDATES]), + ]; + } + protected clampInteger( value: number | undefined, fallback: number, @@ -3876,6 +4069,8 @@ export type { TelegramAdapterMode, TelegramAnimation, TelegramAudio, + TelegramBusinessConnection, + TelegramBusinessBotRights, TelegramCallbackQuery, TelegramChat, TelegramLocation, diff --git a/packages/adapter-telegram/src/types.ts b/packages/adapter-telegram/src/types.ts index adabbf46..d78cff98 100644 --- a/packages/adapter-telegram/src/types.ts +++ b/packages/adapter-telegram/src/types.ts @@ -58,6 +58,12 @@ export interface TelegramAdapterConfig { streamingEditIntervalMs?: number; /** Override bot username (optional). Defaults to TELEGRAM_BOT_USERNAME env var. */ userName?: string; + /** + * Enable Telegram Business mode (Connected Business Bots). When `true`, the + * adapter handles `business_connection` and `business_message` updates and + * passes `business_connection_id` on outbound API calls. Defaults to `false`. + */ + businessMode?: boolean; } export type TelegramAdapterMode = "auto" | "webhook" | "polling"; @@ -97,6 +103,8 @@ export interface TelegramThreadId { chatId: string; /** Optional forum topic ID for supergroup topics. */ messageThreadId?: number; + /** Business connection ID for Connected Business Bot threads. */ + businessConnectionId?: string; } /** @@ -495,6 +503,7 @@ export interface TelegramMessage { media_group_id?: string; message_id: number; message_thread_id?: number; + business_connection_id?: string; photo?: TelegramPhotoSize[]; poll?: { id: string; @@ -569,6 +578,27 @@ export type TelegramReactionType = type: "custom_emoji"; }; +/** + * Rights granted to a bot on a connected business account. + * @see https://core.telegram.org/bots/api#businessbotrights + */ +export interface TelegramBusinessBotRights { + can_reply?: boolean; +} + +/** + * Connected business account metadata. + * @see https://core.telegram.org/bots/api#businessconnection + */ +export interface TelegramBusinessConnection { + date: number; + id: string; + is_enabled: boolean; + rights?: TelegramBusinessBotRights; + user: TelegramUser; + user_chat_id: number; +} + /** * Telegram message reaction update. * @see https://core.telegram.org/bots/api#messagereactionupdated @@ -589,8 +619,11 @@ export interface TelegramMessageReactionUpdated { * @see https://core.telegram.org/bots/api#update */ export interface TelegramUpdate { + business_connection?: TelegramBusinessConnection; + business_message?: TelegramMessage; callback_query?: TelegramCallbackQuery; channel_post?: TelegramMessage; + edited_business_message?: TelegramMessage; edited_channel_post?: TelegramMessage; edited_message?: TelegramMessage; message?: TelegramMessage; From 79fa67c48b455312d70acb257412e182c6073555 Mon Sep 17 00:00:00 2001 From: Psychomet Date: Wed, 2 Sep 2026 09:02:36 +0330 Subject: [PATCH 2/5] fix(telegram): address Business mode PR review feedback Resolve customer userId for allowedUserIds on business_message updates. Omit business_connection_id from deleteMessage and setMessageReaction. Signed-off-by: Psychomet Co-authored-by: Cursor --- packages/adapter-telegram/src/index.test.ts | 81 +++++++++++++++++++++ packages/adapter-telegram/src/index.ts | 27 +++++-- 2 files changed, 101 insertions(+), 7 deletions(-) diff --git a/packages/adapter-telegram/src/index.test.ts b/packages/adapter-telegram/src/index.test.ts index a8604296..77ee9b0b 100644 --- a/packages/adapter-telegram/src/index.test.ts +++ b/packages/adapter-telegram/src/index.test.ts @@ -7096,6 +7096,87 @@ describe("Telegram Business mode", () => { expect(sendMessageBody.business_connection_id).toBe("conn-abc"); }); + it("routes business_message when customer is in allowedUserIds", async () => { + mockFetch + .mockResolvedValueOnce( + telegramOk({ + id: 999, + is_bot: true, + first_name: "Bot", + username: "mybot", + }) + ) + .mockResolvedValue(telegramOk(true)); + + const adapter = createTelegramAdapter({ + allowUnverifiedWebhooks: true, + allowedUserIds: [456], + botToken: "token", + businessMode: true, + mode: "webhook", + logger: mockLogger, + userName: "mybot", + }); + const chat = createMockChat(); + await adapter.initialize(chat); + + const pendingTasks: Promise[] = []; + await adapter.handleWebhook( + new Request("https://example.com/webhooks/telegram", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + update_id: 1, + business_connection: sampleBusinessConnection, + business_message: sampleBusinessMessage(), + }), + }), + { + waitUntil: (task: Promise) => { + pendingTasks.push(task); + }, + } + ); + + await awaitWebhookTasks(pendingTasks); + + expect(chat.processMessage).toHaveBeenCalledTimes(1); + }); + + it("omits business_connection_id on deleteMessage for business threads", async () => { + mockFetch + .mockResolvedValueOnce( + telegramOk({ + id: 999, + is_bot: true, + first_name: "Bot", + username: "mybot", + }) + ) + .mockResolvedValueOnce(telegramOk(true)); + + const adapter = createTelegramAdapter({ + botToken: "token", + mode: "webhook", + logger: mockLogger, + userName: "mybot", + }); + await adapter.initialize(createMockChat()); + + await adapter.deleteMessage("telegram:biz:conn-abc:456", "456:77"); + + const deleteMessageBody = JSON.parse( + String((mockFetch.mock.calls[1]?.[1] as RequestInit).body) + ) as { + business_connection_id?: string; + chat_id: string; + message_id: number; + }; + expect(deleteMessageBody.chat_id).toBe("456"); + expect(deleteMessageBody.message_id).toBe(77); + expect(deleteMessageBody.business_connection_id).toBeUndefined(); + }); + it("merges business update types into custom polling allowedUpdates", () => { class PollingTestAdapter extends TelegramAdapter { resolvePolling( diff --git a/packages/adapter-telegram/src/index.ts b/packages/adapter-telegram/src/index.ts index 926ef3ce..761e4ecf 100644 --- a/packages/adapter-telegram/src/index.ts +++ b/packages/adapter-telegram/src/index.ts @@ -900,10 +900,13 @@ export class TelegramAdapter update.edited_message ?? update.channel_post ?? update.edited_channel_post; + const businessMessageUpdate = + update.business_message ?? update.edited_business_message; const userId = update.callback_query?.from.id ?? update.message_reaction?.user?.id ?? - messageUpdate?.from?.id; + messageUpdate?.from?.id ?? + businessMessageUpdate?.from?.id; if ( this.allowedUserIds && @@ -1748,7 +1751,9 @@ export class TelegramAdapter } = this.decodeCompositeMessageId(messageId, parsedThread.chatId); await this.telegramFetch("deleteMessage", { - ...this.buildChatTargetParams(parsedThread), + ...this.buildChatTargetParams(parsedThread, { + includeBusinessConnection: false, + }), message_id: telegramMessageId, }); @@ -1765,7 +1770,9 @@ export class TelegramAdapter this.decodeCompositeMessageId(messageId, parsedThread.chatId); await this.telegramFetch("setMessageReaction", { - ...this.buildChatTargetParams(parsedThread), + ...this.buildChatTargetParams(parsedThread, { + includeBusinessConnection: false, + }), message_id: telegramMessageId, reaction: [this.toTelegramReaction(emoji)], }); @@ -1781,7 +1788,9 @@ export class TelegramAdapter this.decodeCompositeMessageId(messageId, parsedThread.chatId); await this.telegramFetch("setMessageReaction", { - ...this.buildChatTargetParams(parsedThread), + ...this.buildChatTargetParams(parsedThread, { + includeBusinessConnection: false, + }), message_id: telegramMessageId, reaction: [], }); @@ -3237,14 +3246,18 @@ export class TelegramAdapter } /** - * Bot API chat-target fields (`chat_id`, `message_thread_id`, + * Bot API chat-target fields (`chat_id`, `message_thread_id`, and optionally * `business_connection_id`) derived from a decoded thread. */ - protected buildChatTargetParams(thread: TelegramThreadId): { + protected buildChatTargetParams( + thread: TelegramThreadId, + options: { includeBusinessConnection?: boolean } = {} + ): { business_connection_id?: string; chat_id: string; message_thread_id?: number; } { + const includeBusinessConnection = options.includeBusinessConnection ?? true; const params: { business_connection_id?: string; chat_id: string; @@ -3257,7 +3270,7 @@ export class TelegramAdapter params.message_thread_id = thread.messageThreadId; } - if (thread.businessConnectionId) { + if (includeBusinessConnection && thread.businessConnectionId) { params.business_connection_id = thread.businessConnectionId; } From 53dc439fc56ad03ed133fe12f3b27af12cde93ac Mon Sep 17 00:00:00 2001 From: Ben Sabic Date: Fri, 4 Sep 2026 13:36:21 +1000 Subject: [PATCH 3/5] fix(telegram): route Business mode threads through the connection end to end Business threads are now their own channel, so channel posts keep business_connection_id and locks and history stay separate from the customer's direct bot chat. Callback queries, slash commands, and media groups in business chats resolve to the business thread. Deletes go through deleteBusinessMessages. Reactions throw a NotImplementedError since the Bot API has no business variant. fetchThread carries the connection id and falls back to cached chat details when getChat cannot resolve the customer. Connection state moves from a per-instance Map into the state adapter with a TTL, and is recorded before the allowedUserIds gate so rights changes always land. Outgoing echoes carrying sender_business_bot are skipped and attributed to the bot. Polling always sends an explicit allowed_updates list in Business mode. --- .changeset/telegram-business-mode.md | 4 +- packages/adapter-telegram/src/index.test.ts | 544 +++++++++++++++++++- packages/adapter-telegram/src/index.ts | 335 ++++++++---- packages/adapter-telegram/src/types.ts | 31 +- 4 files changed, 799 insertions(+), 115 deletions(-) diff --git a/.changeset/telegram-business-mode.md b/.changeset/telegram-business-mode.md index 85250f35..c575bc96 100644 --- a/.changeset/telegram-business-mode.md +++ b/.changeset/telegram-business-mode.md @@ -2,4 +2,6 @@ "@chat-adapter/telegram": minor --- -Add Telegram Business mode support (`business_connection`, `business_message`, and `business_connection_id` on outbound API calls). Opt in via `businessMode: true`. +Add Telegram Business mode support. Opt in via `businessMode: true`. + +The adapter handles `business_connection`, `business_message`, and `edited_business_message` updates, encodes business threads as `telegram:biz:{connectionId}:{chatId}`, and passes `business_connection_id` on outbound sends, edits, typing, file uploads, and inline-keyboard callbacks. Business threads are their own channel, slash commands route through `onSlashCommand`, deletes use `deleteBusinessMessages`, and connection state is cached in the state adapter so a revoked connection is honoured by every instance. Reactions on business threads throw a `NotImplementedError`, since the Bot API has no business variant of `setMessageReaction`. diff --git a/packages/adapter-telegram/src/index.test.ts b/packages/adapter-telegram/src/index.test.ts index 77ee9b0b..ce7119ad 100644 --- a/packages/adapter-telegram/src/index.test.ts +++ b/packages/adapter-telegram/src/index.test.ts @@ -31,6 +31,7 @@ import { } from "./markdown"; const mockFetch = vi.fn(); +const BUSINESS_REACTION_ERROR = /business account/; process.env.TELEGRAM_ALLOW_UNVERIFIED_WEBHOOKS = "true"; const SERVERLESS_ENV_KEYS = [ "VERCEL", @@ -7143,7 +7144,7 @@ describe("Telegram Business mode", () => { expect(chat.processMessage).toHaveBeenCalledTimes(1); }); - it("omits business_connection_id on deleteMessage for business threads", async () => { + it("deletes business messages through deleteBusinessMessages", async () => { mockFetch .mockResolvedValueOnce( telegramOk({ @@ -7165,16 +7166,545 @@ describe("Telegram Business mode", () => { await adapter.deleteMessage("telegram:biz:conn-abc:456", "456:77"); - const deleteMessageBody = JSON.parse( + expect(telegramMethods()).toEqual(["getMe", "deleteBusinessMessages"]); + const body = JSON.parse( String((mockFetch.mock.calls[1]?.[1] as RequestInit).body) ) as { business_connection_id?: string; - chat_id: string; - message_id: number; + chat_id?: string; + message_ids?: number[]; + }; + expect(body.business_connection_id).toBe("conn-abc"); + expect(body.message_ids).toEqual([77]); + expect(body.chat_id).toBeUndefined(); + }); + + it("rejects reactions on business threads with a descriptive error", async () => { + mockFetch.mockResolvedValueOnce( + telegramOk({ + id: 999, + is_bot: true, + first_name: "Bot", + username: "mybot", + }) + ); + + const adapter = createTelegramAdapter({ + botToken: "token", + mode: "webhook", + logger: mockLogger, + userName: "mybot", + }); + await adapter.initialize(createMockChat()); + + await expect( + adapter.addReaction("telegram:biz:conn-abc:456", "456:77", "thumbs_up") + ).rejects.toThrow(BUSINESS_REACTION_ERROR); + await expect( + adapter.removeReaction("telegram:biz:conn-abc:456", "456:77", "thumbs_up") + ).rejects.toThrow(BUSINESS_REACTION_ERROR); + expect(telegramMethods()).toEqual(["getMe"]); + }); + + it("keeps business threads as their own channel", () => { + const adapter = createTelegramAdapter({ + botToken: "token", + mode: "webhook", + logger: mockLogger, + userName: "mybot", + }); + + expect(adapter.channelIdFromThreadId("telegram:biz:conn-abc:456")).toBe( + "telegram:biz:conn-abc:456" + ); + expect(adapter.channelIdFromThreadId("telegram:456")).toBe("telegram:456"); + expect(adapter.channelIdFromThreadId("telegram:-100123:42")).toBe( + "telegram:-100123" + ); + }); + + it("routes callback queries from business chats to the business thread", async () => { + mockFetch + .mockResolvedValueOnce( + telegramOk({ + id: 999, + is_bot: true, + first_name: "Bot", + username: "mybot", + }) + ) + .mockResolvedValue(telegramOk(true)); + + const adapter = createTelegramAdapter({ + allowUnverifiedWebhooks: true, + botToken: "token", + businessMode: true, + mode: "webhook", + logger: mockLogger, + userName: "mybot", + }); + const chat = createMockChat(); + await adapter.initialize(chat); + + await adapter.handleWebhook( + new Request("https://example.com/webhooks/telegram", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + update_id: 1, + callback_query: { + id: "callback-1", + from: { + id: 456, + is_bot: false, + first_name: "Customer", + username: "customer", + }, + message: sampleBusinessMessage({ message_id: 77 }), + chat_instance: "ci_1", + data: encodeTelegramCallbackData("approve", "order-1"), + }, + }), + }) + ); + + const processAction = chat.processAction as ReturnType; + expect(processAction).toHaveBeenCalledTimes(1); + const [event] = processAction.mock.calls[0] as [ + { messageId: string; threadId: string }, + ]; + expect(event.threadId).toBe("telegram:biz:conn-abc:456"); + expect(event.messageId).toBe("456:77"); + }); + + it("caches business_connection updates even when the owner is not in allowedUserIds", async () => { + mockFetch + .mockResolvedValueOnce( + telegramOk({ + id: 999, + is_bot: true, + first_name: "Bot", + username: "mybot", + }) + ) + .mockResolvedValue(telegramOk(true)); + + const state = createMockState(); + const adapter = createTelegramAdapter({ + allowUnverifiedWebhooks: true, + allowedUserIds: [456], + botToken: "token", + businessMode: true, + mode: "webhook", + logger: mockLogger, + userName: "mybot", + }); + const chat = createMockChatInstance({ + logger: mockLogger, + state, + userName: "mybot", + }); + await adapter.initialize(chat); + + const pendingTasks: Promise[] = []; + const waitUntil = (task: Promise) => { + pendingTasks.push(task); + }; + + // The owner (111) revokes can_reply. The update has no customer user id. + await adapter.handleWebhook( + new Request("https://example.com/webhooks/telegram", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + update_id: 1, + business_connection: { + ...sampleBusinessConnection, + rights: { can_reply: false }, + }, + }), + }), + { waitUntil } + ); + await Promise.all(pendingTasks); + + expect( + await state.get("telegram:business-connection:conn-abc") + ).toMatchObject({ id: "conn-abc", rights: { can_reply: false } }); + + // The next customer message honours the revocation without refetching. + await adapter.handleWebhook( + new Request("https://example.com/webhooks/telegram", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + update_id: 2, + business_message: sampleBusinessMessage(), + }), + }), + { waitUntil } + ); + await Promise.all(pendingTasks); + + expect(chat.processMessage).not.toHaveBeenCalled(); + expect(telegramMethods()).toEqual(["getMe"]); + expect(mockLogger.debug).toHaveBeenCalledWith( + "Ignoring Telegram business message: connection cannot reply", + expect.objectContaining({ connectionId: "conn-abc", canReply: false }) + ); + }); + + it("shares the business connection cache across adapter instances via state", async () => { + mockFetch + .mockResolvedValueOnce( + telegramOk({ + id: 999, + is_bot: true, + first_name: "Bot", + username: "mybot", + }) + ) + .mockResolvedValueOnce( + telegramOk({ + id: 999, + is_bot: true, + first_name: "Bot", + username: "mybot", + }) + ) + .mockResolvedValue(telegramOk(true)); + + const state = createMockState(); + const config = { + allowUnverifiedWebhooks: true, + botToken: "token", + businessMode: true, + mode: "webhook" as const, + logger: mockLogger, + userName: "mybot", + }; + const instanceA = createTelegramAdapter(config); + const instanceB = createTelegramAdapter(config); + const chatA = createMockChatInstance({ logger: mockLogger, state }); + const chatB = createMockChatInstance({ logger: mockLogger, state }); + await instanceA.initialize(chatA); + await instanceB.initialize(chatB); + + const pendingTasks: Promise[] = []; + const waitUntil = (task: Promise) => { + pendingTasks.push(task); + }; + + await instanceA.handleWebhook( + new Request("https://example.com/webhooks/telegram", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + update_id: 1, + business_connection: sampleBusinessConnection, + }), + }), + { waitUntil } + ); + await instanceB.handleWebhook( + new Request("https://example.com/webhooks/telegram", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + update_id: 2, + business_message: sampleBusinessMessage(), + }), + }), + { waitUntil } + ); + await Promise.all(pendingTasks); + + expect(chatB.processMessage).toHaveBeenCalledTimes(1); + expect(telegramMethods()).not.toContain("getBusinessConnection"); + }); + + it("fetches the connection once and caches it when no update was seen", async () => { + mockFetch + .mockResolvedValueOnce( + telegramOk({ + id: 999, + is_bot: true, + first_name: "Bot", + username: "mybot", + }) + ) + .mockResolvedValueOnce(telegramOk(sampleBusinessConnection)) + .mockResolvedValue(telegramOk(true)); + + const adapter = createTelegramAdapter({ + allowUnverifiedWebhooks: true, + botToken: "token", + businessMode: true, + mode: "webhook", + logger: mockLogger, + userName: "mybot", + }); + const chat = createMockChat(); + await adapter.initialize(chat); + + const pendingTasks: Promise[] = []; + const waitUntil = (task: Promise) => { + pendingTasks.push(task); }; - expect(deleteMessageBody.chat_id).toBe("456"); - expect(deleteMessageBody.message_id).toBe(77); - expect(deleteMessageBody.business_connection_id).toBeUndefined(); + for (const updateId of [1, 2]) { + await adapter.handleWebhook( + new Request("https://example.com/webhooks/telegram", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + update_id: updateId, + business_message: sampleBusinessMessage({ message_id: updateId }), + }), + }), + { waitUntil } + ); + // Let the first message finish populating the cache before the next. + await Promise.all(pendingTasks); + } + + expect(chat.processMessage).toHaveBeenCalledTimes(2); + expect( + telegramMethods().filter((method) => method === "getBusinessConnection") + ).toHaveLength(1); + }); + + it("routes slash commands from business chats to processSlashCommand", async () => { + mockFetch + .mockResolvedValueOnce( + telegramOk({ + id: 999, + is_bot: true, + first_name: "Bot", + username: "mybot", + }) + ) + .mockResolvedValue(telegramOk(true)); + + const adapter = createTelegramAdapter({ + allowUnverifiedWebhooks: true, + botToken: "token", + businessMode: true, + mode: "webhook", + logger: mockLogger, + userName: "mybot", + }); + const chat = createMockChat(); + await adapter.initialize(chat); + + const pendingTasks: Promise[] = []; + await adapter.handleWebhook( + new Request("https://example.com/webhooks/telegram", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + update_id: 1, + business_connection: sampleBusinessConnection, + business_message: sampleBusinessMessage({ + text: "/help orders", + entities: [{ type: "bot_command", offset: 0, length: 5 }], + }), + }), + }), + { + waitUntil: (task: Promise) => { + pendingTasks.push(task); + }, + } + ); + await Promise.all(pendingTasks); + + expect(chat.processMessage).not.toHaveBeenCalled(); + const processSlashCommand = chat.processSlashCommand as ReturnType< + typeof vi.fn + >; + expect(processSlashCommand).toHaveBeenCalledTimes(1); + const [event] = processSlashCommand.mock.calls[0] as [ + { channelId: string; command: string; text: string }, + ]; + expect(event.channelId).toBe("telegram:biz:conn-abc:456"); + expect(event.command).toBe("/help"); + expect(event.text).toBe("orders"); + }); + + it("skips outgoing business messages the bot sent on behalf of the account", async () => { + mockFetch + .mockResolvedValueOnce( + telegramOk({ + id: 999, + is_bot: true, + first_name: "Bot", + username: "mybot", + }) + ) + .mockResolvedValue(telegramOk(true)); + + const adapter = createTelegramAdapter({ + allowUnverifiedWebhooks: true, + botToken: "token", + businessMode: true, + mode: "webhook", + logger: mockLogger, + userName: "mybot", + }); + const chat = createMockChat(); + await adapter.initialize(chat); + + const pendingTasks: Promise[] = []; + await adapter.handleWebhook( + new Request("https://example.com/webhooks/telegram", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + update_id: 1, + business_connection: sampleBusinessConnection, + business_message: sampleBusinessMessage({ + from: sampleBusinessConnection.user, + sender_business_bot: { + id: 999, + is_bot: true, + first_name: "Bot", + username: "mybot", + }, + text: "Thanks, looking into it", + }), + }), + }), + { + waitUntil: (task: Promise) => { + pendingTasks.push(task); + }, + } + ); + await Promise.all(pendingTasks); + + expect(chat.processMessage).not.toHaveBeenCalled(); + }); + + it("attributes sender_business_bot messages to the bot", async () => { + class ParsingAdapter extends TelegramAdapter { + parse(message: TelegramMessage, threadId: string) { + return this.parseTelegramMessage(message, threadId); + } + } + + mockFetch.mockResolvedValueOnce( + telegramOk({ + id: 999, + is_bot: true, + first_name: "Bot", + username: "mybot", + }) + ); + const adapter = new ParsingAdapter({ + botToken: "token", + mode: "webhook", + logger: mockLogger, + userName: "mybot", + }); + await adapter.initialize(createMockChat()); + + const parsed = adapter.parse( + sampleBusinessMessage({ + from: sampleBusinessConnection.user, + sender_business_bot: { + id: 999, + is_bot: true, + first_name: "Bot", + username: "mybot", + }, + }), + "telegram:biz:conn-abc:456" + ); + + expect(parsed.author.isMe).toBe(true); + expect(parsed.author.isBot).toBe(true); + expect(parsed.author.userId).toBe("999"); + }); + + it("returns business thread info with the connection id and a cached chat fallback", async () => { + mockFetch + .mockResolvedValueOnce( + telegramOk({ + id: 999, + is_bot: true, + first_name: "Bot", + username: "mybot", + }) + ) + .mockResolvedValueOnce(telegramOk(sampleBusinessConnection)) + .mockResolvedValueOnce( + telegramError(400, 400, "Bad Request: chat not found") + ); + + const adapter = createTelegramAdapter({ + allowUnverifiedWebhooks: true, + botToken: "token", + businessMode: true, + mode: "webhook", + logger: mockLogger, + userName: "mybot", + }); + const chat = createMockChat(); + await adapter.initialize(chat); + + const pendingTasks: Promise[] = []; + await adapter.handleWebhook( + new Request("https://example.com/webhooks/telegram", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + update_id: 1, + business_message: sampleBusinessMessage(), + }), + }), + { + waitUntil: (task: Promise) => { + pendingTasks.push(task); + }, + } + ); + await Promise.all(pendingTasks); + + const thread = await adapter.fetchThread("telegram:biz:conn-abc:456"); + + expect(thread.id).toBe("telegram:biz:conn-abc:456"); + expect(thread.channelId).toBe("456"); + expect(thread.channelName).toBe("Customer"); + expect(thread.isDM).toBe(true); + expect(thread.metadata.businessConnectionId).toBe("conn-abc"); + }); + + it("sends an explicit allowed_updates list for polling in business mode", () => { + class PollingTestAdapter extends TelegramAdapter { + resolvePolling( + override?: Parameters[0] + ) { + return this.resolvePollingConfig(override); + } + } + + const pollingAdapter = new PollingTestAdapter({ + botToken: "token", + businessMode: true, + mode: "polling", + logger: mockLogger, + }); + + expect(pollingAdapter.resolvePolling()?.allowedUpdates).toEqual([ + "message", + "edited_message", + "channel_post", + "edited_channel_post", + "callback_query", + "business_connection", + "business_message", + "edited_business_message", + ]); }); it("merges business update types into custom polling allowedUpdates", () => { diff --git a/packages/adapter-telegram/src/index.ts b/packages/adapter-telegram/src/index.ts index 761e4ecf..7ff9b604 100644 --- a/packages/adapter-telegram/src/index.ts +++ b/packages/adapter-telegram/src/index.ts @@ -194,6 +194,24 @@ const TELEGRAM_BUSINESS_ALLOWED_UPDATES = [ "business_message", "edited_business_message", ] as const; +/** + * Update types the adapter routes that Telegram also delivers by default. + * `message_reaction` is deliberately absent: Telegram excludes it unless a + * caller opts in, and Business mode should not change that. + */ +const TELEGRAM_DEFAULT_ALLOWED_UPDATES = [ + "message", + "edited_message", + "channel_post", + "edited_channel_post", + "callback_query", +] as const; +/** + * How long a `BusinessConnection` stays cached in shared state. Telegram + * pushes a `business_connection` update whenever rights change, which + * overwrites the entry; the TTL only bounds staleness if that update is lost. + */ +const TELEGRAM_BUSINESS_CONNECTION_TTL_MS = 60 * 60 * 1000; function normalizeBotTokenProvider( botToken: string | (() => string | Promise) @@ -457,10 +475,6 @@ export class TelegramAdapter protected readonly streamingEditIntervalMs?: number; protected readonly longPolling?: TelegramLongPollingConfig; protected readonly businessMode: boolean; - private readonly businessConnectionCache = new Map< - string, - TelegramBusinessConnection - >(); private _runtimeMode: TelegramRuntimeMode = "webhook"; private pollingAbortController: AbortController | null = null; private pollingTask: Promise | null = null; @@ -895,6 +909,20 @@ export class TelegramAdapter update: TelegramUpdate, options?: WebhookOptions ): void { + // Connection state is recorded before the allowlist gate: a + // `business_connection` update carries no customer user id, and dropping + // it would leave a revoked connection looking usable. + if (this.businessMode && update.business_connection) { + const connection = update.business_connection; + const task = this.cacheBusinessConnection(connection).catch((error) => { + this.logger.warn("Failed to cache Telegram business connection", { + error: String(error), + connectionId: connection.id, + }); + }); + options?.waitUntil?.(task); + } + const messageUpdate = update.message ?? update.edited_message ?? @@ -918,7 +946,14 @@ export class TelegramAdapter const handledSlashCommand = update.message !== undefined && !update.message.media_group_id && - this.handleSlashCommandUpdate(update.message, options); + this.handleSlashCommandUpdate( + update.message, + this.encodeThreadId({ + chatId: String(update.message.chat.id), + messageThreadId: update.message.message_thread_id, + }), + options + ); if (messageUpdate && !handledSlashCommand) { this.handleIncomingMessageUpdate(messageUpdate, options); @@ -932,39 +967,51 @@ export class TelegramAdapter this.handleMessageReactionUpdate(update.message_reaction, options); } - if (this.businessMode) { - if (update.business_connection) { - this.cacheBusinessConnection(update.business_connection); - } - - const businessMessage = - update.business_message ?? update.edited_business_message; - if (businessMessage) { - const task = this.handleBusinessMessageUpdate( - businessMessage, - businessMessage.business_connection_id, - options - ).catch((error) => { - this.logger.warn("Failed to process Telegram business message", { - error: String(error), - connectionId: businessMessage.business_connection_id, - }); + if (this.businessMode && businessMessageUpdate) { + const task = this.handleBusinessMessageUpdate( + businessMessageUpdate, + { isEdit: update.business_message === undefined }, + options + ).catch((error) => { + this.logger.warn("Failed to process Telegram business message", { + error: String(error), + connectionId: businessMessageUpdate.business_connection_id, }); - options?.waitUntil?.(task); - } + }); + options?.waitUntil?.(task); } } - protected cacheBusinessConnection( + protected businessConnectionCacheKey(connectionId: string): string { + return `${this.name}:business-connection:${connectionId}`; + } + + /** + * Business connections live in shared state rather than a per-instance map + * so a rights change that reaches one serverless instance is visible to all + * of them on the next message. + */ + protected async cacheBusinessConnection( connection: TelegramBusinessConnection - ): void { - this.businessConnectionCache.set(connection.id, connection); + ): Promise { + if (!this.chat) { + return; + } + await this.chat + .getState() + .set( + this.businessConnectionCacheKey(connection.id), + connection, + TELEGRAM_BUSINESS_CONNECTION_TTL_MS + ); } protected async resolveBusinessConnection( connectionId: string ): Promise { - const cached = this.businessConnectionCache.get(connectionId); + const state = this.chat?.getState(); + const key = this.businessConnectionCacheKey(connectionId); + const cached = await state?.get(key); if (cached) { return cached; } @@ -975,53 +1022,84 @@ export class TelegramAdapter business_connection_id: connectionId, } ); - this.businessConnectionCache.set(connectionId, connection); + await state?.set(key, connection, TELEGRAM_BUSINESS_CONNECTION_TTL_MS); return connection; } protected isBusinessConnectionUsable( connection: TelegramBusinessConnection ): boolean { - return ( - connection.is_enabled && (connection.rights?.can_reply ?? false) - ); + return connection.is_enabled && (connection.rights?.can_reply ?? false); } protected async handleBusinessMessageUpdate( telegramMessage: TelegramMessage, - connectionId: string | undefined, + routing: { isEdit: boolean }, options?: WebhookOptions ): Promise { if (!this.chat) { return; } - const resolvedConnectionId = - connectionId ?? telegramMessage.business_connection_id; - if (!resolvedConnectionId) { + const connectionId = telegramMessage.business_connection_id; + if (!connectionId) { this.logger.warn("Telegram business message missing connection id", { messageId: telegramMessage.message_id, }); return; } - const connection = - await this.resolveBusinessConnection(resolvedConnectionId); + const connection = await this.resolveBusinessConnection(connectionId); if (!this.isBusinessConnectionUsable(connection)) { + this.logger.debug( + "Ignoring Telegram business message: connection cannot reply", + { + connectionId, + isEnabled: connection.is_enabled, + canReply: connection.rights?.can_reply ?? false, + } + ); return; } + // Telegram echoes outgoing business messages back as `business_message`: + // both the ones this bot sent on the account's behalf (`sender_business_bot`) + // and the ones the owner typed themselves. Neither is customer input. + if (telegramMessage.sender_business_bot) { + return; + } if (telegramMessage.from?.id === connection.user.id) { + this.logger.debug( + "Ignoring Telegram business message typed by the account owner", + { connectionId, messageId: telegramMessage.message_id } + ); return; } const threadId = this.encodeThreadId({ chatId: String(telegramMessage.chat.id), - businessConnectionId: resolvedConnectionId, + businessConnectionId: connectionId, }); if (telegramMessage.media_group_id) { - await this.processIncomingMediaGroup(telegramMessage, threadId); + const task = this.processIncomingMediaGroup( + telegramMessage, + threadId + ).catch((error) => { + this.logger.warn("Failed to process incoming Telegram media group", { + error: String(error), + mediaGroupId: telegramMessage.media_group_id, + threadId, + }); + }); + options?.waitUntil?.(task); + return; + } + + if ( + !routing.isEdit && + this.handleSlashCommandUpdate(telegramMessage, threadId, options) + ) { return; } @@ -1030,12 +1108,8 @@ export class TelegramAdapter const parsedMessage = this.parseTelegramMessage(telegramMessage, threadId); this.cacheMessage(parsedMessage); - await this.chat.processMessage( - this, - threadId, - parsedMessage, - options - ); + // Not awaited: processMessage logs and waitUntil-tracks its own failures. + this.chat.processMessage(this, threadId, parsedMessage, options); } protected handleIncomingMessageUpdate( @@ -1173,6 +1247,7 @@ export class TelegramAdapter protected handleSlashCommandUpdate( telegramMessage: TelegramMessage, + threadId: string, options?: WebhookOptions ): boolean { if (!this.chat) { @@ -1184,11 +1259,6 @@ export class TelegramAdapter return false; } - const threadId = this.encodeThreadId({ - chatId: String(telegramMessage.chat.id), - messageThreadId: telegramMessage.message_thread_id, - }); - this.startTypingForPrivateMessage(telegramMessage, threadId, options); const parsedMessage = this.parseTelegramMessage(telegramMessage, threadId); @@ -1291,6 +1361,7 @@ export class TelegramAdapter const threadId = this.encodeThreadId({ chatId: String(callbackQuery.message.chat.id), messageThreadId: callbackQuery.message.message_thread_id, + businessConnectionId: callbackQuery.message.business_connection_id, }); const messageId = this.encodeMessageId( @@ -1613,11 +1684,8 @@ export class TelegramAdapter message: AdapterPostableMessage ): Promise> { const parsedThread = this.resolveThreadId(threadId); - const { - chatId, - messageId: telegramMessageId, - compositeId, - } = this.decodeCompositeMessageId(messageId, parsedThread.chatId); + const { messageId: telegramMessageId, compositeId } = + this.decodeCompositeMessageId(messageId, parsedThread.chatId); const card = extractCard(message); const replyMarkup = card ? cardToTelegramInlineKeyboard(card) : undefined; @@ -1744,18 +1812,23 @@ export class TelegramAdapter async deleteMessage(threadId: string, messageId: string): Promise { const parsedThread = this.resolveThreadId(threadId); - const { - chatId, - messageId: telegramMessageId, - compositeId, - } = this.decodeCompositeMessageId(messageId, parsedThread.chatId); - - await this.telegramFetch("deleteMessage", { - ...this.buildChatTargetParams(parsedThread, { - includeBusinessConnection: false, - }), - message_id: telegramMessageId, - }); + const { messageId: telegramMessageId, compositeId } = + this.decodeCompositeMessageId(messageId, parsedThread.chatId); + + if (parsedThread.businessConnectionId) { + // The bot is not a participant in the owner <-> customer chat, so the + // bot-scoped `deleteMessage` cannot address it. Requires the + // `can_delete_sent_messages` (or `can_delete_all_messages`) right. + await this.telegramFetch("deleteBusinessMessages", { + business_connection_id: parsedThread.businessConnectionId, + message_ids: [telegramMessageId], + }); + } else { + await this.telegramFetch("deleteMessage", { + chat_id: parsedThread.chatId, + message_id: telegramMessageId, + }); + } this.deleteCachedMessage(compositeId); } @@ -1766,13 +1839,14 @@ export class TelegramAdapter emoji: EmojiValue | string ): Promise { const parsedThread = this.resolveThreadId(threadId); - const { chatId, messageId: telegramMessageId } = - this.decodeCompositeMessageId(messageId, parsedThread.chatId); + this.assertReactionsSupported(parsedThread, "addReaction"); + const { messageId: telegramMessageId } = this.decodeCompositeMessageId( + messageId, + parsedThread.chatId + ); await this.telegramFetch("setMessageReaction", { - ...this.buildChatTargetParams(parsedThread, { - includeBusinessConnection: false, - }), + chat_id: parsedThread.chatId, message_id: telegramMessageId, reaction: [this.toTelegramReaction(emoji)], }); @@ -1784,18 +1858,37 @@ export class TelegramAdapter _emoji: EmojiValue | string ): Promise { const parsedThread = this.resolveThreadId(threadId); - const { chatId, messageId: telegramMessageId } = - this.decodeCompositeMessageId(messageId, parsedThread.chatId); + this.assertReactionsSupported(parsedThread, "removeReaction"); + const { messageId: telegramMessageId } = this.decodeCompositeMessageId( + messageId, + parsedThread.chatId + ); await this.telegramFetch("setMessageReaction", { - ...this.buildChatTargetParams(parsedThread, { - includeBusinessConnection: false, - }), + chat_id: parsedThread.chatId, message_id: telegramMessageId, reaction: [], }); } + /** + * `setMessageReaction` has no `business_connection_id` parameter, and the + * bot cannot address a business chat by `chat_id` alone, so reactions on + * business threads fail with an opaque Telegram error. Surface that up + * front instead. + */ + protected assertReactionsSupported( + thread: TelegramThreadId, + method: string + ): void { + if (thread.businessConnectionId) { + throw new NotImplementedError( + "Telegram Bot API does not support reactions on behalf of a business account", + method + ); + } + } + async startTyping(threadId: string): Promise { const parsedThread = this.resolveThreadId(threadId); await this.telegramFetch("sendChatAction", { @@ -2286,12 +2379,31 @@ export class TelegramAdapter async fetchThread(threadId: string): Promise { const parsedThread = this.resolveThreadId(threadId); + const encodedThreadId = this.encodeThreadId(parsedThread); + + if (parsedThread.businessConnectionId) { + // `getChat` has no business context and only resolves private chats + // the bot has met directly, so a business customer may be unknown to + // it. Fall back to the chat object from messages already seen. + const chat = await this.fetchBusinessChat(parsedThread, encodedThreadId); + return { + id: encodedThreadId, + channelId: parsedThread.chatId, + channelName: chat ? this.chatDisplayName(chat) : undefined, + isDM: true, + metadata: { + chat, + businessConnectionId: parsedThread.businessConnectionId, + }, + }; + } + const chat = await this.telegramFetch("getChat", { chat_id: parsedThread.chatId, }); return { - id: this.encodeThreadId(parsedThread), + id: encodedThreadId, channelId: String(chat.id), channelName: this.chatDisplayName(chat), isDM: chat.type === "private", @@ -2302,6 +2414,27 @@ export class TelegramAdapter }; } + protected async fetchBusinessChat( + thread: TelegramThreadId, + threadId: string + ): Promise { + const cachedChat = this.messageCache.get(threadId)?.at(-1)?.raw.chat; + try { + return await this.telegramFetch("getChat", { + chat_id: thread.chatId, + }); + } catch (error) { + if (!cachedChat) { + throw error; + } + this.logger.debug( + "Telegram getChat failed for business thread, using cached chat", + { error: String(error), threadId } + ); + return cachedChat; + } + } + async fetchChannelInfo(channelId: string): Promise { const chat = await this.telegramFetch("getChat", { chat_id: channelId, @@ -2329,7 +2462,14 @@ export class TelegramAdapter } channelIdFromThreadId(threadId: string): string { - const { chatId } = this.resolveThreadId(threadId); + const { chatId, businessConnectionId } = this.resolveThreadId(threadId); + // A business conversation is its own channel. Collapsing it to + // `telegram:{chatId}` would post channel-level messages without + // `business_connection_id` and share lock/history keys with the + // customer's direct chat with the bot, which Telegram keeps separate. + if (businessConnectionId) { + return this.encodeThreadId({ chatId, businessConnectionId }); + } return `telegram:${chatId}`; } @@ -2465,7 +2605,11 @@ export class TelegramAdapter : applyTelegramEntities(plainText, entities); let author: TelegramMessageAuthor; - if (raw.from) { + if (raw.sender_business_bot) { + // Outgoing business messages carry the owner in `from`; the bot that + // actually sent them is `sender_business_bot`, so `isMe` follows it. + author = this.toAuthor(raw.sender_business_bot); + } else if (raw.from) { author = this.toAuthor(raw.from); } else if (raw.sender_chat) { author = this.toReactionActorAuthor(raw.sender_chat); @@ -3249,15 +3393,11 @@ export class TelegramAdapter * Bot API chat-target fields (`chat_id`, `message_thread_id`, and optionally * `business_connection_id`) derived from a decoded thread. */ - protected buildChatTargetParams( - thread: TelegramThreadId, - options: { includeBusinessConnection?: boolean } = {} - ): { + protected buildChatTargetParams(thread: TelegramThreadId): { business_connection_id?: string; chat_id: string; message_thread_id?: number; } { - const includeBusinessConnection = options.includeBusinessConnection ?? true; const params: { business_connection_id?: string; chat_id: string; @@ -3270,7 +3410,7 @@ export class TelegramAdapter params.message_thread_id = thread.messageThreadId; } - if (includeBusinessConnection && thread.businessConnectionId) { + if (thread.businessConnectionId) { params.business_connection_id = thread.businessConnectionId; } @@ -3286,10 +3426,7 @@ export class TelegramAdapter formData.append("message_thread_id", String(thread.messageThreadId)); } if (thread.businessConnectionId) { - formData.append( - "business_connection_id", - thread.businessConnectionId - ); + formData.append("business_connection_id", thread.businessConnectionId); } } @@ -3852,13 +3989,15 @@ export class TelegramAdapter : undefined; } - if (!allowedUpdates || allowedUpdates.length === 0) { - return undefined; - } + // Omitting `allowed_updates` makes Telegram reuse whatever list a previous + // setWebhook/getUpdates call set, which may exclude the business types. + // Business mode always sends an explicit list. + const base = + allowedUpdates && allowedUpdates.length > 0 + ? allowedUpdates + : TELEGRAM_DEFAULT_ALLOWED_UPDATES; - return [ - ...new Set([...allowedUpdates, ...TELEGRAM_BUSINESS_ALLOWED_UPDATES]), - ]; + return [...new Set([...base, ...TELEGRAM_BUSINESS_ALLOWED_UPDATES])]; } protected clampInteger( @@ -4082,8 +4221,8 @@ export type { TelegramAdapterMode, TelegramAnimation, TelegramAudio, - TelegramBusinessConnection, TelegramBusinessBotRights, + TelegramBusinessConnection, TelegramCallbackQuery, TelegramChat, TelegramLocation, diff --git a/packages/adapter-telegram/src/types.ts b/packages/adapter-telegram/src/types.ts index d78cff98..cfb70b53 100644 --- a/packages/adapter-telegram/src/types.ts +++ b/packages/adapter-telegram/src/types.ts @@ -17,6 +17,12 @@ export interface TelegramAdapterConfig { apiUrl?: string; /** Telegram bot token from BotFather, or a resolver invoked for each Bot API request. Defaults to TELEGRAM_BOT_TOKEN env var. */ botToken?: string | (() => string | Promise); + /** + * Enable Telegram Business mode (Connected Business Bots). When `true`, the + * adapter handles `business_connection` and `business_message` updates and + * passes `business_connection_id` on outbound API calls. Defaults to `false`. + */ + businessMode?: boolean; /** Logger instance for error reporting. Defaults to ConsoleLogger. */ logger?: Logger; /** Optional long-polling configuration for getUpdates flow. */ @@ -58,12 +64,6 @@ export interface TelegramAdapterConfig { streamingEditIntervalMs?: number; /** Override bot username (optional). Defaults to TELEGRAM_BOT_USERNAME env var. */ userName?: string; - /** - * Enable Telegram Business mode (Connected Business Bots). When `true`, the - * adapter handles `business_connection` and `business_message` updates and - * passes `business_connection_id` on outbound API calls. Defaults to `false`. - */ - businessMode?: boolean; } export type TelegramAdapterMode = "auto" | "webhook" | "polling"; @@ -99,12 +99,12 @@ export interface TelegramLongPollingConfig { * Telegram thread ID components. */ export interface TelegramThreadId { + /** Business connection ID for Connected Business Bot threads. */ + businessConnectionId?: string; /** Telegram chat ID. */ chatId: string; /** Optional forum topic ID for supergroup topics. */ messageThreadId?: number; - /** Business connection ID for Connected Business Bot threads. */ - businessConnectionId?: string; } /** @@ -476,6 +476,12 @@ export interface TelegramMessage { mime_type?: string; file_name?: string; }; + /** + * Business connection the message was received through. When set, the + * chat belongs to the connected business account and is independent from + * any direct bot chat that shares the same `chat.id`. + */ + business_connection_id?: string; caption?: string; caption_entities?: TelegramMessageEntity[]; chat: TelegramChat; @@ -503,7 +509,6 @@ export interface TelegramMessage { media_group_id?: string; message_id: number; message_thread_id?: number; - business_connection_id?: string; photo?: TelegramPhotoSize[]; poll?: { id: string; @@ -513,6 +518,11 @@ export interface TelegramMessage { }; reply_to_message?: TelegramMessage; rich_message?: TelegramRichMessage; + /** + * The bot that sent this message on behalf of the business account. Only + * present on outgoing business messages. + */ + sender_business_bot?: TelegramUser; sender_chat?: TelegramChat; sticker?: TelegramFile & { emoji?: string; @@ -583,6 +593,9 @@ export type TelegramReactionType = * @see https://core.telegram.org/bots/api#businessbotrights */ export interface TelegramBusinessBotRights { + can_delete_all_messages?: boolean; + can_delete_sent_messages?: boolean; + can_read_messages?: boolean; can_reply?: boolean; } From c5f435773d694d1700da87e289331455b532f6e8 Mon Sep 17 00:00:00 2001 From: Ben Sabic Date: Fri, 4 Sep 2026 13:36:29 +1000 Subject: [PATCH 4/5] docs(telegram): describe Business mode behavior and limits Cover channel separation, slash commands, deletes via deleteBusinessMessages, the reaction limitation, polling allowed_updates, and the telegram:biz sharding caveat. --- .../docs/content/adapters/official/telegram.mdx | 16 +++++++++++++--- .../vendor-official/cloudflare-agents.mdx | 2 ++ packages/adapter-telegram/README.md | 17 ++++++++++++----- 3 files changed, 27 insertions(+), 8 deletions(-) diff --git a/apps/docs/content/adapters/official/telegram.mdx b/apps/docs/content/adapters/official/telegram.mdx index b37b4a08..68c6c52a 100644 --- a/apps/docs/content/adapters/official/telegram.mdx +++ b/apps/docs/content/adapters/official/telegram.mdx @@ -167,7 +167,7 @@ const telegram = createTelegramAdapter({ type: "boolean", default: "false", description: - "Enable Telegram Business mode for Connected Business Bots. Handles `business_connection` and `business_message` updates and passes `business_connection_id` on outbound API calls.", + "Enable Telegram Business mode for Connected Business Bots. Handles `business_connection`, `business_message`, and `edited_business_message` updates and passes `business_connection_id` on outbound API calls.", }, mentionOnReply: { type: "boolean", @@ -212,9 +212,19 @@ const telegram = createTelegramAdapter({ }); ``` -Business threads use the ID format `telegram:biz:{connectionId}:{chatId}`. Outbound sends automatically include `business_connection_id`. +Business threads use the ID format `telegram:biz:{connectionId}:{chatId}`. Each business conversation is its own channel, separate from any direct chat the same customer has with the bot. Outbound sends, edits, typing, file uploads, and threads created from inline-keyboard callbacks include `business_connection_id`. Slash commands from business chats reach `onSlashCommand` like any other chat. -When configuring webhooks manually, add business update types to `allowed_updates` (or rely on the adapter's polling merge when `longPolling.allowedUpdates` is set). +The adapter ignores messages typed by the business owner and messages the bot sent on the account's behalf, and it stops replying when the connection is disabled or loses `can_reply`. Connection state lives in your state adapter, so a change reaches every instance. + +A few Bot API limits apply to business threads: + +- `delete()` uses `deleteBusinessMessages`, which needs the `can_delete_sent_messages` right. +- Reactions are not supported. `addReaction` and `removeReaction` throw a `NotImplementedError`. +- `fetchThread()` falls back to the chat details from messages already seen when `getChat` cannot resolve the customer. + +When polling, Business mode always sends an explicit `allowed_updates` list (the default update types plus the business ones, or your `longPolling.allowedUpdates` merged with them). Telegram otherwise reuses the list from an earlier call, which can silently exclude business updates. When registering a webhook yourself, add `business_connection`, `business_message`, and `edited_business_message` to `allowed_updates`. + +Business thread IDs start with `telegram:biz`, so state adapters that shard by the first two ID segments (such as [Cloudflare Agents](/adapters/vendor-official/cloudflare-agents#state-sharding)) place every business conversation in one shard. Override the sharder with a key that includes the connection ID if that matters for your deployment. ## Authentication diff --git a/apps/docs/content/adapters/vendor-official/cloudflare-agents.mdx b/apps/docs/content/adapters/vendor-official/cloudflare-agents.mdx index fcd2b911..41b1b701 100644 --- a/apps/docs/content/adapters/vendor-official/cloudflare-agents.mdx +++ b/apps/docs/content/adapters/vendor-official/cloudflare-agents.mdx @@ -110,6 +110,8 @@ The default key sharder recognizes these Chat SDK key prefixes: Unknown keys use the adapter's default shard name, `default`. +Telegram Business threads (`telegram:biz:{connectionId}:{chatId}`) all share the shard `telegram:biz` under this rule. Use a custom `shardKey` that keeps the connection ID if you expect many business conversations. + ### Custom sharding Use `shardKey` to control how thread IDs map to state sub-agent names, and `keyShard` for non-thread-shaped keys that should still route to a provider-specific shard: diff --git a/packages/adapter-telegram/README.md b/packages/adapter-telegram/README.md index 7161b7f9..03e61384 100644 --- a/packages/adapter-telegram/README.md +++ b/packages/adapter-telegram/README.md @@ -104,11 +104,16 @@ const telegram = createTelegramAdapter({ With `businessMode: true`, the adapter: - Handles `business_connection`, `business_message`, and `edited_business_message` updates -- Skips messages sent by the business owner and connections without `can_reply` -- Encodes business threads as `telegram:biz:{connectionId}:{chatId}` -- Passes `business_connection_id` on outbound `sendMessage`, edits, typing, and file uploads +- Skips messages typed by the business owner, messages the bot itself sent on the account's behalf (`sender_business_bot`), and connections without `can_reply` +- Encodes business threads as `telegram:biz:{connectionId}:{chatId}`. Each business conversation is its own channel, kept apart from any direct chat the same customer has with the bot +- Passes `business_connection_id` on outbound `sendMessage`, edits, typing, file uploads, and on threads created from inline-keyboard callbacks +- Routes `/commands` from business chats through `onSlashCommand`, the same as regular chats +- Deletes messages with `deleteBusinessMessages`, which needs the `can_delete_sent_messages` right on the connection +- Stores connection state in the Chat SDK state adapter, so a revoked or disabled connection is honoured by every running instance -When registering a webhook or starting polling with a custom `allowedUpdates` list, include the business update types (the adapter merges them automatically for polling when `businessMode` is `true`): +Reactions are not available on business threads. The Bot API has no business variant of `setMessageReaction`, so `addReaction` and `removeReaction` throw a `NotImplementedError` there. + +When polling, the adapter always sends an explicit `allowed_updates` list in Business mode: the default update types plus the business ones, or your `longPolling.allowedUpdates` merged with the business ones. Telegram otherwise reuses whatever list an earlier call set, which can silently exclude business updates. When registering a webhook yourself, include the business update types: ```json [ @@ -119,6 +124,8 @@ When registering a webhook or starting polling with a custom `allowedUpdates` li ] ``` +Business thread IDs start with `telegram:biz`, so state adapters that shard by the first two ID segments (such as Cloudflare Agents) place every business conversation in one shard. Override the sharder with a key that includes the connection ID if that matters for your deployment. + Defaults to `false` for backward compatibility. ## Polling (local development) @@ -191,7 +198,7 @@ Most options are auto-detected from environment variables when not provided. `na | `secretToken` | Webhook* | Webhook secret token. Auto-detected from `TELEGRAM_WEBHOOK_SECRET_TOKEN` | | `mode` | No | Adapter mode: `auto` (default), `webhook`, or `polling` | | `longPolling` | No | Optional long polling config for `getUpdates` (`timeout`, `limit`, `allowedUpdates`, `deleteWebhook`, `dropPendingUpdates`, `retryDelayMs`) | -| `businessMode` | No | Enable Telegram Business mode (`business_connection`, `business_message`). Defaults to `false` | +| `businessMode` | No | Enable Telegram Business mode (`business_connection`, `business_message`, `edited_business_message`). Defaults to `false` | | `userName` | No | Bot username used for mention detection. Auto-detected from `TELEGRAM_BOT_USERNAME` or `getMe` | | `mentionOnReply` | No | Treat a reply to one of the bot's own messages as a mention, so it routes to `onNewMention`. Defaults to `false`. Auto-detected from `TELEGRAM_MENTION_ON_REPLY=true`. Implicit forum-topic replies and the bot's own messages never count | | `nativeStreaming` | No | Stream with Telegram's native draft previews in private chats. Defaults to `false`, which uses post-and-edit in every chat type | From 4ba410d415d041550ceaeccc73c26ed3f28152e5 Mon Sep 17 00:00:00 2001 From: Ben Sabic Date: Fri, 4 Sep 2026 13:36:31 +1000 Subject: [PATCH 5/5] docs(telegram): add business update fixtures and routing notes Split the business fixture into separate connection, customer message, bot-sent echo, and edited message updates. Document the telegram:biz thread format and business update routing in AGENTS.md. --- packages/adapter-telegram/AGENTS.md | 13 +++ packages/adapter-telegram/sample-messages.md | 83 +++++++++++++++++++- 2 files changed, 93 insertions(+), 3 deletions(-) diff --git a/packages/adapter-telegram/AGENTS.md b/packages/adapter-telegram/AGENTS.md index 053ca19f..2edeb1d4 100644 --- a/packages/adapter-telegram/AGENTS.md +++ b/packages/adapter-telegram/AGENTS.md @@ -93,10 +93,17 @@ adapter encodes both: ``` telegram:{chatId} # plain chat / DM telegram:{chatId}:{messageThreadId} # topic in a forum group +telegram:biz:{connectionId}:{chatId} # Connected Business Bot chat ``` `isDM(threadId)` returns `true` when `chat.type === "private"`. +Business threads are their own channel: `channelIdFromThreadId` returns +the full `telegram:biz:...` ID, never the bare `telegram:{chatId}`, so +they do not share lock or history keys with the customer's direct bot +chat. Note that sharders keyed on the first two segments group every +business thread under `telegram:biz`. + ## Webhook flow `TelegramAdapter.handleWebhook(request, options)` is the entry point. @@ -115,6 +122,12 @@ telegram:{chatId}:{messageThreadId} # topic in a forum group hooks. - `message_reaction` → `chat.handleReaction`. - `my_chat_member` / `chat_member` → membership change hooks. + - `business_connection` → cached in the state adapter (before the + `allowedUserIds` gate, since it carries no customer id). + - `business_message` / `edited_business_message` → resolve the + connection from state (falling back to `getBusinessConnection`), + drop owner-typed and `sender_business_bot` echoes, then route + slash commands, media groups, and messages like `message`. 3. **`waitUntil`** — outbound API calls (`sendMessage`, `editMessageText`, etc.) run inside `waitUntil` so the webhook response lands quickly. diff --git a/packages/adapter-telegram/sample-messages.md b/packages/adapter-telegram/sample-messages.md index d8682d2a..fe11f351 100644 --- a/packages/adapter-telegram/sample-messages.md +++ b/packages/adapter-telegram/sample-messages.md @@ -577,7 +577,11 @@ and the story id. } ``` -## Business message (Connected Business Bot) +## Business connection (Connected Business Bot) + +Sent when the owner connects the bot, changes its rights, or disconnects +it. Arrives on its own; Telegram never combines it with a message in one +update. ```json { @@ -594,9 +598,19 @@ and the story id. "date": 1756290700, "is_enabled": true, "rights": { - "can_reply": true + "can_reply": true, + "can_read_messages": true, + "can_delete_sent_messages": true } - }, + } +} +``` + +## Business message (customer to business account) + +```json +{ + "update_id": 312744901, "business_message": { "message_id": 42, "from": { @@ -617,3 +631,66 @@ and the story id. } } ``` + +## Business message (sent by the bot on behalf of the account) + +Outgoing replies are echoed back as `business_message`. `from` is the +account owner and `sender_business_bot` is the bot that sent it. The +adapter skips these. + +```json +{ + "update_id": 312744902, + "business_message": { + "message_id": 43, + "from": { + "id": 100000010, + "is_bot": false, + "first_name": "Business Owner", + "username": "owner" + }, + "sender_business_bot": { + "id": 100000099, + "is_bot": true, + "first_name": "Support Bot", + "username": "supportbot" + }, + "chat": { + "id": 100000001, + "first_name": "Customer", + "username": "customer", + "type": "private" + }, + "date": 1756290702, + "business_connection_id": "conn-example-abc", + "text": "Thanks, looking into your order now" + } +} +``` + +## Edited business message + +```json +{ + "update_id": 312744903, + "edited_business_message": { + "message_id": 42, + "from": { + "id": 100000001, + "is_bot": false, + "first_name": "Customer", + "username": "customer" + }, + "chat": { + "id": 100000001, + "first_name": "Customer", + "username": "customer", + "type": "private" + }, + "date": 1756290701, + "edit_date": 1756290760, + "business_connection_id": "conn-example-abc", + "text": "Hi, I need help with my order #4821" + } +} +```