From fd405ef86056cb9488689573147b15e7c480a946 Mon Sep 17 00:00:00 2001 From: Roman Stolyarchuk Date: Sun, 10 May 2026 08:46:08 +0300 Subject: [PATCH 1/2] feat(telegram): add business mode support and related configurations - Introduced business mode in Telegram settings to handle business messages. - Updated Telegram channel to process business messages and connections. - Enhanced message handling to include business connection IDs. - Added tests for business message handling and allowed updates. - Updated frontend to include business mode toggle in the Telegram form. - Added translations for business mode in multiple languages. --- pkg/channels/telegram/telegram.go | 310 ++++++++++++++---- .../telegram/telegram_dispatch_test.go | 97 ++++++ pkg/channels/telegram/telegram_test.go | 46 +++ pkg/config/config.go | 1 + pkg/config/config_channel_test.go | 3 + pkg/config/defaults.go | 1 + .../channels/channel-forms/telegram-form.tsx | 10 + web/frontend/src/i18n/locales/en.json | 2 + web/frontend/src/i18n/locales/pt-br.json | 2 + web/frontend/src/i18n/locales/zh.json | 2 + 10 files changed, 408 insertions(+), 66 deletions(-) diff --git a/pkg/channels/telegram/telegram.go b/pkg/channels/telegram/telegram.go index cebebfed6..707dcdbaa 100644 --- a/pkg/channels/telegram/telegram.go +++ b/pkg/channels/telegram/telegram.go @@ -43,6 +43,31 @@ var ( reInlineCode = regexp.MustCompile("`([^`]+)`") ) +func telegramAllowedUpdates(businessMode bool) []string { + updates := []string{ + telego.MessageUpdates, + telego.EditedMessageUpdates, + telego.ChannelPostUpdates, + telego.EditedChannelPostUpdates, + telego.CallbackQueryUpdates, + telego.InlineQueryUpdates, + telego.ChosenInlineResultUpdates, + telego.ShippingQueryUpdates, + telego.PreCheckoutQueryUpdates, + telego.PollUpdates, + telego.PollAnswerUpdates, + } + if businessMode { + updates = append(updates, + telego.BusinessConnectionUpdates, + telego.BusinessMessageUpdates, + telego.EditedBusinessMessageUpdates, + telego.DeletedBusinessMessagesUpdates, + ) + } + return updates +} + type TelegramChannel struct { *channels.BaseChannel bot *telego.Bot @@ -123,7 +148,8 @@ func (c *TelegramChannel) Start(ctx context.Context) error { c.ctx, c.cancel = context.WithCancel(ctx) updates, err := c.bot.UpdatesViaLongPolling(c.ctx, &telego.GetUpdatesParams{ - Timeout: 30, + Timeout: 30, + AllowedUpdates: telegramAllowedUpdates(c.businessModeEnabled()), }) if err != nil { c.cancel() @@ -140,6 +166,20 @@ func (c *TelegramChannel) Start(ctx context.Context) error { bh.HandleMessage(func(ctx *th.Context, message telego.Message) error { return c.handleMessage(ctx, &message) }, th.AnyMessage()) + if c.businessModeEnabled() { + bh.HandleBusinessMessage(func(ctx *th.Context, message telego.Message) error { + return c.handleBusinessMessage(ctx, &message) + }, th.AnyBusinessMessage()) + bh.HandleEditedBusinessMessage(func(ctx *th.Context, message telego.Message) error { + return c.handleBusinessMessage(ctx, &message) + }, th.AnyEditedBusinessMessage()) + bh.HandleBusinessConnection(func(ctx *th.Context, connection telego.BusinessConnection) error { + return c.handleBusinessConnection(ctx, connection) + }, th.AnyBusinessConnection()) + bh.HandleDeletedBusinessMessages(func(ctx *th.Context, deleted telego.BusinessMessagesDeleted) error { + return c.handleDeletedBusinessMessages(ctx, deleted) + }, th.AnyDeletedBusinessMessages()) + } c.SetRunning(true) logger.InfoCF("telegram", "Telegram bot connected", map[string]any{ @@ -189,7 +229,7 @@ func (c *TelegramChannel) Send(ctx context.Context, msg bus.OutboundMessage) ([] useMarkdownV2 := c.tgCfg.UseMarkdownV2 - chatID, threadID, err := resolveTelegramOutboundTarget(msg.ChatID, &msg.Context) + target, err := resolveTelegramOutboundTarget(msg.ChatID, &msg.Context) if err != nil { return nil, fmt.Errorf("invalid chat ID %s: %w", msg.ChatID, channels.ErrSendFailed) } @@ -253,8 +293,7 @@ func (c *TelegramChannel) Send(ctx context.Context, msg bus.OutboundMessage) ([] if smallerLen <= 0 { msgID, err := c.sendChunk(ctx, sendChunkParams{ - chatID: chatID, - threadID: threadID, + target: target, content: content, replyToID: replyToID, mdFallback: chunk, @@ -293,8 +332,7 @@ func (c *TelegramChannel) Send(ctx context.Context, msg bus.OutboundMessage) ([] } msgID, err := c.sendChunk(ctx, sendChunkParams{ - chatID: chatID, - threadID: threadID, + target: target, content: content, replyToID: replyToID, mdFallback: chunk, @@ -318,8 +356,7 @@ func (c *TelegramChannel) Send(ctx context.Context, msg bus.OutboundMessage) ([] } type sendChunkParams struct { - chatID int64 - threadID int + target telegramOutboundTarget content string replyToID string mdFallback string @@ -332,8 +369,9 @@ func (c *TelegramChannel) sendChunk( ctx context.Context, params sendChunkParams, ) (string, error) { - tgMsg := tu.Message(tu.ID(params.chatID), params.content) - tgMsg.MessageThreadID = params.threadID + tgMsg := tu.Message(tu.ID(params.target.chatID), params.content) + tgMsg.MessageThreadID = params.target.threadID + tgMsg.BusinessConnectionID = params.target.businessConnectionID if params.useMarkdownV2 { tgMsg.WithParseMode(telego.ModeMarkdownV2) } else { @@ -375,13 +413,14 @@ const maxTypingDuration = 5 * time.Minute // The goroutine also exits automatically after maxTypingDuration if cancel is // never called (e.g. when the LLM fails or times out without publishing). func (c *TelegramChannel) StartTyping(ctx context.Context, chatID string) (func(), error) { - cid, threadID, err := parseTelegramChatID(chatID) + target, err := resolveTelegramOutboundTarget(chatID, nil) if err != nil { return func() {}, err } - action := tu.ChatAction(tu.ID(cid), telego.ChatActionTyping) - action.MessageThreadID = threadID + action := tu.ChatAction(tu.ID(target.chatID), telego.ChatActionTyping) + action.MessageThreadID = target.threadID + action.BusinessConnectionID = target.businessConnectionID // Send the first typing action immediately _ = c.bot.SendChatAction(ctx, action) @@ -398,8 +437,9 @@ func (c *TelegramChannel) StartTyping(ctx context.Context, chatID string) (func( case <-maxCtx.Done(): return case <-ticker.C: - a := tu.ChatAction(tu.ID(cid), telego.ChatActionTyping) - a.MessageThreadID = threadID + a := tu.ChatAction(tu.ID(target.chatID), telego.ChatActionTyping) + a.MessageThreadID = target.threadID + a.BusinessConnectionID = target.businessConnectionID _ = c.bot.SendChatAction(typingCtx, a) } } @@ -411,7 +451,7 @@ func (c *TelegramChannel) StartTyping(ctx context.Context, chatID string) (func( // EditMessage implements channels.MessageEditor. func (c *TelegramChannel) EditMessage(ctx context.Context, chatID string, messageID string, content string) error { useMarkdownV2 := c.tgCfg.UseMarkdownV2 - cid, _, err := parseTelegramChatID(chatID) + target, err := resolveTelegramOutboundTarget(chatID, nil) if err != nil { return err } @@ -420,7 +460,8 @@ func (c *TelegramChannel) EditMessage(ctx context.Context, chatID string, messag return err } parsedContent := parseContent(content, useMarkdownV2) - editMsg := tu.EditMessageText(tu.ID(cid), mid, parsedContent) + editMsg := tu.EditMessageText(tu.ID(target.chatID), mid, parsedContent) + editMsg.BusinessConnectionID = target.businessConnectionID if useMarkdownV2 { editMsg.WithParseMode(telego.ModeMarkdownV2) } else { @@ -439,7 +480,9 @@ func (c *TelegramChannel) EditMessage(ctx context.Context, chatID string, messag // Network errors or timeouts should NOT trigger a retry with different content. if strings.Contains(err.Error(), "Bad Request") { logParseFailed(err, useMarkdownV2) - _, err = c.bot.EditMessageText(ctx, tu.EditMessageText(tu.ID(cid), mid, content)) + plainEdit := tu.EditMessageText(tu.ID(target.chatID), mid, content) + plainEdit.BusinessConnectionID = target.businessConnectionID + _, err = c.bot.EditMessageText(ctx, plainEdit) } } @@ -467,7 +510,7 @@ func (c *TelegramChannel) EditMessage(ctx context.Context, chatID string, messag // DeleteMessage implements channels.MessageDeleter. func (c *TelegramChannel) DeleteMessage(ctx context.Context, chatID string, messageID string) error { - cid, _, err := parseTelegramChatID(chatID) + target, err := resolveTelegramOutboundTarget(chatID, nil) if err != nil { return err } @@ -475,8 +518,14 @@ func (c *TelegramChannel) DeleteMessage(ctx context.Context, chatID string, mess if err != nil { return err } + if target.businessConnectionID != "" { + return c.bot.DeleteBusinessMessages(ctx, &telego.DeleteBusinessMessagesParams{ + BusinessConnectionID: target.businessConnectionID, + MessageIDs: []int{mid}, + }) + } return c.bot.DeleteMessage(ctx, &telego.DeleteMessageParams{ - ChatID: tu.ID(cid), + ChatID: tu.ID(target.chatID), MessageID: mid, }) } @@ -575,13 +624,14 @@ func (c *TelegramChannel) SendPlaceholder(ctx context.Context, chatID string) (s text := phCfg.GetRandomText() - cid, threadID, err := parseTelegramChatID(chatID) + target, err := resolveTelegramOutboundTarget(chatID, nil) if err != nil { return "", err } - phMsg := tu.Message(tu.ID(cid), text) - phMsg.MessageThreadID = threadID + phMsg := tu.Message(tu.ID(target.chatID), text) + phMsg.MessageThreadID = target.threadID + phMsg.BusinessConnectionID = target.businessConnectionID pMsg, err := c.bot.SendMessage(ctx, phMsg) if err != nil { return "", err @@ -598,7 +648,7 @@ func (c *TelegramChannel) SendMedia(ctx context.Context, msg bus.OutboundMediaMe trackedChatID := telegramToolFeedbackChatKey(msg.ChatID, &msg.Context) trackedMsgID, hasTrackedMsg := c.currentToolFeedbackMessage(trackedChatID) - chatID, threadID, err := resolveTelegramOutboundTarget(msg.ChatID, &msg.Context) + target, err := resolveTelegramOutboundTarget(msg.ChatID, &msg.Context) if err != nil { return nil, fmt.Errorf("invalid chat ID %s: %w", msg.ChatID, channels.ErrSendFailed) } @@ -632,10 +682,11 @@ func (c *TelegramChannel) SendMedia(ctx context.Context, msg bus.OutboundMediaMe switch part.Type { case "image": params := &telego.SendPhotoParams{ - ChatID: tu.ID(chatID), - MessageThreadID: threadID, - Photo: telego.InputFile{File: file}, - Caption: part.Caption, + BusinessConnectionID: target.businessConnectionID, + ChatID: tu.ID(target.chatID), + MessageThreadID: target.threadID, + Photo: telego.InputFile{File: file}, + Caption: part.Caption, } tgResult, err = c.bot.SendPhoto(ctx, params) if err != nil && strings.Contains(err.Error(), "PHOTO_INVALID_DIMENSIONS") { @@ -645,10 +696,11 @@ func (c *TelegramChannel) SendMedia(ctx context.Context, msg bus.OutboundMediaMe } docParams := &telego.SendDocumentParams{ - ChatID: tu.ID(chatID), - MessageThreadID: threadID, - Document: telego.InputFile{File: file}, - Caption: part.Caption, + BusinessConnectionID: target.businessConnectionID, + ChatID: tu.ID(target.chatID), + MessageThreadID: target.threadID, + Document: telego.InputFile{File: file}, + Caption: part.Caption, } tgResult, err = c.bot.SendDocument(ctx, docParams) } @@ -658,35 +710,39 @@ func (c *TelegramChannel) SendMedia(ctx context.Context, msg bus.OutboundMediaMe fn := strings.ToLower(part.Filename) if strings.Contains(fn, "voice") && (strings.HasSuffix(fn, ".ogg") || strings.HasSuffix(fn, ".oga")) { vparams := &telego.SendVoiceParams{ - ChatID: tu.ID(chatID), - MessageThreadID: threadID, - Voice: telego.InputFile{File: file}, - Caption: part.Caption, + BusinessConnectionID: target.businessConnectionID, + ChatID: tu.ID(target.chatID), + MessageThreadID: target.threadID, + Voice: telego.InputFile{File: file}, + Caption: part.Caption, } tgResult, err = c.bot.SendVoice(ctx, vparams) } else { params := &telego.SendAudioParams{ - ChatID: tu.ID(chatID), - MessageThreadID: threadID, - Audio: telego.InputFile{File: file}, - Caption: part.Caption, + BusinessConnectionID: target.businessConnectionID, + ChatID: tu.ID(target.chatID), + MessageThreadID: target.threadID, + Audio: telego.InputFile{File: file}, + Caption: part.Caption, } tgResult, err = c.bot.SendAudio(ctx, params) } case "video": params := &telego.SendVideoParams{ - ChatID: tu.ID(chatID), - MessageThreadID: threadID, - Video: telego.InputFile{File: file}, - Caption: part.Caption, + BusinessConnectionID: target.businessConnectionID, + ChatID: tu.ID(target.chatID), + MessageThreadID: target.threadID, + Video: telego.InputFile{File: file}, + Caption: part.Caption, } tgResult, err = c.bot.SendVideo(ctx, params) default: // "file" or unknown types params := &telego.SendDocumentParams{ - ChatID: tu.ID(chatID), - MessageThreadID: threadID, - Document: telego.InputFile{File: file}, - Caption: part.Caption, + BusinessConnectionID: target.businessConnectionID, + ChatID: tu.ID(target.chatID), + MessageThreadID: target.threadID, + Document: telego.InputFile{File: file}, + Caption: part.Caption, } tgResult, err = c.bot.SendDocument(ctx, params) } @@ -713,6 +769,52 @@ func (c *TelegramChannel) SendMedia(ctx context.Context, msg bus.OutboundMediaMe } func (c *TelegramChannel) handleMessage(ctx context.Context, message *telego.Message) error { + return c.handleTelegramMessage(ctx, message, "") +} + +func (c *TelegramChannel) businessModeEnabled() bool { + return c != nil && c.tgCfg != nil && c.tgCfg.BusinessMode +} + +func (c *TelegramChannel) handleBusinessMessage(ctx context.Context, message *telego.Message) error { + if !c.businessModeEnabled() { + return nil + } + if message == nil { + return fmt.Errorf("message is nil") + } + businessConnectionID := strings.TrimSpace(message.BusinessConnectionID) + if businessConnectionID == "" { + return fmt.Errorf("business message missing business_connection_id") + } + return c.handleTelegramMessage(ctx, message, businessConnectionID) +} + +func (c *TelegramChannel) handleBusinessConnection(_ context.Context, connection telego.BusinessConnection) error { + if !c.businessModeEnabled() { + return nil + } + logger.InfoCF("telegram", "Business connection updated", map[string]any{ + "business_connection_id": connection.ID, + "user_chat_id": connection.UserChatID, + "is_enabled": connection.IsEnabled, + }) + return nil +} + +func (c *TelegramChannel) handleDeletedBusinessMessages(_ context.Context, deleted telego.BusinessMessagesDeleted) error { + if !c.businessModeEnabled() { + return nil + } + logger.DebugCF("telegram", "Business messages deleted", map[string]any{ + "business_connection_id": deleted.BusinessConnectionID, + "chat_id": deleted.Chat.ID, + "message_count": len(deleted.MessageIDs), + }) + return nil +} + +func (c *TelegramChannel) handleTelegramMessage(ctx context.Context, message *telego.Message, businessConnectionID string) error { if message == nil { return fmt.Errorf("message is nil") } @@ -745,7 +847,7 @@ func (c *TelegramChannel) handleMessage(ctx context.Context, message *telego.Mes content := "" mediaPaths := []string{} - chatIDStr := fmt.Sprintf("%d", chatID) + chatIDStr := formatTelegramDeliveryChatID(chatID, businessConnectionID, 0) messageIDStr := fmt.Sprintf("%d", message.MessageID) scope := channels.BuildMediaScope("telegram", chatIDStr, messageIDStr) @@ -864,10 +966,10 @@ func (c *TelegramChannel) handleMessage(ctx context.Context, message *telego.Mes // route to the correct topic and each topic gets its own session. // Only forum groups (IsForum) are handled; regular group reply threads // must share one session per group. - compositeChatID := fmt.Sprintf("%d", chatID) + compositeChatID := formatTelegramDeliveryChatID(chatID, businessConnectionID, 0) threadID := message.MessageThreadID if message.Chat.IsForum && threadID != 0 { - compositeChatID = fmt.Sprintf("%d/%d", chatID, threadID) + compositeChatID = formatTelegramDeliveryChatID(chatID, businessConnectionID, threadID) } logger.DebugCF("telegram", "Received message", map[string]any{ @@ -889,16 +991,22 @@ func (c *TelegramChannel) handleMessage(ctx context.Context, message *telego.Mes "first_name": user.FirstName, "is_group": fmt.Sprintf("%t", message.Chat.Type != "private"), } + if businessConnectionID != "" { + metadata["business_connection_id"] = businessConnectionID + } inboundCtx := bus.InboundContext{ Channel: c.Name(), - ChatID: fmt.Sprintf("%d", chatID), + ChatID: formatTelegramDeliveryChatID(chatID, businessConnectionID, 0), ChatType: peerKind, SenderID: platformID, MessageID: messageID, Mentioned: isMentioned, Raw: metadata, } + if businessConnectionID != "" { + inboundCtx.Account = businessConnectionID + } if message.Chat.IsForum && threadID != 0 { inboundCtx.TopicID = fmt.Sprintf("%d", threadID) } @@ -1123,11 +1231,11 @@ func (c *TelegramChannel) PrepareToolFeedbackMessageContent(content string) stri } func telegramToolFeedbackChatKey(chatID string, outboundCtx *bus.InboundContext) string { - resolvedChatID, threadID, err := resolveTelegramOutboundTarget(chatID, outboundCtx) - if err != nil || threadID == 0 { + target, err := resolveTelegramOutboundTarget(chatID, outboundCtx) + if err != nil { return strings.TrimSpace(chatID) } - return fmt.Sprintf("%d/%d", resolvedChatID, threadID) + return target.String() } func (c *TelegramChannel) ToolFeedbackMessageChatID(chatID string, outboundCtx *bus.InboundContext) string { @@ -1153,26 +1261,93 @@ func parseTelegramChatID(chatID string) (int64, int, error) { return cid, tid, nil } -func resolveTelegramOutboundTarget(chatID string, outboundCtx *bus.InboundContext) (int64, int, error) { +type telegramOutboundTarget struct { + businessConnectionID string + chatID int64 + threadID int +} + +func (t telegramOutboundTarget) String() string { + return formatTelegramDeliveryChatID(t.chatID, t.businessConnectionID, t.threadID) +} + +func formatTelegramDeliveryChatID(chatID int64, businessConnectionID string, threadID int) string { + base := fmt.Sprintf("%d", chatID) + if strings.TrimSpace(businessConnectionID) != "" { + base = fmt.Sprintf("business:%s:%d", url.QueryEscape(strings.TrimSpace(businessConnectionID)), chatID) + } + if threadID != 0 { + return fmt.Sprintf("%s/%d", base, threadID) + } + return base +} + +func parseTelegramOutboundTarget(chatID string) (telegramOutboundTarget, error) { + targetChatID := strings.TrimSpace(chatID) + if strings.HasPrefix(targetChatID, "business:") { + rest := strings.TrimPrefix(targetChatID, "business:") + parts := strings.SplitN(rest, ":", 2) + if len(parts) != 2 || strings.TrimSpace(parts[0]) == "" || strings.TrimSpace(parts[1]) == "" { + return telegramOutboundTarget{}, fmt.Errorf("invalid business chat ID %q", chatID) + } + businessConnectionID, unescapeErr := url.QueryUnescape(strings.TrimSpace(parts[0])) + if unescapeErr != nil { + return telegramOutboundTarget{}, fmt.Errorf("invalid business connection ID in chat ID %q: %w", chatID, unescapeErr) + } + cid, threadID, err := parseTelegramChatID(parts[1]) + if err != nil { + return telegramOutboundTarget{}, err + } + return telegramOutboundTarget{ + businessConnectionID: businessConnectionID, + chatID: cid, + threadID: threadID, + }, nil + } + + cid, threadID, err := parseTelegramChatID(targetChatID) + if err != nil { + return telegramOutboundTarget{}, err + } + return telegramOutboundTarget{chatID: cid, threadID: threadID}, nil +} + +func businessConnectionIDFromContext(outboundCtx *bus.InboundContext) string { + if outboundCtx == nil { + return "" + } + if outboundCtx.Raw != nil { + if businessConnectionID := strings.TrimSpace(outboundCtx.Raw["business_connection_id"]); businessConnectionID != "" { + return businessConnectionID + } + } + return strings.TrimSpace(outboundCtx.Account) +} + +func resolveTelegramOutboundTarget(chatID string, outboundCtx *bus.InboundContext) (telegramOutboundTarget, error) { targetChatID := strings.TrimSpace(chatID) if targetChatID == "" && outboundCtx != nil { targetChatID = strings.TrimSpace(outboundCtx.ChatID) } - resolvedChatID, resolvedThreadID, err := parseTelegramChatID(targetChatID) + resolved, err := parseTelegramOutboundTarget(targetChatID) if err != nil { - return 0, 0, err + return telegramOutboundTarget{}, err } - if resolvedThreadID != 0 || outboundCtx == nil { - return resolvedChatID, resolvedThreadID, nil + if resolved.businessConnectionID == "" { + resolved.businessConnectionID = businessConnectionIDFromContext(outboundCtx) + } + if resolved.threadID != 0 || outboundCtx == nil { + return resolved, nil } topicID := strings.TrimSpace(outboundCtx.TopicID) if topicID == "" { - return resolvedChatID, resolvedThreadID, nil + return resolved, nil } if threadID, convErr := strconv.Atoi(topicID); convErr == nil { - return resolvedChatID, threadID, nil + resolved.threadID = threadID + return resolved, nil } - return resolvedChatID, resolvedThreadID, nil + return resolved, nil } func logParseFailed(err error, useMarkdownV2 bool) { @@ -1284,16 +1459,19 @@ func (c *TelegramChannel) BeginStream(ctx context.Context, chatID string) (chann return nil, fmt.Errorf("streaming disabled in config") } - cid, threadID, err := parseTelegramChatID(chatID) + target, err := resolveTelegramOutboundTarget(chatID, nil) if err != nil { return nil, err } + if target.businessConnectionID != "" { + return nil, fmt.Errorf("streaming is not supported for telegram business chats") + } streamCfg := c.tgCfg.Streaming return &telegramStreamer{ bot: c.bot, - chatID: cid, - threadID: threadID, + chatID: target.chatID, + threadID: target.threadID, draftID: cryptoRandInt(), throttleInterval: time.Duration(streamCfg.ThrottleSeconds) * time.Second, minGrowth: streamCfg.MinGrowthChars, diff --git a/pkg/channels/telegram/telegram_dispatch_test.go b/pkg/channels/telegram/telegram_dispatch_test.go index 0eb1de5ea..eae6fd0e3 100644 --- a/pkg/channels/telegram/telegram_dispatch_test.go +++ b/pkg/channels/telegram/telegram_dispatch_test.go @@ -2,12 +2,14 @@ package telegram import ( "context" + "strings" "testing" "github.com/mymmrac/telego" "github.com/sipeed/picoclaw/pkg/bus" "github.com/sipeed/picoclaw/pkg/channels" + "github.com/sipeed/picoclaw/pkg/config" ) func TestHandleMessage_DoesNotConsumeGenericCommandsLocally(t *testing.T) { @@ -16,6 +18,7 @@ func TestHandleMessage_DoesNotConsumeGenericCommandsLocally(t *testing.T) { BaseChannel: channels.NewBaseChannel("telegram", nil, messageBus, nil), chatIDs: make(map[string]int64), ctx: context.Background(), + tgCfg: &config.TelegramSettings{BusinessMode: true}, } msg := &telego.Message{ @@ -46,3 +49,97 @@ func TestHandleMessage_DoesNotConsumeGenericCommandsLocally(t *testing.T) { t.Fatalf("content=%q", inbound.Content) } } + +func TestHandleBusinessMessage_DisabledBusinessModeIgnoresMessage(t *testing.T) { + messageBus := bus.NewMessageBus() + ch := &TelegramChannel{ + BaseChannel: channels.NewBaseChannel("telegram", nil, messageBus, nil), + chatIDs: make(map[string]int64), + ctx: context.Background(), + tgCfg: &config.TelegramSettings{BusinessMode: false}, + } + + msg := &telego.Message{ + Text: "ignored business message", + MessageID: 18, + BusinessConnectionID: "biz-conn-1", + Chat: telego.Chat{ + ID: 777, + Type: "private", + }, + From: &telego.User{ + ID: 42, + FirstName: "Alice", + }, + } + + if err := ch.handleBusinessMessage(context.Background(), msg); err != nil { + t.Fatalf("handleBusinessMessage error: %v", err) + } + + select { + case inbound := <-messageBus.InboundChan(): + t.Fatalf("expected disabled business mode to ignore message, got %#v", inbound) + default: + } +} + +func TestTelegramAllowedUpdates_BusinessMode(t *testing.T) { + disabled := strings.Join(telegramAllowedUpdates(false), ",") + if strings.Contains(disabled, telego.BusinessMessageUpdates) { + t.Fatalf("disabled updates include business messages: %s", disabled) + } + + enabled := strings.Join(telegramAllowedUpdates(true), ",") + if !strings.Contains(enabled, telego.BusinessMessageUpdates) { + t.Fatalf("enabled updates do not include business messages: %s", enabled) + } + if !strings.Contains(enabled, telego.DeletedBusinessMessagesUpdates) { + t.Fatalf("enabled updates do not include deleted business messages: %s", enabled) + } +} + +func TestHandleBusinessMessage_ForwardsWithBusinessContext(t *testing.T) { + messageBus := bus.NewMessageBus() + ch := &TelegramChannel{ + BaseChannel: channels.NewBaseChannel("telegram", nil, messageBus, nil), + chatIDs: make(map[string]int64), + ctx: context.Background(), + tgCfg: &config.TelegramSettings{BusinessMode: true}, + } + + msg := &telego.Message{ + Text: "hello from business", + MessageID: 17, + BusinessConnectionID: "biz-conn-1", + Chat: telego.Chat{ + ID: 777, + Type: "private", + }, + From: &telego.User{ + ID: 42, + FirstName: "Alice", + }, + } + + if err := ch.handleBusinessMessage(context.Background(), msg); err != nil { + t.Fatalf("handleBusinessMessage error: %v", err) + } + + inbound, ok := <-messageBus.InboundChan() + if !ok { + t.Fatal("expected inbound message to be forwarded") + } + if inbound.ChatID != "business:biz-conn-1:777" { + t.Fatalf("chat_id=%q", inbound.ChatID) + } + if inbound.Context.Account != "biz-conn-1" { + t.Fatalf("account=%q", inbound.Context.Account) + } + if inbound.Context.Raw["business_connection_id"] != "biz-conn-1" { + t.Fatalf("business_connection_id=%q", inbound.Context.Raw["business_connection_id"]) + } + if inbound.Content != "hello from business" { + t.Fatalf("content=%q", inbound.Content) + } +} diff --git a/pkg/channels/telegram/telegram_test.go b/pkg/channels/telegram/telegram_test.go index 69c76b430..1393312ad 100644 --- a/pkg/channels/telegram/telegram_test.go +++ b/pkg/channels/telegram/telegram_test.go @@ -271,6 +271,41 @@ func TestSend_ShortMessage_SingleCall(t *testing.T) { assert.Len(t, caller.calls, 1, "short message should result in exactly one SendMessage call") } +func TestSend_BusinessMessageIncludesBusinessConnectionID(t *testing.T) { + caller := &stubCaller{ + callFn: func(ctx context.Context, url string, data *ta.RequestData) (*ta.Response, error) { + return successResponse(t), nil + }, + } + ch := newTestChannel(t, caller) + + _, err := ch.Send(context.Background(), bus.OutboundMessage{ + ChatID: "business:biz-conn-1:777", + Content: "hello business", + Context: bus.InboundContext{ + Channel: "telegram", + ChatID: "business:biz-conn-1:777", + Account: "biz-conn-1", + Raw: map[string]string{ + "business_connection_id": "biz-conn-1", + }, + }, + }) + + require.NoError(t, err) + require.Len(t, caller.calls, 1) + + var params struct { + ChatID int64 `json:"chat_id"` + BusinessConnectionID string `json:"business_connection_id"` + Text string `json:"text"` + } + require.NoError(t, json.Unmarshal(caller.calls[0].Data.BodyRaw, ¶ms)) + assert.Equal(t, int64(777), params.ChatID) + assert.Equal(t, "biz-conn-1", params.BusinessConnectionID) + assert.Equal(t, "hello business", params.Text) +} + func TestSend_NonToolFeedbackDeletesTrackedProgressMessage(t *testing.T) { caller := &stubCaller{ callFn: func(ctx context.Context, url string, data *ta.RequestData) (*ta.Response, error) { @@ -686,6 +721,17 @@ func TestParseTelegramChatID_InvalidThreadID(t *testing.T) { assert.Contains(t, err.Error(), "invalid thread ID") } +func TestParseTelegramOutboundTarget_BusinessConnectionIDWithColon(t *testing.T) { + formatted := formatTelegramDeliveryChatID(777, "biz:conn:1", 42) + + target, err := parseTelegramOutboundTarget(formatted) + + require.NoError(t, err) + assert.Equal(t, int64(777), target.chatID) + assert.Equal(t, 42, target.threadID) + assert.Equal(t, "biz:conn:1", target.businessConnectionID) +} + func TestSend_WithForumThreadID(t *testing.T) { caller := &stubCaller{ callFn: func(ctx context.Context, url string, data *ta.RequestData) (*ta.Response, error) { diff --git a/pkg/config/config.go b/pkg/config/config.go index c9d90e0f8..d96c6d7c5 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -363,6 +363,7 @@ type TelegramSettings struct { BaseURL string `json:"base_url" yaml:"-" env:"PICOCLAW_CHANNELS_TELEGRAM_BASE_URL"` Proxy string `json:"proxy" yaml:"-" env:"PICOCLAW_CHANNELS_TELEGRAM_PROXY"` Streaming StreamingConfig `json:"streaming,omitempty" yaml:"-"` + BusinessMode bool `json:"business_mode" yaml:"-" env:"PICOCLAW_CHANNELS_TELEGRAM_BUSINESS_MODE"` UseMarkdownV2 bool `json:"use_markdown_v2" yaml:"-" env:"PICOCLAW_CHANNELS_TELEGRAM_USE_MARKDOWN_V2"` } diff --git a/pkg/config/config_channel_test.go b/pkg/config/config_channel_test.go index fd3cd8246..e942ffe3f 100644 --- a/pkg/config/config_channel_test.go +++ b/pkg/config/config_channel_test.go @@ -17,6 +17,7 @@ import ( type testTelegramConfig struct { BaseURL string `json:"base_url" yaml:"-"` Proxy string `json:"proxy" yaml:"-"` + BusinessMode bool `json:"business_mode" yaml:"-"` UseMarkdownV2 bool `json:"use_markdown_v2" yaml:"-"` Streaming StreamingConfig `json:"streaming,omitempty" yaml:"-"` Token SecureString `json:"token,omitzero" yaml:"token,omitempty"` @@ -107,6 +108,7 @@ func TestChannel_JSON_Unmarshal(t *testing.T) { "reasoning_channel_id": "-100xxx", "settings": { "base_url": "https://custom-api.example.com", + "business_mode": true, "use_markdown_v2": true, "streaming": {"enabled": true, "throttle_seconds": 2}, "token": "[NOT_HERE]" @@ -126,6 +128,7 @@ func TestChannel_JSON_Unmarshal(t *testing.T) { var cfg testTelegramConfig require.NoError(t, ch.Decode(&cfg)) assert.Equal(t, "https://custom-api.example.com", cfg.BaseURL) + assert.True(t, cfg.BusinessMode) assert.True(t, cfg.UseMarkdownV2) assert.True(t, cfg.Streaming.Enabled) assert.Equal(t, 2, cfg.Streaming.ThrottleSeconds) diff --git a/pkg/config/defaults.go b/pkg/config/defaults.go index 8e2494ae5..d4def2790 100644 --- a/pkg/config/defaults.go +++ b/pkg/config/defaults.go @@ -497,6 +497,7 @@ func defaultChannels() ChannelsConfig { "placeholder": map[string]any{"enabled": true, "text": []string{"Thinking... 💭"}}, "settings": map[string]any{ "streaming": map[string]any{"enabled": true, "throttle_seconds": 3, "min_growth_chars": 200}, + "business_mode": false, "use_markdown_v2": false, }, }, diff --git a/web/frontend/src/components/channels/channel-forms/telegram-form.tsx b/web/frontend/src/components/channels/channel-forms/telegram-form.tsx index f9c7c778a..b45d3dd74 100644 --- a/web/frontend/src/components/channels/channel-forms/telegram-form.tsx +++ b/web/frontend/src/components/channels/channel-forms/telegram-form.tsx @@ -86,6 +86,16 @@ export function TelegramForm({ placeholder="https://api.telegram.org" /> + +
+ onChange("business_mode", checked)} + ariaLabel={t("channels.field.businessMode")} + /> +
diff --git a/web/frontend/src/i18n/locales/en.json b/web/frontend/src/i18n/locales/en.json index 985efdfe2..744d89050 100644 --- a/web/frontend/src/i18n/locales/en.json +++ b/web/frontend/src/i18n/locales/en.json @@ -375,6 +375,7 @@ "verificationToken": "Verification Token", "encryptKey": "Encrypt Key", "baseUrl": "API Base URL", + "businessMode": "Business Mode", "proxy": "HTTP Proxy", "mentionOnly": "Mention Only", "typingEnabled": "Typing Indicator", @@ -414,6 +415,7 @@ "verificationToken": "Verification token for event callbacks.", "encryptKey": "Encryption key used to decrypt callback payloads.", "baseUrl": "Platform API base URL. Official endpoint is used by default.", + "businessMode": "Receive and reply to Telegram Business messages for connected business accounts.", "proxy": "HTTP proxy address for outbound network access.", "mentionOnly": "Only respond when the bot is explicitly mentioned in group chats.", "typingEnabled": "Display typing status while the assistant is generating a response.", diff --git a/web/frontend/src/i18n/locales/pt-br.json b/web/frontend/src/i18n/locales/pt-br.json index c091625bb..c5fc59e83 100644 --- a/web/frontend/src/i18n/locales/pt-br.json +++ b/web/frontend/src/i18n/locales/pt-br.json @@ -367,6 +367,7 @@ "verificationToken": "Token de Verificação", "encryptKey": "Chave de Criptografia", "baseUrl": "URL Base da API", + "businessMode": "Modo Business", "proxy": "Proxy HTTP", "mentionOnly": "Apenas com Menção", "typingEnabled": "Indicador de Digitação", @@ -406,6 +407,7 @@ "verificationToken": "Token de verificação para callbacks de eventos.", "encryptKey": "Chave de criptografia usada para descriptografar payloads de callback.", "baseUrl": "URL base da API da plataforma. O endpoint oficial é usado por padrão.", + "businessMode": "Receber e responder mensagens do Telegram Business para contas comerciais conectadas.", "proxy": "Endereço de proxy HTTP para acesso de rede de saída.", "mentionOnly": "Responder apenas quando o bot for explicitamente mencionado em chats em grupo.", "typingEnabled": "Exibir status de digitação enquanto o assistente está gerando uma resposta.", diff --git a/web/frontend/src/i18n/locales/zh.json b/web/frontend/src/i18n/locales/zh.json index 9af5c39fe..6e70d5c76 100644 --- a/web/frontend/src/i18n/locales/zh.json +++ b/web/frontend/src/i18n/locales/zh.json @@ -375,6 +375,7 @@ "verificationToken": "Verification Token", "encryptKey": "Encrypt Key", "baseUrl": "API Base URL", + "businessMode": "Business Mode", "proxy": "HTTP 代理", "mentionOnly": "仅提及时响应", "typingEnabled": "输入中提示", @@ -414,6 +415,7 @@ "verificationToken": "事件回调验证令牌", "encryptKey": "消息加密密钥，用于解密回调内容", "baseUrl": "平台 API 地址，默认使用官方地址", + "businessMode": "接收并回复已连接商业账号的 Telegram Business 消息", "proxy": "HTTP 代理地址，用于网络访问", "mentionOnly": "在群聊中仅当明确提及时才响应", "typingEnabled": "在生成回复时显示“正在输入”状态", From 4aa6ffb0433f7a49a373406bc84bda9050341118 Mon Sep 17 00:00:00 2001 From: Roman Stolyarchuk Date: Sun, 10 May 2026 08:53:42 +0300 Subject: [PATCH 2/2] feat(telegram): mark business messages as read and add tests for handling - Implement markBusinessMessageRead function to mark messages as read - Add tests for successful message read and failure scenarios --- pkg/channels/telegram/telegram.go | 24 ++++++++ pkg/channels/telegram/telegram_test.go | 84 ++++++++++++++++++++++++++ 2 files changed, 108 insertions(+) diff --git a/pkg/channels/telegram/telegram.go b/pkg/channels/telegram/telegram.go index 707dcdbaa..85ca25142 100644 --- a/pkg/channels/telegram/telegram.go +++ b/pkg/channels/telegram/telegram.go @@ -787,9 +787,33 @@ func (c *TelegramChannel) handleBusinessMessage(ctx context.Context, message *te if businessConnectionID == "" { return fmt.Errorf("business message missing business_connection_id") } + c.markBusinessMessageRead(ctx, businessConnectionID, message.Chat.ID, message.MessageID) return c.handleTelegramMessage(ctx, message, businessConnectionID) } +func (c *TelegramChannel) markBusinessMessageRead( + ctx context.Context, + businessConnectionID string, + chatID int64, + messageID int, +) { + if c == nil || c.bot == nil || strings.TrimSpace(businessConnectionID) == "" || chatID == 0 || messageID == 0 { + return + } + if err := c.bot.ReadBusinessMessage(ctx, &telego.ReadBusinessMessageParams{ + BusinessConnectionID: businessConnectionID, + ChatID: chatID, + MessageID: messageID, + }); err != nil { + logger.DebugCF("telegram", "Failed to mark business message as read", map[string]any{ + "business_connection_id": businessConnectionID, + "chat_id": chatID, + "message_id": messageID, + "error": err.Error(), + }) + } +} + func (c *TelegramChannel) handleBusinessConnection(_ context.Context, connection telego.BusinessConnection) error { if !c.businessModeEnabled() { return nil diff --git a/pkg/channels/telegram/telegram_test.go b/pkg/channels/telegram/telegram_test.go index 1393312ad..a70d55c0b 100644 --- a/pkg/channels/telegram/telegram_test.go +++ b/pkg/channels/telegram/telegram_test.go @@ -150,6 +150,90 @@ func newTestChannelWithConstructor( } } +func TestHandleBusinessMessage_MarksMessageRead(t *testing.T) { + messageBus := bus.NewMessageBus() + caller := &stubCaller{ + callFn: func(ctx context.Context, url string, data *ta.RequestData) (*ta.Response, error) { + if !strings.Contains(url, "readBusinessMessage") { + t.Fatalf("unexpected API call: %s", url) + } + b, err := json.Marshal(true) + require.NoError(t, err) + return &ta.Response{Ok: true, Result: b}, nil + }, + } + ch := newTestChannel(t, caller) + ch.BaseChannel = channels.NewBaseChannel("telegram", nil, messageBus, nil) + ch.BaseChannel.SetOwner(ch) + ch.ctx = context.Background() + ch.tgCfg.BusinessMode = true + + msg := &telego.Message{ + Text: "hello from business", + MessageID: 17, + BusinessConnectionID: "biz-conn-1", + Chat: telego.Chat{ + ID: 777, + Type: "private", + }, + From: &telego.User{ + ID: 42, + FirstName: "Alice", + }, + } + + require.NoError(t, ch.handleBusinessMessage(context.Background(), msg)) + require.Len(t, caller.calls, 1) + assert.Contains(t, caller.calls[0].URL, "readBusinessMessage") + + var params struct { + BusinessConnectionID string `json:"business_connection_id"` + ChatID int64 `json:"chat_id"` + MessageID int `json:"message_id"` + } + require.NoError(t, json.Unmarshal(caller.calls[0].Data.BodyRaw, ¶ms)) + assert.Equal(t, "biz-conn-1", params.BusinessConnectionID) + assert.Equal(t, int64(777), params.ChatID) + assert.Equal(t, 17, params.MessageID) + + inbound := <-messageBus.InboundChan() + assert.Equal(t, "hello from business", inbound.Content) +} + +func TestHandleBusinessMessage_ReadFailureStillForwardsMessage(t *testing.T) { + messageBus := bus.NewMessageBus() + caller := &stubCaller{ + callFn: func(ctx context.Context, url string, data *ta.RequestData) (*ta.Response, error) { + return nil, errors.New("missing can_read_messages") + }, + } + ch := newTestChannel(t, caller) + ch.BaseChannel = channels.NewBaseChannel("telegram", nil, messageBus, nil) + ch.BaseChannel.SetOwner(ch) + ch.ctx = context.Background() + ch.tgCfg.BusinessMode = true + + msg := &telego.Message{ + Text: "still forward me", + MessageID: 18, + BusinessConnectionID: "biz-conn-1", + Chat: telego.Chat{ + ID: 777, + Type: "private", + }, + From: &telego.User{ + ID: 42, + FirstName: "Alice", + }, + } + + require.NoError(t, ch.handleBusinessMessage(context.Background(), msg)) + require.Len(t, caller.calls, 1) + + inbound := <-messageBus.InboundChan() + assert.Equal(t, "still forward me", inbound.Content) +} + func TestSendMedia_ImageFallbacksToDocumentOnInvalidDimensions(t *testing.T) { constructor := &multipartRecordingConstructor{} caller := &stubCaller{