diff --git a/docs/channels/telegram/README.md b/docs/channels/telegram/README.md index a4138009e..215d80afe 100644 --- a/docs/channels/telegram/README.md +++ b/docs/channels/telegram/README.md @@ -15,7 +15,8 @@ The Telegram channel uses long polling via the Telegram Bot API for bot-based co "token": "123456789:ABCdefGHIjklMNOpqrsTUVwxyz", "allow_from": ["123456789"], "proxy": "", - "use_markdown_v2": false + "use_markdown_v2": false, + "media_group_delay_ms": 500 } } } @@ -28,6 +29,7 @@ The Telegram channel uses long polling via the Telegram Bot API for bot-based co | allow_from | array | No | Allowlist of user IDs; empty means all users are allowed | | proxy | string | No | Proxy URL for connecting to the Telegram API (e.g. http://127.0.0.1:7890) | | use_markdown_v2 | bool | No | Enable Telegram MarkdownV2 formatting | +| media_group_delay_ms | int | No | Idle delay before processing Telegram media groups/albums. Defaults to 500 ms | ## Setup diff --git a/pkg/channels/telegram/telegram.go b/pkg/channels/telegram/telegram.go index cebebfed6..0965bcedc 100644 --- a/pkg/channels/telegram/telegram.go +++ b/pkg/channels/telegram/telegram.go @@ -11,6 +11,7 @@ import ( "net/url" "os" "regexp" + "slices" "strconv" "strings" "sync" @@ -43,20 +44,38 @@ var ( reInlineCode = regexp.MustCompile("`([^`]+)`") ) +const defaultMediaGroupDelay = 500 * time.Millisecond + type TelegramChannel struct { *channels.BaseChannel - bot *telego.Bot - bh *th.BotHandler - bc *config.Channel - chatIDs map[string]int64 - ctx context.Context - cancel context.CancelFunc - tgCfg *config.TelegramSettings - progress *channels.ToolFeedbackAnimator + bot *telego.Bot + bh *th.BotHandler + bc *config.Channel + chatIDsMu sync.Mutex + chatIDs map[string]int64 + ctx context.Context + cancel context.CancelFunc + tgCfg *config.TelegramSettings + progress *channels.ToolFeedbackAnimator registerFunc func(context.Context, []commands.Definition) error commandRegDelayFn func(int) time.Duration commandRegCancel context.CancelFunc + + mediaGroupMu sync.Mutex + mediaGroups map[string]*telegramMediaGroup + mediaGroupDelay time.Duration +} + +type telegramMediaGroup struct { + messages []*telego.Message + timer *time.Timer + generation uint64 +} + +type telegramMessageParts struct { + content []string + mediaPaths []string } func NewTelegramChannel( @@ -112,11 +131,21 @@ func NewTelegramChannel( bc: bc, chatIDs: make(map[string]int64), tgCfg: telegramCfg, + + mediaGroups: make(map[string]*telegramMediaGroup), + mediaGroupDelay: telegramMediaGroupDelay(telegramCfg), } ch.progress = channels.NewToolFeedbackAnimator(ch.EditMessage) return ch, nil } +func telegramMediaGroupDelay(telegramCfg *config.TelegramSettings) time.Duration { + if telegramCfg != nil && telegramCfg.MediaGroupDelayMS > 0 { + return time.Duration(telegramCfg.MediaGroupDelayMS) * time.Millisecond + } + return defaultMediaGroupDelay +} + func (c *TelegramChannel) Start(ctx context.Context) error { logger.InfoC("telegram", "Starting Telegram bot (polling mode)...") @@ -167,6 +196,7 @@ func (c *TelegramChannel) Stop(ctx context.Context) error { if c.bh != nil { _ = c.bh.StopWithContext(ctx) } + c.flushPendingMediaGroups(ctx) // Cancel our context (stops long polling) if c.cancel != nil { @@ -713,6 +743,131 @@ func (c *TelegramChannel) SendMedia(ctx context.Context, msg bus.OutboundMediaMe } func (c *TelegramChannel) handleMessage(ctx context.Context, message *telego.Message) error { + if message != nil && strings.TrimSpace(message.MediaGroupID) != "" { + return c.bufferMediaGroupMessage(ctx, message) + } + return c.handleMessages(ctx, []*telego.Message{message}) +} + +func (c *TelegramChannel) bufferMediaGroupMessage(ctx context.Context, message *telego.Message) error { + if message == nil { + return fmt.Errorf("message is nil") + } + groupID := strings.TrimSpace(message.MediaGroupID) + if groupID == "" { + return c.handleMessages(ctx, []*telego.Message{message}) + } + + msgCopy := *message + msgCopy.Photo = append([]telego.PhotoSize(nil), message.Photo...) + key := fmt.Sprintf("%d:%s", message.Chat.ID, groupID) + + c.mediaGroupMu.Lock() + if c.mediaGroups == nil { + c.mediaGroups = make(map[string]*telegramMediaGroup) + } + group := c.mediaGroups[key] + if group == nil { + group = &telegramMediaGroup{} + c.mediaGroups[key] = group + } + group.messages = append(group.messages, &msgCopy) + group.generation++ + generation := group.generation + if group.timer != nil { + group.timer.Stop() + } + delay := c.mediaGroupDelay + if delay <= 0 { + delay = defaultMediaGroupDelay + } + group.timer = time.AfterFunc(delay, func() { + c.flushMediaGroup(c.ctx, key, generation) + }) + c.mediaGroupMu.Unlock() + + logger.DebugCF("telegram", "Buffered media group message", map[string]any{ + "chat_id": message.Chat.ID, + "media_group_id": groupID, + "message_id": message.MessageID, + }) + return nil +} + +func (c *TelegramChannel) flushPendingMediaGroups(ctx context.Context) { + c.mediaGroupMu.Lock() + keys := make([]string, 0, len(c.mediaGroups)) + for key, group := range c.mediaGroups { + if group.timer != nil { + group.timer.Stop() + } + keys = append(keys, key) + } + c.mediaGroupMu.Unlock() + + for _, key := range keys { + c.flushMediaGroup(ctx, key, 0) + } +} + +func (c *TelegramChannel) flushMediaGroup(ctx context.Context, key string, generation uint64) { + c.mediaGroupMu.Lock() + group := c.mediaGroups[key] + if group == nil { + c.mediaGroupMu.Unlock() + return + } + if generation != 0 && group.generation != generation { + c.mediaGroupMu.Unlock() + return + } + delete(c.mediaGroups, key) + if group.timer != nil { + group.timer.Stop() + } + messages := append([]*telego.Message(nil), group.messages...) + c.mediaGroupMu.Unlock() + + if len(messages) == 0 { + return + } + slices.SortFunc(messages, func(a, b *telego.Message) int { + switch { + case a == nil && b == nil: + return 0 + case a == nil: + return -1 + case b == nil: + return 1 + default: + return a.MessageID - b.MessageID + } + }) + if ctx == nil { + ctx = context.Background() + } + if err := c.handleMessages(ctx, messages); err != nil { + logger.ErrorCF("telegram", "Failed to handle media group", map[string]any{ + "key": key, + "error": err.Error(), + }) + } +} + +func (c *TelegramChannel) handleMessages(ctx context.Context, messages []*telego.Message) error { + if len(messages) == 0 { + return nil + } + message := messages[0] + for _, candidate := range messages { + if candidate == nil { + continue + } + if strings.TrimSpace(candidate.Text) != "" || strings.TrimSpace(candidate.Caption) != "" { + message = candidate + break + } + } if message == nil { return fmt.Errorf("message is nil") } @@ -740,7 +895,9 @@ func (c *TelegramChannel) handleMessage(ctx context.Context, message *telego.Mes } chatID := message.Chat.ID + c.chatIDsMu.Lock() c.chatIDs[platformID] = chatID + c.chatIDsMu.Unlock() content := "" mediaPaths := []string{} @@ -764,61 +921,18 @@ func (c *TelegramChannel) handleMessage(ctx context.Context, message *telego.Mes return localPath // fallback: use raw path } - if message.Text != "" { - content += message.Text - } - - if message.Caption != "" { - if content != "" { - content += "\n" + for i, msg := range messages { + if msg == nil { + continue } - content += message.Caption - } - - if len(message.Photo) > 0 { - photo := message.Photo[len(message.Photo)-1] - photoPath := c.downloadPhoto(ctx, photo.FileID) - if photoPath != "" { - mediaPaths = append(mediaPaths, storeMedia(photoPath, "photo.jpg")) + parts := c.collectTelegramMessageParts(ctx, msg, i, len(messages), storeMedia) + for _, part := range parts.content { if content != "" { content += "\n" } - content += "[image: photo]" - } - } - - if message.Voice != nil { - voicePath := c.downloadFile(ctx, message.Voice.FileID, ".ogg") - if voicePath != "" { - mediaPaths = append(mediaPaths, storeMedia(voicePath, "voice.ogg")) - - if content != "" { - content += "\n" - } - content += "[voice]" - } - } - - if message.Audio != nil { - audioPath := c.downloadFile(ctx, message.Audio.FileID, ".mp3") - if audioPath != "" { - mediaPaths = append(mediaPaths, storeMedia(audioPath, "audio.mp3")) - if content != "" { - content += "\n" - } - content += "[audio]" - } - } - - if message.Document != nil { - docPath := c.downloadFile(ctx, message.Document.FileID, "") - if docPath != "" { - mediaPaths = append(mediaPaths, storeMedia(docPath, "document")) - if content != "" { - content += "\n" - } - content += "[file]" + content += part } + mediaPaths = append(mediaPaths, parts.mediaPaths...) } if content == "" && len(mediaPaths) == 0 { @@ -917,6 +1031,74 @@ func (c *TelegramChannel) handleMessage(ctx context.Context, message *telego.Mes return nil } +func (c *TelegramChannel) collectTelegramMessageParts( + ctx context.Context, + msg *telego.Message, + index int, + total int, + storeMedia func(localPath, filename string) string, +) telegramMessageParts { + parts := telegramMessageParts{} + if msg == nil { + return parts + } + if text := strings.TrimSpace(msg.Text); text != "" { + parts.content = append(parts.content, text) + } + if caption := strings.TrimSpace(msg.Caption); caption != "" { + parts.content = append(parts.content, caption) + } + if len(msg.Photo) > 0 { + photo := msg.Photo[len(msg.Photo)-1] + photoPath := c.downloadPhoto(ctx, photo.FileID) + if photoPath != "" { + photoNumber := index + 1 + parts.mediaPaths = append(parts.mediaPaths, storeMedia(photoPath, fmt.Sprintf("photo-%d.jpg", photoNumber))) + parts.content = append(parts.content, fmt.Sprintf("[image: photo %d]", photoNumber)) + } + } + if msg.Voice != nil { + voicePath := c.downloadFile(ctx, msg.Voice.FileID, ".ogg") + if voicePath != "" { + parts.mediaPaths = append( + parts.mediaPaths, + storeMedia(voicePath, indexedMediaFilename("voice", ".ogg", index, total)), + ) + parts.content = append(parts.content, "[voice]") + } + } + if msg.Audio != nil { + audioPath := c.downloadFile(ctx, msg.Audio.FileID, ".mp3") + if audioPath != "" { + filename := msg.Audio.FileName + if strings.TrimSpace(filename) == "" { + filename = indexedMediaFilename("audio", ".mp3", index, total) + } + parts.mediaPaths = append(parts.mediaPaths, storeMedia(audioPath, filename)) + parts.content = append(parts.content, "[audio]") + } + } + if msg.Document != nil { + docPath := c.downloadFile(ctx, msg.Document.FileID, "") + if docPath != "" { + filename := msg.Document.FileName + if strings.TrimSpace(filename) == "" { + filename = indexedMediaFilename("document", "", index, total) + } + parts.mediaPaths = append(parts.mediaPaths, storeMedia(docPath, filename)) + parts.content = append(parts.content, "[file]") + } + } + return parts +} + +func indexedMediaFilename(prefix, ext string, index int, total int) string { + if total <= 1 { + return prefix + ext + } + return fmt.Sprintf("%s-%d%s", prefix, index+1, ext) +} + func (c *TelegramChannel) prependTelegramQuotedReply(content string, reply *telego.Message) string { quoted := strings.TrimSpace(telegramQuotedContent(reply)) if quoted == "" { diff --git a/pkg/channels/telegram/telegram_test.go b/pkg/channels/telegram/telegram_test.go index 69c76b430..14d025064 100644 --- a/pkg/channels/telegram/telegram_test.go +++ b/pkg/channels/telegram/telegram_test.go @@ -10,6 +10,7 @@ import ( "strconv" "strings" "testing" + "time" "github.com/mymmrac/telego" ta "github.com/mymmrac/telego/telegoapi" @@ -1100,3 +1101,190 @@ func TestHandleMessage_EmptyContent_Ignored(t *testing.T) { default: } } + +func TestHandleMessage_MediaGroupCombinesCaptionMessages(t *testing.T) { + messageBus, ch := newMediaGroupTestChannel(10 * time.Millisecond) + base := testMediaGroupMessage("album-1") + first := base + first.MessageID = 1 + second := base + second.MessageID = 2 + second.Caption = "meal caption" + + require.NoError(t, ch.handleMessage(context.Background(), &first)) + require.NoError(t, ch.handleMessage(context.Background(), &second)) + + select { + case inbound := <-messageBus.InboundChan(): + assert.Equal(t, "2", inbound.Context.MessageID) + assert.Equal(t, "meal caption", inbound.Content) + case <-time.After(time.Second): + t.Fatal("timed out waiting for combined media group message") + } +} + +func TestHandleMessage_MediaGroupWaitsForStaggeredMessages(t *testing.T) { + messageBus, ch := newMediaGroupTestChannel(100 * time.Millisecond) + base := testMediaGroupMessage("album-staggered") + first := base + first.MessageID = 1 + first.Caption = "first caption" + second := base + second.MessageID = 2 + second.Caption = "second caption" + + require.NoError(t, ch.handleMessage(context.Background(), &first)) + time.Sleep(50 * time.Millisecond) + require.NoError(t, ch.handleMessage(context.Background(), &second)) + + select { + case inbound := <-messageBus.InboundChan(): + t.Fatalf("media group flushed before idle delay reset: %#v", inbound) + case <-time.After(75 * time.Millisecond): + } + + select { + case inbound := <-messageBus.InboundChan(): + assert.Equal(t, "1", inbound.Context.MessageID) + assert.Equal(t, "first caption\nsecond caption", inbound.Content) + case <-time.After(time.Second): + t.Fatal("timed out waiting for staggered media group message") + } +} + +func TestFlushMediaGroupIgnoresStaleTimerGeneration(t *testing.T) { + messageBus, ch := newMediaGroupTestChannel(time.Hour) + base := testMediaGroupMessage("album-generation") + first := base + first.MessageID = 1 + first.Caption = "first" + second := base + second.MessageID = 2 + second.Caption = "second" + key := "456:album-generation" + + ch.mediaGroupMu.Lock() + ch.mediaGroups[key] = &telegramMediaGroup{ + messages: []*telego.Message{&first, &second}, + generation: 2, + } + ch.mediaGroupMu.Unlock() + + ch.flushMediaGroup(context.Background(), key, 1) + + select { + case inbound := <-messageBus.InboundChan(): + t.Fatalf("stale media group generation flushed unexpectedly: %#v", inbound) + default: + } + + ch.mediaGroupMu.Lock() + _, stillPending := ch.mediaGroups[key] + ch.mediaGroupMu.Unlock() + require.True(t, stillPending, "stale flush should leave the current batch pending") + + ch.flushMediaGroup(context.Background(), key, 2) + + select { + case inbound := <-messageBus.InboundChan(): + assert.Equal(t, "1", inbound.Context.MessageID) + assert.Equal(t, "first\nsecond", inbound.Content) + case <-time.After(time.Second): + t.Fatal("timed out waiting for current generation media group flush") + } +} + +func TestHandleMessage_MediaGroupAfterDelayStartsNewBatch(t *testing.T) { + messageBus, ch := newMediaGroupTestChannel(10 * time.Millisecond) + base := testMediaGroupMessage("album-split") + first := base + first.MessageID = 1 + first.Caption = "first" + second := base + second.MessageID = 2 + second.Caption = "second" + + require.NoError(t, ch.handleMessage(context.Background(), &first)) + select { + case inbound := <-messageBus.InboundChan(): + assert.Equal(t, "1", inbound.Context.MessageID) + assert.Equal(t, "first", inbound.Content) + case <-time.After(time.Second): + t.Fatal("timed out waiting for first media group batch") + } + + require.NoError(t, ch.handleMessage(context.Background(), &second)) + select { + case inbound := <-messageBus.InboundChan(): + assert.Equal(t, "2", inbound.Context.MessageID) + assert.Equal(t, "second", inbound.Content) + case <-time.After(time.Second): + t.Fatal("timed out waiting for second media group batch") + } +} + +func TestStopFlushesPendingMediaGroups(t *testing.T) { + messageBus, ch := newMediaGroupTestChannel(time.Hour) + base := testMediaGroupMessage("album-stop") + msg := base + msg.MessageID = 1 + msg.Caption = "caption before stop" + + require.NoError(t, ch.handleMessage(context.Background(), &msg)) + require.NoError(t, ch.Stop(context.Background())) + + select { + case inbound := <-messageBus.InboundChan(): + assert.Equal(t, "1", inbound.Context.MessageID) + assert.Equal(t, "caption before stop", inbound.Content) + case <-time.After(time.Second): + t.Fatal("timed out waiting for pending media group flush on stop") + } +} + +func TestNewTelegramChannelUsesConfiguredMediaGroupDelay(t *testing.T) { + ch, err := NewTelegramChannel( + &config.Channel{Type: config.ChannelTelegram, Enabled: true}, + &config.TelegramSettings{ + Token: *config.NewSecureString(testToken), + MediaGroupDelayMS: 750, + }, + bus.NewMessageBus(), + ) + require.NoError(t, err) + assert.Equal(t, 750*time.Millisecond, ch.mediaGroupDelay) + + ch, err = NewTelegramChannel( + &config.Channel{Type: config.ChannelTelegram, Enabled: true}, + &config.TelegramSettings{Token: *config.NewSecureString(testToken)}, + bus.NewMessageBus(), + ) + require.NoError(t, err) + assert.Equal(t, defaultMediaGroupDelay, ch.mediaGroupDelay) +} + +func newMediaGroupTestChannel(delay time.Duration) (*bus.MessageBus, *TelegramChannel) { + messageBus := bus.NewMessageBus() + ch := &TelegramChannel{ + BaseChannel: channels.NewBaseChannel("telegram", nil, messageBus, nil), + chatIDs: make(map[string]int64), + ctx: context.Background(), + mediaGroups: make(map[string]*telegramMediaGroup), + mediaGroupDelay: delay, + } + return messageBus, ch +} + +func testMediaGroupMessage(mediaGroupID string) telego.Message { + return telego.Message{ + Chat: telego.Chat{ + ID: 456, + Type: "private", + }, + From: &telego.User{ + ID: 789, + FirstName: "User", + }, + MediaGroupID: mediaGroupID, + } +} diff --git a/pkg/config/config.go b/pkg/config/config.go index 0a347e3db..49a8028c6 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -480,11 +480,12 @@ type WhatsAppSettings struct { } type TelegramSettings struct { - Token SecureString `json:"token,omitzero" yaml:"token,omitempty" env:"PICOCLAW_CHANNELS_TELEGRAM_TOKEN"` - 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:"-"` - UseMarkdownV2 bool `json:"use_markdown_v2" yaml:"-" env:"PICOCLAW_CHANNELS_TELEGRAM_USE_MARKDOWN_V2"` + Token SecureString `json:"token,omitzero" yaml:"token,omitempty" env:"PICOCLAW_CHANNELS_TELEGRAM_TOKEN"` + 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:"-"` + UseMarkdownV2 bool `json:"use_markdown_v2" yaml:"-" env:"PICOCLAW_CHANNELS_TELEGRAM_USE_MARKDOWN_V2"` + MediaGroupDelayMS int `json:"media_group_delay_ms" yaml:"-" env:"PICOCLAW_CHANNELS_TELEGRAM_MEDIA_GROUP_DELAY_MS"` } type FeishuSettings struct { diff --git a/pkg/config/defaults.go b/pkg/config/defaults.go index 1742a4b87..588184adf 100644 --- a/pkg/config/defaults.go +++ b/pkg/config/defaults.go @@ -503,8 +503,9 @@ func defaultChannels() ChannelsConfig { "typing": map[string]any{"enabled": true}, "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}, - "use_markdown_v2": false, + "streaming": map[string]any{"enabled": true, "throttle_seconds": 3, "min_growth_chars": 200}, + "use_markdown_v2": false, + "media_group_delay_ms": 500, }, }, "feishu": map[string]any{},