From ffb6786ea4bf7dc0aa487d8b92a3d53213ff0ba5 Mon Sep 17 00:00:00 2001 From: viettranx Date: Fri, 27 Mar 2026 10:39:10 +0700 Subject: [PATCH] feat(telegram): lazy-resolve media from pending history on mention When users send media in a group without mentioning the bot, store Telegram file_ids as lightweight MediaRef in history entries (no download). When the bot is mentioned, resolve refs by downloading media and including them in the LLM context. Safeguards: 5 MB file size cap, max 15 refs per mention, 30s batch timeout. Mirrors existing CollectMedia pattern from Discord/Zalo. --- internal/channels/history.go | 28 +++++++- internal/channels/telegram/handlers.go | 20 ++++++ internal/channels/telegram/media.go | 99 ++++++++++++++++++++++++++ 3 files changed, 146 insertions(+), 1 deletion(-) diff --git a/internal/channels/history.go b/internal/channels/history.go index 1ea5d96b..36692857 100644 --- a/internal/channels/history.go +++ b/internal/channels/history.go @@ -32,12 +32,22 @@ const ( compactSweepInterval = 10 * time.Minute // periodic compaction sweep for post-restart safety ) +// MediaRef is a lightweight reference to platform media for deferred download. +// Stored in RAM only (not persisted to DB) — used by channels that defer media +// download until the bot is actually mentioned (e.g. Telegram). +type MediaRef struct { + Type string // "image", "video", "audio", "voice", "document", "animation" + FileID string // platform-specific file ID for lazy download + FileSize int64 // file size in bytes (0 if unknown) — used to skip large files +} + // HistoryEntry represents a single tracked group message. type HistoryEntry struct { Sender string SenderID string Body string - Media []string // temp file paths for images/attachments (RAM-only, not persisted to DB) + Media []string // temp file paths for images/attachments (RAM-only, not persisted to DB) + MediaRefs []MediaRef // deferred media refs for lazy download (RAM-only, not persisted) Timestamp time.Time MessageID string } @@ -336,6 +346,22 @@ func (ph *PendingHistory) CollectMedia(historyKey string) []string { return paths } +// CollectMediaRefs returns all deferred media references from pending entries +// and removes them from the entries to prevent double-processing. +// Used by channels that defer media download until the bot is mentioned. +func (ph *PendingHistory) CollectMediaRefs(historyKey string) []MediaRef { + ph.mu.Lock() + defer ph.mu.Unlock() + + entries := ph.entries[historyKey] + var refs []MediaRef + for i := range entries { + refs = append(refs, entries[i].MediaRefs...) + entries[i].MediaRefs = nil // prevent double-processing + } + return refs +} + // cleanupMedia removes temp files from history entries. Best-effort, logs warnings. func cleanupMedia(entries []HistoryEntry) { for _, e := range entries { diff --git a/internal/channels/telegram/handlers.go b/internal/channels/telegram/handlers.go index 1b95648f..d1ba7a9c 100644 --- a/internal/channels/telegram/handlers.go +++ b/internal/channels/telegram/handlers.go @@ -267,6 +267,7 @@ func (c *Channel) handleMessage(ctx context.Context, update telego.Update) { Sender: senderLabel, SenderID: senderID, Body: content, + MediaRefs: extractMediaRefs(message), Timestamp: time.Unix(int64(message.Date), 0), MessageID: fmt.Sprintf("%d", message.MessageID), }, c.historyLimit) @@ -322,6 +323,7 @@ func (c *Channel) handleMessage(ctx context.Context, update telego.Update) { Sender: senderLabel, SenderID: senderID, Body: content, + MediaRefs: extractMediaRefs(message), Timestamp: time.Unix(int64(message.Date), 0), MessageID: fmt.Sprintf("%d", message.MessageID), }, c.historyLimit) @@ -467,6 +469,24 @@ func (c *Channel) handleMessage(ctx context.Context, update telego.Update) { if isGroup { annotated := fmt.Sprintf("[From: %s]\n%s", senderLabel, content) if c.historyLimit > 0 { + // Resolve deferred media from history entries (lazy download). + if histRefs := c.groupHistory.CollectMediaRefs(localKey); len(histRefs) > 0 { + histMedia, histErrors := c.resolveMediaRefs(ctx, histRefs) + for _, m := range histMedia { + mediaFiles = append(mediaFiles, bus.MediaFile{ + Path: m.FilePath, + MimeType: m.ContentType, + }) + } + if len(histMedia) > 0 { + histTags := buildMediaTags(histMedia) + annotated = histTags + "\n\n" + annotated + } + for _, e := range histErrors { + slog.Warn("telegram: history media download failed", + "type", e.Type, "reason", e.Reason) + } + } finalContent = c.groupHistory.BuildContext(localKey, annotated, c.historyLimit) } else { finalContent = annotated diff --git a/internal/channels/telegram/media.go b/internal/channels/telegram/media.go index f8e77797..d83a6ec1 100644 --- a/internal/channels/telegram/media.go +++ b/internal/channels/telegram/media.go @@ -14,6 +14,7 @@ import ( "github.com/mymmrac/telego" + "github.com/nextlevelbuilder/goclaw/internal/channels" "github.com/nextlevelbuilder/goclaw/internal/channels/media" "github.com/nextlevelbuilder/goclaw/internal/tools" ) @@ -398,6 +399,104 @@ func lightweightMediaTags(msg *telego.Message) string { return strings.Join(tags, "\n") } +// extractMediaRefs extracts lightweight media references (file_ids + sizes) from a Telegram +// message without downloading any files. Stored in HistoryEntry.MediaRefs for lazy download +// when the bot is later mentioned. +func extractMediaRefs(msg *telego.Message) []channels.MediaRef { + var refs []channels.MediaRef + if msg.Photo != nil && len(msg.Photo) > 0 { + photo := msg.Photo[len(msg.Photo)-1] // highest resolution + refs = append(refs, channels.MediaRef{Type: "image", FileID: photo.FileID, FileSize: int64(photo.FileSize)}) + } + if msg.Video != nil { + refs = append(refs, channels.MediaRef{Type: "video", FileID: msg.Video.FileID, FileSize: int64(msg.Video.FileSize)}) + } + if msg.VideoNote != nil { + refs = append(refs, channels.MediaRef{Type: "video", FileID: msg.VideoNote.FileID, FileSize: int64(msg.VideoNote.FileSize)}) + } + if msg.Animation != nil { + refs = append(refs, channels.MediaRef{Type: "animation", FileID: msg.Animation.FileID, FileSize: int64(msg.Animation.FileSize)}) + } + if msg.Audio != nil { + refs = append(refs, channels.MediaRef{Type: "audio", FileID: msg.Audio.FileID, FileSize: int64(msg.Audio.FileSize)}) + } + if msg.Voice != nil { + refs = append(refs, channels.MediaRef{Type: "voice", FileID: msg.Voice.FileID, FileSize: int64(msg.Voice.FileSize)}) + } + if msg.Document != nil { + refs = append(refs, channels.MediaRef{Type: "document", FileID: msg.Document.FileID, FileSize: int64(msg.Document.FileSize)}) + } + return refs +} + +// historyMediaMaxBytes is the max file size for deferred history media downloads. +// Caps large files (videos, big documents) to prevent slow mention handling. +const historyMediaMaxBytes int64 = 5 * 1024 * 1024 // 5 MB + +// maxHistoryMediaRefs is the max number of deferred media refs to resolve per mention. +// Caps total download time — at worst ~2s each = ~30s for 15 files. +const maxHistoryMediaRefs = 15 + +// resolveMediaRefs downloads media from deferred file_id references stored in pending history. +// Used to resolve history media when the bot is mentioned. +// Caps at maxHistoryMediaRefs most-recent refs and skips files exceeding historyMediaMaxBytes. +func (c *Channel) resolveMediaRefs(ctx context.Context, refs []channels.MediaRef) ([]MediaInfo, []MediaError) { + // Only resolve the most recent refs to avoid blocking mention handling. + if len(refs) > maxHistoryMediaRefs { + slog.Debug("telegram: capping history media refs", + "total", len(refs), "cap", maxHistoryMediaRefs) + refs = refs[len(refs)-maxHistoryMediaRefs:] + } + + maxBytes := c.config.MediaMaxBytes + if maxBytes == 0 { + if c.config.APIServer != "" { + maxBytes = localAPIDefaultMaxBytes + } else { + maxBytes = defaultMediaMaxBytes + } + } + // Use the stricter of channel config and history cap. + if historyMediaMaxBytes < maxBytes { + maxBytes = historyMediaMaxBytes + } + + // Batch timeout: abort remaining downloads if total time exceeds limit. + batchCtx, cancel := context.WithTimeout(ctx, 30*time.Second) + defer cancel() + + var results []MediaInfo + var errs []MediaError + for _, ref := range refs { + // Pre-flight size check — skip without downloading if known to exceed limit. + if ref.FileSize > 0 && ref.FileSize > maxBytes { + slog.Debug("telegram: skipping oversized history media ref", + "type", ref.Type, "size", ref.FileSize, "max", maxBytes) + errs = append(errs, MediaError{Type: ref.Type, Reason: "file too large for history resolve", MaxBytes: maxBytes}) + continue + } + filePath, err := c.downloadMedia(batchCtx, ref.FileID, maxBytes) + if err != nil { + // On batch timeout, stop processing remaining refs. + if batchCtx.Err() != nil { + slog.Warn("telegram: history media batch timeout, skipping remaining", + "resolved", len(results), "remaining", len(refs)) + break + } + slog.Warn("telegram: history media ref download failed", + "type", ref.Type, "file_id", ref.FileID, "error", err) + errs = append(errs, newMediaError(ref.Type, err, maxBytes)) + continue + } + results = append(results, MediaInfo{ + Type: ref.Type, + FilePath: filePath, + FileID: ref.FileID, + }) + } + return results, errs +} + // lightweightTagForType returns the single lightweight tag that matches a given media type // within a Telegram message. Used for targeted replacement when a specific media fails. func lightweightTagForType(mediaType string, msg *telego.Message) string {