diff --git a/pkg/channels/telegram/events.go b/pkg/channels/telegram/events.go new file mode 100644 index 000000000..43626703d --- /dev/null +++ b/pkg/channels/telegram/events.go @@ -0,0 +1,193 @@ +package telegram + +import ( + "context" + "fmt" + + "github.com/mymmrac/telego" + + "jane/pkg/bus" + "jane/pkg/channels" + "jane/pkg/identity" + "jane/pkg/logger" + "jane/pkg/media" + "jane/pkg/utils" +) + +func (c *TelegramChannel) handleMessage(ctx context.Context, message *telego.Message) error { + if message == nil { + return fmt.Errorf("message is nil") + } + + user := message.From + if user == nil { + return fmt.Errorf("message sender (user) is nil") + } + + platformID := fmt.Sprintf("%d", user.ID) + sender := bus.SenderInfo{ + Platform: "telegram", + PlatformID: platformID, + CanonicalID: identity.BuildCanonicalID("telegram", platformID), + Username: user.Username, + DisplayName: user.FirstName, + } + + // check allowlist to avoid downloading attachments for rejected users + if !c.IsAllowedSender(sender) { + logger.DebugCF("telegram", "Message rejected by allowlist", map[string]any{ + "user_id": platformID, + }) + return nil + } + + chatID := message.Chat.ID + c.chatIDs[platformID] = chatID + + content := "" + mediaPaths := []string{} + + chatIDStr := fmt.Sprintf("%d", chatID) + messageIDStr := fmt.Sprintf("%d", message.MessageID) + scope := channels.BuildMediaScope("telegram", chatIDStr, messageIDStr) + + // Helper to register a local file with the media store + storeMedia := func(localPath, filename string) string { + if store := c.GetMediaStore(); store != nil { + ref, err := store.Store(localPath, media.MediaMeta{ + Filename: filename, + Source: "telegram", + }, scope) + if err == nil { + return ref + } + } + return localPath // fallback: use raw path + } + + if message.Text != "" { + content += message.Text + } + + if message.Caption != "" { + if content != "" { + content += "\n" + } + 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")) + 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]" + } + } + + if content == "" { + content = "[empty message]" + } + + // In group chats, apply unified group trigger filtering + if message.Chat.Type != "private" { + isMentioned := c.isBotMentioned(message) + if isMentioned { + content = c.stripBotMention(content) + } + respond, cleaned := c.ShouldRespondInGroup(isMentioned, content) + if !respond { + return nil + } + content = cleaned + } + + // For forum topics, embed the thread ID as "chatID/threadID" so replies + // 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) + threadID := message.MessageThreadID + if message.Chat.IsForum && threadID != 0 { + compositeChatID = fmt.Sprintf("%d/%d", chatID, threadID) + } + + logger.DebugCF("telegram", "Received message", map[string]any{ + "sender_id": sender.CanonicalID, + "chat_id": compositeChatID, + "thread_id": threadID, + "preview": utils.Truncate(content, 50), + }) + + peerKind := "direct" + peerID := fmt.Sprintf("%d", user.ID) + if message.Chat.Type != "private" { + peerKind = "group" + peerID = compositeChatID + } + + peer := bus.Peer{Kind: peerKind, ID: peerID} + messageID := fmt.Sprintf("%d", message.MessageID) + + metadata := map[string]string{ + "user_id": fmt.Sprintf("%d", user.ID), + "username": user.Username, + "first_name": user.FirstName, + "is_group": fmt.Sprintf("%t", message.Chat.Type != "private"), + } + + // Set parent_peer metadata for per-topic agent binding. + if message.Chat.IsForum && threadID != 0 { + metadata["parent_peer_kind"] = "topic" + metadata["parent_peer_id"] = fmt.Sprintf("%d", threadID) + } + + c.HandleMessage(c.ctx, + peer, + messageID, + platformID, + compositeChatID, + content, + mediaPaths, + metadata, + sender, + ) + return nil +} diff --git a/pkg/channels/telegram/media.go b/pkg/channels/telegram/media.go new file mode 100644 index 000000000..87cec3451 --- /dev/null +++ b/pkg/channels/telegram/media.go @@ -0,0 +1,138 @@ +package telegram + +import ( + "context" + "fmt" + "os" + + "github.com/mymmrac/telego" + tu "github.com/mymmrac/telego/telegoutil" + + "jane/pkg/bus" + "jane/pkg/channels" + "jane/pkg/logger" + "jane/pkg/utils" +) + +// SendMedia implements the channels.MediaSender interface. +func (c *TelegramChannel) SendMedia(ctx context.Context, msg bus.OutboundMediaMessage) error { + if !c.IsRunning() { + return channels.ErrNotRunning + } + + chatID, threadID, err := parseTelegramChatID(msg.ChatID) + if err != nil { + return fmt.Errorf("invalid chat ID %s: %w", msg.ChatID, channels.ErrSendFailed) + } + + store := c.GetMediaStore() + if store == nil { + return fmt.Errorf("no media store available: %w", channels.ErrSendFailed) + } + + for _, part := range msg.Parts { + localPath, err := store.Resolve(part.Ref) + if err != nil { + logger.ErrorCF("telegram", "Failed to resolve media ref", map[string]any{ + "ref": part.Ref, + "error": err.Error(), + }) + continue + } + + file, err := os.Open(localPath) + if err != nil { + logger.ErrorCF("telegram", "Failed to open media file", map[string]any{ + "path": localPath, + "error": err.Error(), + }) + continue + } + + switch part.Type { + case "image": + params := &telego.SendPhotoParams{ + ChatID: tu.ID(chatID), + MessageThreadID: threadID, + Photo: telego.InputFile{File: file}, + Caption: part.Caption, + } + _, err = c.bot.SendPhoto(ctx, params) + case "audio": + params := &telego.SendAudioParams{ + ChatID: tu.ID(chatID), + MessageThreadID: threadID, + Audio: telego.InputFile{File: file}, + Caption: part.Caption, + } + _, 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, + } + _, 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, + } + _, err = c.bot.SendDocument(ctx, params) + } + + file.Close() + + if err != nil { + logger.ErrorCF("telegram", "Failed to send media", map[string]any{ + "type": part.Type, + "error": err.Error(), + }) + return fmt.Errorf("telegram send media: %w", channels.ErrTemporary) + } + } + + return nil +} + +func (c *TelegramChannel) downloadPhoto(ctx context.Context, fileID string) string { + file, err := c.bot.GetFile(ctx, &telego.GetFileParams{FileID: fileID}) + if err != nil { + logger.ErrorCF("telegram", "Failed to get photo file", map[string]any{ + "error": err.Error(), + }) + return "" + } + + return c.downloadFileWithInfo(file, ".jpg") +} + +func (c *TelegramChannel) downloadFileWithInfo(file *telego.File, ext string) string { + if file.FilePath == "" { + return "" + } + + url := c.bot.FileDownloadURL(file.FilePath) + logger.DebugCF("telegram", "File URL", map[string]any{"url": url}) + + // Use FilePath as filename for better identification + filename := file.FilePath + ext + return utils.DownloadFile(url, filename, utils.DownloadOptions{ + LoggerPrefix: "telegram", + }) +} + +func (c *TelegramChannel) downloadFile(ctx context.Context, fileID, ext string) string { + file, err := c.bot.GetFile(ctx, &telego.GetFileParams{FileID: fileID}) + if err != nil { + logger.ErrorCF("telegram", "Failed to get file", map[string]any{ + "error": err.Error(), + }) + return "" + } + + return c.downloadFileWithInfo(file, ext) +} diff --git a/pkg/channels/telegram/messages.go b/pkg/channels/telegram/messages.go new file mode 100644 index 000000000..cc3f3d244 --- /dev/null +++ b/pkg/channels/telegram/messages.go @@ -0,0 +1,138 @@ +package telegram + +import ( + "context" + "fmt" + "strconv" + + "github.com/mymmrac/telego" + tu "github.com/mymmrac/telego/telegoutil" + + "jane/pkg/bus" + "jane/pkg/channels" + "jane/pkg/logger" +) + +func (c *TelegramChannel) Send(ctx context.Context, msg bus.OutboundMessage) error { + if !c.IsRunning() { + return channels.ErrNotRunning + } + + chatID, threadID, err := parseTelegramChatID(msg.ChatID) + if err != nil { + return fmt.Errorf("invalid chat ID %s: %w", msg.ChatID, channels.ErrSendFailed) + } + + if msg.Content == "" { + return nil + } + + // The Manager already splits messages to ≤4000 chars (WithMaxMessageLength), + // so msg.Content is guaranteed to be within that limit. We still need to + // check if HTML expansion pushes it beyond Telegram's 4096-char API limit. + replyToID := msg.ReplyToMessageID + queue := []string{msg.Content} + for len(queue) > 0 { + chunk := queue[0] + queue = queue[1:] + + htmlContent := markdownToTelegramHTML(chunk) + + if len([]rune(htmlContent)) > 4096 { + ratio := float64(len([]rune(chunk))) / float64(len([]rune(htmlContent))) + smallerLen := int(float64(4096) * ratio * 0.95) // 5% safety margin + if smallerLen < 100 { + smallerLen = 100 + } + // Push sub-chunks back to the front of the queue for + // re-validation instead of sending them blindly. + subChunks := channels.SplitMessage(chunk, smallerLen) + queue = append(subChunks, queue...) + continue + } + + if err := c.sendHTMLChunk(ctx, chatID, threadID, htmlContent, chunk, replyToID); err != nil { + return err + } + // Only the first chunk should be a reply; subsequent chunks are normal messages. + replyToID = "" + } + + return nil +} + +// sendHTMLChunk sends a single HTML message, falling back to the original +// markdown as plain text on parse failure so users never see raw HTML tags. +func (c *TelegramChannel) sendHTMLChunk( + ctx context.Context, chatID int64, threadID int, htmlContent, mdFallback string, replyToID string, +) error { + tgMsg := tu.Message(tu.ID(chatID), htmlContent) + tgMsg.ParseMode = telego.ModeHTML + tgMsg.MessageThreadID = threadID + + if replyToID != "" { + if mid, parseErr := strconv.Atoi(replyToID); parseErr == nil { + tgMsg.ReplyParameters = &telego.ReplyParameters{ + MessageID: mid, + } + } + } + + if _, err := c.bot.SendMessage(ctx, tgMsg); err != nil { + logger.ErrorCF("telegram", "HTML parse failed, falling back to plain text", map[string]any{ + "error": err.Error(), + }) + tgMsg.Text = mdFallback + tgMsg.ParseMode = "" + if _, err = c.bot.SendMessage(ctx, tgMsg); err != nil { + return fmt.Errorf("telegram send: %w", channels.ErrTemporary) + } + } + return nil +} + +// EditMessage implements channels.MessageEditor. +func (c *TelegramChannel) EditMessage(ctx context.Context, chatID string, messageID string, content string) error { + cid, _, err := parseTelegramChatID(chatID) + if err != nil { + return err + } + mid, err := strconv.Atoi(messageID) + if err != nil { + return err + } + htmlContent := markdownToTelegramHTML(content) + editMsg := tu.EditMessageText(tu.ID(cid), mid, htmlContent) + editMsg.ParseMode = telego.ModeHTML + _, err = c.bot.EditMessageText(ctx, editMsg) + return err +} + +// SendPlaceholder implements channels.PlaceholderCapable. +// It sends a placeholder message (e.g. "Thinking... 💭") that will later be +// edited to the actual response via EditMessage (channels.MessageEditor). +func (c *TelegramChannel) SendPlaceholder(ctx context.Context, chatID string) (string, error) { + phCfg := c.config.Channels.Telegram.Placeholder + if !phCfg.Enabled { + return "", nil + } + + text := phCfg.Text + if text == "" { + text = "Thinking... 💭" + } + + cid, threadID, err := parseTelegramChatID(chatID) + if err != nil { + return "", err + } + + phMsg := tu.Message(tu.ID(cid), text) + phMsg.MessageThreadID = threadID + pMsg, err := c.bot.SendMessage(ctx, phMsg) + if err != nil { + return "", err + } + + return fmt.Sprintf("%d", pMsg.MessageID), nil +} diff --git a/pkg/channels/telegram/parsing.go b/pkg/channels/telegram/parsing.go new file mode 100644 index 000000000..7b9c4305b --- /dev/null +++ b/pkg/channels/telegram/parsing.go @@ -0,0 +1,125 @@ +package telegram + +import ( + "fmt" + "regexp" + "strings" +) + +var ( + reHeading = regexp.MustCompile(`^#{1,6}\s+(.+)$`) + reBlockquote = regexp.MustCompile(`^>\s*(.*)$`) + reLink = regexp.MustCompile(`\[([^\]]+)\]\(([^)]+)\)`) + reBoldStar = regexp.MustCompile(`\*\*(.+?)\*\*`) + reBoldUnder = regexp.MustCompile(`__(.+?)__`) + reItalic = regexp.MustCompile(`_([^_]+)_`) + reStrike = regexp.MustCompile(`~~(.+?)~~`) + reListItem = regexp.MustCompile(`^[-*]\s+`) + reCodeBlock = regexp.MustCompile("```[\\w]*\\n?([\\s\\S]*?)```") + reInlineCode = regexp.MustCompile("`([^`]+)`") +) + +func markdownToTelegramHTML(text string) string { + if text == "" { + return "" + } + + codeBlocks := extractCodeBlocks(text) + text = codeBlocks.text + + inlineCodes := extractInlineCodes(text) + text = inlineCodes.text + + text = reHeading.ReplaceAllString(text, "$1") + + text = reBlockquote.ReplaceAllString(text, "$1") + + text = escapeHTML(text) + + text = reLink.ReplaceAllString(text, `$1`) + + text = reBoldStar.ReplaceAllString(text, "$1") + + text = reBoldUnder.ReplaceAllString(text, "$1") + + text = reItalic.ReplaceAllStringFunc(text, func(s string) string { + match := reItalic.FindStringSubmatch(s) + if len(match) < 2 { + return s + } + return "" + match[1] + "" + }) + + text = reStrike.ReplaceAllString(text, "$1") + + text = reListItem.ReplaceAllString(text, "• ") + + for i, code := range inlineCodes.codes { + escaped := escapeHTML(code) + text = strings.ReplaceAll(text, fmt.Sprintf("\x00IC%d\x00", i), fmt.Sprintf("%s", escaped)) + } + + for i, code := range codeBlocks.codes { + escaped := escapeHTML(code) + text = strings.ReplaceAll( + text, + fmt.Sprintf("\x00CB%d\x00", i), + fmt.Sprintf("
%s
", escaped), + ) + } + + return text +} + +type codeBlockMatch struct { + text string + codes []string +} + +func extractCodeBlocks(text string) codeBlockMatch { + matches := reCodeBlock.FindAllStringSubmatch(text, -1) + + codes := make([]string, 0, len(matches)) + for _, match := range matches { + codes = append(codes, match[1]) + } + + i := 0 + text = reCodeBlock.ReplaceAllStringFunc(text, func(m string) string { + placeholder := fmt.Sprintf("\x00CB%d\x00", i) + i++ + return placeholder + }) + + return codeBlockMatch{text: text, codes: codes} +} + +type inlineCodeMatch struct { + text string + codes []string +} + +func extractInlineCodes(text string) inlineCodeMatch { + matches := reInlineCode.FindAllStringSubmatch(text, -1) + + codes := make([]string, 0, len(matches)) + for _, match := range matches { + codes = append(codes, match[1]) + } + + i := 0 + text = reInlineCode.ReplaceAllStringFunc(text, func(m string) string { + placeholder := fmt.Sprintf("\x00IC%d\x00", i) + i++ + return placeholder + }) + + return inlineCodeMatch{text: text, codes: codes} +} + +func escapeHTML(text string) string { + text = strings.ReplaceAll(text, "&", "&") + text = strings.ReplaceAll(text, "<", "<") + text = strings.ReplaceAll(text, ">", ">") + return text +} diff --git a/pkg/channels/telegram/telegram.go b/pkg/channels/telegram/telegram.go index 33c501a2b..b56a829c7 100644 --- a/pkg/channels/telegram/telegram.go +++ b/pkg/channels/telegram/telegram.go @@ -6,36 +6,16 @@ import ( "net/http" "net/url" "os" - "regexp" - "strconv" "strings" - "time" "github.com/mymmrac/telego" th "github.com/mymmrac/telego/telegohandler" - tu "github.com/mymmrac/telego/telegoutil" "jane/pkg/bus" "jane/pkg/channels" "jane/pkg/commands" "jane/pkg/config" - "jane/pkg/identity" "jane/pkg/logger" - "jane/pkg/media" - "jane/pkg/utils" -) - -var ( - reHeading = regexp.MustCompile(`^#{1,6}\s+(.+)$`) - reBlockquote = regexp.MustCompile(`^>\s*(.*)$`) - reLink = regexp.MustCompile(`\[([^\]]+)\]\(([^)]+)\)`) - reBoldStar = regexp.MustCompile(`\*\*(.+?)\*\*`) - reBoldUnder = regexp.MustCompile(`__(.+?)__`) - reItalic = regexp.MustCompile(`_([^_]+)_`) - reStrike = regexp.MustCompile(`~~(.+?)~~`) - reListItem = regexp.MustCompile(`^[-*]\s+`) - reCodeBlock = regexp.MustCompile("```[\\w]*\\n?([\\s\\S]*?)```") - reInlineCode = regexp.MustCompile("`([^`]+)`") ) type TelegramChannel struct { @@ -163,676 +143,3 @@ func (c *TelegramChannel) Stop(ctx context.Context) error { return nil } - -func (c *TelegramChannel) Send(ctx context.Context, msg bus.OutboundMessage) error { - if !c.IsRunning() { - return channels.ErrNotRunning - } - - chatID, threadID, err := parseTelegramChatID(msg.ChatID) - if err != nil { - return fmt.Errorf("invalid chat ID %s: %w", msg.ChatID, channels.ErrSendFailed) - } - - if msg.Content == "" { - return nil - } - - // The Manager already splits messages to ≤4000 chars (WithMaxMessageLength), - // so msg.Content is guaranteed to be within that limit. We still need to - // check if HTML expansion pushes it beyond Telegram's 4096-char API limit. - replyToID := msg.ReplyToMessageID - queue := []string{msg.Content} - for len(queue) > 0 { - chunk := queue[0] - queue = queue[1:] - - htmlContent := markdownToTelegramHTML(chunk) - - if len([]rune(htmlContent)) > 4096 { - ratio := float64(len([]rune(chunk))) / float64(len([]rune(htmlContent))) - smallerLen := int(float64(4096) * ratio * 0.95) // 5% safety margin - if smallerLen < 100 { - smallerLen = 100 - } - // Push sub-chunks back to the front of the queue for - // re-validation instead of sending them blindly. - subChunks := channels.SplitMessage(chunk, smallerLen) - queue = append(subChunks, queue...) - continue - } - - if err := c.sendHTMLChunk(ctx, chatID, threadID, htmlContent, chunk, replyToID); err != nil { - return err - } - // Only the first chunk should be a reply; subsequent chunks are normal messages. - replyToID = "" - } - - return nil -} - -// sendHTMLChunk sends a single HTML message, falling back to the original -// markdown as plain text on parse failure so users never see raw HTML tags. -func (c *TelegramChannel) sendHTMLChunk( - ctx context.Context, chatID int64, threadID int, htmlContent, mdFallback string, replyToID string, -) error { - tgMsg := tu.Message(tu.ID(chatID), htmlContent) - tgMsg.ParseMode = telego.ModeHTML - tgMsg.MessageThreadID = threadID - - if replyToID != "" { - if mid, parseErr := strconv.Atoi(replyToID); parseErr == nil { - tgMsg.ReplyParameters = &telego.ReplyParameters{ - MessageID: mid, - } - } - } - - if _, err := c.bot.SendMessage(ctx, tgMsg); err != nil { - logger.ErrorCF("telegram", "HTML parse failed, falling back to plain text", map[string]any{ - "error": err.Error(), - }) - tgMsg.Text = mdFallback - tgMsg.ParseMode = "" - if _, err = c.bot.SendMessage(ctx, tgMsg); err != nil { - return fmt.Errorf("telegram send: %w", channels.ErrTemporary) - } - } - return nil -} - -// StartTyping implements channels.TypingCapable. -// It sends ChatAction(typing) immediately and then repeats every 4 seconds -// (Telegram's typing indicator expires after ~5s) in a background goroutine. -// The returned stop function is idempotent and cancels the goroutine. -func (c *TelegramChannel) StartTyping(ctx context.Context, chatID string) (func(), error) { - cid, threadID, err := parseTelegramChatID(chatID) - if err != nil { - return func() {}, err - } - - action := tu.ChatAction(tu.ID(cid), telego.ChatActionTyping) - action.MessageThreadID = threadID - - // Send the first typing action immediately - _ = c.bot.SendChatAction(ctx, action) - - typingCtx, cancel := context.WithCancel(ctx) - go func() { - ticker := time.NewTicker(4 * time.Second) - defer ticker.Stop() - for { - select { - case <-typingCtx.Done(): - return - case <-ticker.C: - a := tu.ChatAction(tu.ID(cid), telego.ChatActionTyping) - a.MessageThreadID = threadID - _ = c.bot.SendChatAction(typingCtx, a) - } - } - }() - - return cancel, nil -} - -// EditMessage implements channels.MessageEditor. -func (c *TelegramChannel) EditMessage(ctx context.Context, chatID string, messageID string, content string) error { - cid, _, err := parseTelegramChatID(chatID) - if err != nil { - return err - } - mid, err := strconv.Atoi(messageID) - if err != nil { - return err - } - htmlContent := markdownToTelegramHTML(content) - editMsg := tu.EditMessageText(tu.ID(cid), mid, htmlContent) - editMsg.ParseMode = telego.ModeHTML - _, err = c.bot.EditMessageText(ctx, editMsg) - return err -} - -// SendPlaceholder implements channels.PlaceholderCapable. -// It sends a placeholder message (e.g. "Thinking... 💭") that will later be -// edited to the actual response via EditMessage (channels.MessageEditor). -func (c *TelegramChannel) SendPlaceholder(ctx context.Context, chatID string) (string, error) { - phCfg := c.config.Channels.Telegram.Placeholder - if !phCfg.Enabled { - return "", nil - } - - text := phCfg.Text - if text == "" { - text = "Thinking... 💭" - } - - cid, threadID, err := parseTelegramChatID(chatID) - if err != nil { - return "", err - } - - phMsg := tu.Message(tu.ID(cid), text) - phMsg.MessageThreadID = threadID - pMsg, err := c.bot.SendMessage(ctx, phMsg) - if err != nil { - return "", err - } - - return fmt.Sprintf("%d", pMsg.MessageID), nil -} - -// SendMedia implements the channels.MediaSender interface. -func (c *TelegramChannel) SendMedia(ctx context.Context, msg bus.OutboundMediaMessage) error { - if !c.IsRunning() { - return channels.ErrNotRunning - } - - chatID, threadID, err := parseTelegramChatID(msg.ChatID) - if err != nil { - return fmt.Errorf("invalid chat ID %s: %w", msg.ChatID, channels.ErrSendFailed) - } - - store := c.GetMediaStore() - if store == nil { - return fmt.Errorf("no media store available: %w", channels.ErrSendFailed) - } - - for _, part := range msg.Parts { - localPath, err := store.Resolve(part.Ref) - if err != nil { - logger.ErrorCF("telegram", "Failed to resolve media ref", map[string]any{ - "ref": part.Ref, - "error": err.Error(), - }) - continue - } - - file, err := os.Open(localPath) - if err != nil { - logger.ErrorCF("telegram", "Failed to open media file", map[string]any{ - "path": localPath, - "error": err.Error(), - }) - continue - } - - switch part.Type { - case "image": - params := &telego.SendPhotoParams{ - ChatID: tu.ID(chatID), - MessageThreadID: threadID, - Photo: telego.InputFile{File: file}, - Caption: part.Caption, - } - _, err = c.bot.SendPhoto(ctx, params) - case "audio": - params := &telego.SendAudioParams{ - ChatID: tu.ID(chatID), - MessageThreadID: threadID, - Audio: telego.InputFile{File: file}, - Caption: part.Caption, - } - _, 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, - } - _, 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, - } - _, err = c.bot.SendDocument(ctx, params) - } - - file.Close() - - if err != nil { - logger.ErrorCF("telegram", "Failed to send media", map[string]any{ - "type": part.Type, - "error": err.Error(), - }) - return fmt.Errorf("telegram send media: %w", channels.ErrTemporary) - } - } - - return nil -} - -func (c *TelegramChannel) handleMessage(ctx context.Context, message *telego.Message) error { - if message == nil { - return fmt.Errorf("message is nil") - } - - user := message.From - if user == nil { - return fmt.Errorf("message sender (user) is nil") - } - - platformID := fmt.Sprintf("%d", user.ID) - sender := bus.SenderInfo{ - Platform: "telegram", - PlatformID: platformID, - CanonicalID: identity.BuildCanonicalID("telegram", platformID), - Username: user.Username, - DisplayName: user.FirstName, - } - - // check allowlist to avoid downloading attachments for rejected users - if !c.IsAllowedSender(sender) { - logger.DebugCF("telegram", "Message rejected by allowlist", map[string]any{ - "user_id": platformID, - }) - return nil - } - - chatID := message.Chat.ID - c.chatIDs[platformID] = chatID - - content := "" - mediaPaths := []string{} - - chatIDStr := fmt.Sprintf("%d", chatID) - messageIDStr := fmt.Sprintf("%d", message.MessageID) - scope := channels.BuildMediaScope("telegram", chatIDStr, messageIDStr) - - // Helper to register a local file with the media store - storeMedia := func(localPath, filename string) string { - if store := c.GetMediaStore(); store != nil { - ref, err := store.Store(localPath, media.MediaMeta{ - Filename: filename, - Source: "telegram", - }, scope) - if err == nil { - return ref - } - } - return localPath // fallback: use raw path - } - - if message.Text != "" { - content += message.Text - } - - if message.Caption != "" { - if content != "" { - content += "\n" - } - 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")) - 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]" - } - } - - if content == "" { - content = "[empty message]" - } - - // In group chats, apply unified group trigger filtering - if message.Chat.Type != "private" { - isMentioned := c.isBotMentioned(message) - if isMentioned { - content = c.stripBotMention(content) - } - respond, cleaned := c.ShouldRespondInGroup(isMentioned, content) - if !respond { - return nil - } - content = cleaned - } - - // For forum topics, embed the thread ID as "chatID/threadID" so replies - // 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) - threadID := message.MessageThreadID - if message.Chat.IsForum && threadID != 0 { - compositeChatID = fmt.Sprintf("%d/%d", chatID, threadID) - } - - logger.DebugCF("telegram", "Received message", map[string]any{ - "sender_id": sender.CanonicalID, - "chat_id": compositeChatID, - "thread_id": threadID, - "preview": utils.Truncate(content, 50), - }) - - peerKind := "direct" - peerID := fmt.Sprintf("%d", user.ID) - if message.Chat.Type != "private" { - peerKind = "group" - peerID = compositeChatID - } - - peer := bus.Peer{Kind: peerKind, ID: peerID} - messageID := fmt.Sprintf("%d", message.MessageID) - - metadata := map[string]string{ - "user_id": fmt.Sprintf("%d", user.ID), - "username": user.Username, - "first_name": user.FirstName, - "is_group": fmt.Sprintf("%t", message.Chat.Type != "private"), - } - - // Set parent_peer metadata for per-topic agent binding. - if message.Chat.IsForum && threadID != 0 { - metadata["parent_peer_kind"] = "topic" - metadata["parent_peer_id"] = fmt.Sprintf("%d", threadID) - } - - c.HandleMessage(c.ctx, - peer, - messageID, - platformID, - compositeChatID, - content, - mediaPaths, - metadata, - sender, - ) - return nil -} - -func (c *TelegramChannel) downloadPhoto(ctx context.Context, fileID string) string { - file, err := c.bot.GetFile(ctx, &telego.GetFileParams{FileID: fileID}) - if err != nil { - logger.ErrorCF("telegram", "Failed to get photo file", map[string]any{ - "error": err.Error(), - }) - return "" - } - - return c.downloadFileWithInfo(file, ".jpg") -} - -func (c *TelegramChannel) downloadFileWithInfo(file *telego.File, ext string) string { - if file.FilePath == "" { - return "" - } - - url := c.bot.FileDownloadURL(file.FilePath) - logger.DebugCF("telegram", "File URL", map[string]any{"url": url}) - - // Use FilePath as filename for better identification - filename := file.FilePath + ext - return utils.DownloadFile(url, filename, utils.DownloadOptions{ - LoggerPrefix: "telegram", - }) -} - -func (c *TelegramChannel) downloadFile(ctx context.Context, fileID, ext string) string { - file, err := c.bot.GetFile(ctx, &telego.GetFileParams{FileID: fileID}) - if err != nil { - logger.ErrorCF("telegram", "Failed to get file", map[string]any{ - "error": err.Error(), - }) - return "" - } - - return c.downloadFileWithInfo(file, ext) -} - -// parseTelegramChatID splits "chatID/threadID" into its components. -// Returns threadID=0 when no "/" is present (non-forum messages). -func parseTelegramChatID(chatID string) (int64, int, error) { - idx := strings.Index(chatID, "/") - if idx == -1 { - cid, err := strconv.ParseInt(chatID, 10, 64) - return cid, 0, err - } - cid, err := strconv.ParseInt(chatID[:idx], 10, 64) - if err != nil { - return 0, 0, err - } - tid, err := strconv.Atoi(chatID[idx+1:]) - if err != nil { - return 0, 0, fmt.Errorf("invalid thread ID in chat ID %q: %w", chatID, err) - } - return cid, tid, nil -} - -func markdownToTelegramHTML(text string) string { - if text == "" { - return "" - } - - codeBlocks := extractCodeBlocks(text) - text = codeBlocks.text - - inlineCodes := extractInlineCodes(text) - text = inlineCodes.text - - text = reHeading.ReplaceAllString(text, "$1") - - text = reBlockquote.ReplaceAllString(text, "$1") - - text = escapeHTML(text) - - text = reLink.ReplaceAllString(text, `$1`) - - text = reBoldStar.ReplaceAllString(text, "$1") - - text = reBoldUnder.ReplaceAllString(text, "$1") - - text = reItalic.ReplaceAllStringFunc(text, func(s string) string { - match := reItalic.FindStringSubmatch(s) - if len(match) < 2 { - return s - } - return "" + match[1] + "" - }) - - text = reStrike.ReplaceAllString(text, "$1") - - text = reListItem.ReplaceAllString(text, "• ") - - for i, code := range inlineCodes.codes { - escaped := escapeHTML(code) - text = strings.ReplaceAll(text, fmt.Sprintf("\x00IC%d\x00", i), fmt.Sprintf("%s", escaped)) - } - - for i, code := range codeBlocks.codes { - escaped := escapeHTML(code) - text = strings.ReplaceAll( - text, - fmt.Sprintf("\x00CB%d\x00", i), - fmt.Sprintf("
%s
", escaped), - ) - } - - return text -} - -type codeBlockMatch struct { - text string - codes []string -} - -func extractCodeBlocks(text string) codeBlockMatch { - matches := reCodeBlock.FindAllStringSubmatch(text, -1) - - codes := make([]string, 0, len(matches)) - for _, match := range matches { - codes = append(codes, match[1]) - } - - i := 0 - text = reCodeBlock.ReplaceAllStringFunc(text, func(m string) string { - placeholder := fmt.Sprintf("\x00CB%d\x00", i) - i++ - return placeholder - }) - - return codeBlockMatch{text: text, codes: codes} -} - -type inlineCodeMatch struct { - text string - codes []string -} - -func extractInlineCodes(text string) inlineCodeMatch { - matches := reInlineCode.FindAllStringSubmatch(text, -1) - - codes := make([]string, 0, len(matches)) - for _, match := range matches { - codes = append(codes, match[1]) - } - - i := 0 - text = reInlineCode.ReplaceAllStringFunc(text, func(m string) string { - placeholder := fmt.Sprintf("\x00IC%d\x00", i) - i++ - return placeholder - }) - - return inlineCodeMatch{text: text, codes: codes} -} - -func escapeHTML(text string) string { - text = strings.ReplaceAll(text, "&", "&") - text = strings.ReplaceAll(text, "<", "<") - text = strings.ReplaceAll(text, ">", ">") - return text -} - -// isBotMentioned checks if the bot is mentioned in the message via entities. -func (c *TelegramChannel) isBotMentioned(message *telego.Message) bool { - text, entities := telegramEntityTextAndList(message) - if text == "" || len(entities) == 0 { - return false - } - - botUsername := "" - if c.bot != nil { - botUsername = c.bot.Username() - } - runes := []rune(text) - - for _, entity := range entities { - entityText, ok := telegramEntityText(runes, entity) - if !ok { - continue - } - - switch entity.Type { - case telego.EntityTypeMention: - if botUsername != "" && strings.EqualFold(entityText, "@"+botUsername) { - return true - } - case telego.EntityTypeTextMention: - if botUsername != "" && entity.User != nil && strings.EqualFold(entity.User.Username, botUsername) { - return true - } - case telego.EntityTypeBotCommand: - if isBotCommandEntityForThisBot(entityText, botUsername) { - return true - } - } - } - return false -} - -func telegramEntityTextAndList(message *telego.Message) (string, []telego.MessageEntity) { - if message.Text != "" { - return message.Text, message.Entities - } - return message.Caption, message.CaptionEntities -} - -func telegramEntityText(runes []rune, entity telego.MessageEntity) (string, bool) { - if entity.Offset < 0 || entity.Length <= 0 { - return "", false - } - end := entity.Offset + entity.Length - if entity.Offset >= len(runes) || end > len(runes) { - return "", false - } - return string(runes[entity.Offset:end]), true -} - -func isBotCommandEntityForThisBot(entityText, botUsername string) bool { - if !strings.HasPrefix(entityText, "/") { - return false - } - command := strings.TrimPrefix(entityText, "/") - if command == "" { - return false - } - - at := strings.IndexRune(command, '@') - if at == -1 { - // A bare /command delivered to this bot is intended for this bot. - return true - } - - mentionUsername := command[at+1:] - if mentionUsername == "" || botUsername == "" { - return false - } - return strings.EqualFold(mentionUsername, botUsername) -} - -// stripBotMention removes the @bot mention from the content. -func (c *TelegramChannel) stripBotMention(content string) string { - botUsername := c.bot.Username() - if botUsername == "" { - return content - } - // Case-insensitive replacement - re := regexp.MustCompile(`(?i)@` + regexp.QuoteMeta(botUsername)) - content = re.ReplaceAllString(content, "") - return strings.TrimSpace(content) -} diff --git a/pkg/channels/telegram/typing.go b/pkg/channels/telegram/typing.go new file mode 100644 index 000000000..be3e08fe6 --- /dev/null +++ b/pkg/channels/telegram/typing.go @@ -0,0 +1,44 @@ +package telegram + +import ( + "context" + "time" + + "github.com/mymmrac/telego" + tu "github.com/mymmrac/telego/telegoutil" +) + +// StartTyping implements channels.TypingCapable. +// It sends ChatAction(typing) immediately and then repeats every 4 seconds +// (Telegram's typing indicator expires after ~5s) in a background goroutine. +// The returned stop function is idempotent and cancels the goroutine. +func (c *TelegramChannel) StartTyping(ctx context.Context, chatID string) (func(), error) { + cid, threadID, err := parseTelegramChatID(chatID) + if err != nil { + return func() {}, err + } + + action := tu.ChatAction(tu.ID(cid), telego.ChatActionTyping) + action.MessageThreadID = threadID + + // Send the first typing action immediately + _ = c.bot.SendChatAction(ctx, action) + + typingCtx, cancel := context.WithCancel(ctx) + go func() { + ticker := time.NewTicker(4 * time.Second) + defer ticker.Stop() + for { + select { + case <-typingCtx.Done(): + return + case <-ticker.C: + a := tu.ChatAction(tu.ID(cid), telego.ChatActionTyping) + a.MessageThreadID = threadID + _ = c.bot.SendChatAction(typingCtx, a) + } + } + }() + + return cancel, nil +} diff --git a/pkg/channels/telegram/utils.go b/pkg/channels/telegram/utils.go new file mode 100644 index 000000000..66d91e805 --- /dev/null +++ b/pkg/channels/telegram/utils.go @@ -0,0 +1,118 @@ +package telegram + +import ( + "fmt" + "regexp" + "strconv" + "strings" + + "github.com/mymmrac/telego" +) + +// parseTelegramChatID splits "chatID/threadID" into its components. +// Returns threadID=0 when no "/" is present (non-forum messages). +func parseTelegramChatID(chatID string) (int64, int, error) { + idx := strings.Index(chatID, "/") + if idx == -1 { + cid, err := strconv.ParseInt(chatID, 10, 64) + return cid, 0, err + } + cid, err := strconv.ParseInt(chatID[:idx], 10, 64) + if err != nil { + return 0, 0, err + } + tid, err := strconv.Atoi(chatID[idx+1:]) + if err != nil { + return 0, 0, fmt.Errorf("invalid thread ID in chat ID %q: %w", chatID, err) + } + return cid, tid, nil +} + +// isBotMentioned checks if the bot is mentioned in the message via entities. +func (c *TelegramChannel) isBotMentioned(message *telego.Message) bool { + text, entities := telegramEntityTextAndList(message) + if text == "" || len(entities) == 0 { + return false + } + + botUsername := "" + if c.bot != nil { + botUsername = c.bot.Username() + } + runes := []rune(text) + + for _, entity := range entities { + entityText, ok := telegramEntityText(runes, entity) + if !ok { + continue + } + + switch entity.Type { + case telego.EntityTypeMention: + if botUsername != "" && strings.EqualFold(entityText, "@"+botUsername) { + return true + } + case telego.EntityTypeTextMention: + if botUsername != "" && entity.User != nil && strings.EqualFold(entity.User.Username, botUsername) { + return true + } + case telego.EntityTypeBotCommand: + if isBotCommandEntityForThisBot(entityText, botUsername) { + return true + } + } + } + return false +} + +func telegramEntityTextAndList(message *telego.Message) (string, []telego.MessageEntity) { + if message.Text != "" { + return message.Text, message.Entities + } + return message.Caption, message.CaptionEntities +} + +func telegramEntityText(runes []rune, entity telego.MessageEntity) (string, bool) { + if entity.Offset < 0 || entity.Length <= 0 { + return "", false + } + end := entity.Offset + entity.Length + if entity.Offset >= len(runes) || end > len(runes) { + return "", false + } + return string(runes[entity.Offset:end]), true +} + +func isBotCommandEntityForThisBot(entityText, botUsername string) bool { + if !strings.HasPrefix(entityText, "/") { + return false + } + command := strings.TrimPrefix(entityText, "/") + if command == "" { + return false + } + + at := strings.IndexRune(command, '@') + if at == -1 { + // A bare /command delivered to this bot is intended for this bot. + return true + } + + mentionUsername := command[at+1:] + if mentionUsername == "" || botUsername == "" { + return false + } + return strings.EqualFold(mentionUsername, botUsername) +} + +// stripBotMention removes the @bot mention from the content. +func (c *TelegramChannel) stripBotMention(content string) string { + botUsername := c.bot.Username() + if botUsername == "" { + return content + } + // Case-insensitive replacement + re := regexp.MustCompile(`(?i)@` + regexp.QuoteMeta(botUsername)) + content = re.ReplaceAllString(content, "") + return strings.TrimSpace(content) +}