Merge pull request #2758 from bogdanovich/codex/telegram-media-groups

fix(telegram): media group album handling

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
 

pkg/channels/telegram/telegram.go

@@ -11,6 +11,7 @@ import (
 	"net/url"
 	"os"
 	"regexp"
+	"slices"
 	"strconv"
 	"strings"
 	"sync"
@@ -43,11 +44,14 @@ var (
 	reInlineCode = regexp.MustCompile("`([^`]+)`")
 )
 
+const defaultMediaGroupDelay = 500 * time.Millisecond
+
 type TelegramChannel struct {
 	*channels.BaseChannel
 	bot       *telego.Bot
 	bh        *th.BotHandler
 	bc        *config.Channel
+	chatIDsMu sync.Mutex
 	chatIDs   map[string]int64
 	ctx       context.Context
 	cancel    context.CancelFunc
@@ -57,6 +61,21 @@ type TelegramChannel struct {
 	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 != "" {
+	for i, msg := range messages {
-		content += message.Text
+		if msg == nil {
+			continue
 		}
-
+		parts := c.collectTelegramMessageParts(ctx, msg, i, len(messages), storeMedia)
-	if message.Caption != "" {
+		for _, part := range parts.content {
 			if content != "" {
 				content += "\n"
 			}
-		content += message.Caption
+			content += part
-	}
-
-	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"))
-			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]"
 		}
+		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 == "" {

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,
+	}
+}

pkg/config/config.go

@@ -485,6 +485,7 @@ type TelegramSettings struct {
 	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 {

pkg/config/defaults.go

@@ -505,6 +505,7 @@ func defaultChannels() ChannelsConfig {
 			"settings": map[string]any{
 				"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{},