Merge pull request #52 from dj-oyu/fix/telegram-polling-timeout

fix: Telegram long polling HTTP timeout + reconnection indicator

pkg/channels/dedupe.go

@@ -1,8 +1,8 @@
-package wecom
+package channels
 
 import "sync"
 
-const wecomMaxProcessedMessages = 1000
+const defaultMaxProcessedMessages = 1000
 
 // MessageDeduplicator provides thread-safe message deduplication using a circular queue (ring buffer)
 // combined with a hash map. This ensures fast O(1) lookups while naturally evicting the oldest
@@ -18,7 +18,7 @@ type MessageDeduplicator struct {
 // NewMessageDeduplicator creates a new deduplicator with the specified capacity.
 func NewMessageDeduplicator(maxEntries int) *MessageDeduplicator {
 	if maxEntries <= 0 {
-		maxEntries = wecomMaxProcessedMessages
+		maxEntries = defaultMaxProcessedMessages
 	}
 	return &MessageDeduplicator{
 		msgs: make(map[string]bool, maxEntries),

pkg/channels/dedupe_test.go

@@ -1,4 +1,4 @@
-package wecom
+package channels
 
 import (
 	"sync"
@@ -6,7 +6,7 @@ import (
 )
 
 func TestMessageDeduplicator_DuplicateDetection(t *testing.T) {
-	d := NewMessageDeduplicator(wecomMaxProcessedMessages)
+	d := NewMessageDeduplicator(defaultMaxProcessedMessages)
 
 	if ok := d.MarkMessageProcessed("msg-1"); !ok {
 		t.Fatalf("first message should be accepted")
@@ -18,7 +18,7 @@ func TestMessageDeduplicator_DuplicateDetection(t *testing.T) {
 }
 
 func TestMessageDeduplicator_ConcurrentSameMessage(t *testing.T) {
-	d := NewMessageDeduplicator(wecomMaxProcessedMessages)
+	d := NewMessageDeduplicator(defaultMaxProcessedMessages)
 
 	const goroutines = 64
 	var wg sync.WaitGroup

pkg/channels/telegram/telegram.go

@@ -26,15 +26,23 @@ import (
 	"github.com/sipeed/picoclaw/pkg/utils"
 )
 
+// telegramHTTPTimeout is the HTTP client timeout for Telegram API requests.
+// Long polling uses Timeout=30s on the API side; the HTTP client timeout
+// must be longer to avoid canceling valid long-poll responses.
+const telegramHTTPTimeout = 65 * time.Second
+
 type TelegramChannel struct {
 	*channels.BaseChannel
 	bot     *telego.Bot
 	bh      *th.BotHandler
 	config  *config.Config
 	chatIDs map[string]int64
+	dedupe  *channels.MessageDeduplicator
 	ctx     context.Context
 	cancel  context.CancelFunc
 
+	lastActiveChatID string // composite chat ID of last received message
+
 	registerFunc     func(context.Context, []commands.Definition) error
 	commandRegCancel context.CancelFunc
 }
@@ -43,25 +51,47 @@ func NewTelegramChannel(cfg *config.Config, bus *bus.MessageBus) (*TelegramChann
 	var opts []telego.BotOption
 	telegramCfg := cfg.Channels.Telegram
 
+	// Build the base transport (with optional proxy) and wrap it with
+	// resilientTransport for connection-failure detection and recovery logging.
+	var baseTransport http.RoundTripper
 	if telegramCfg.Proxy != "" {
 		proxyURL, parseErr := url.Parse(telegramCfg.Proxy)
 		if parseErr != nil {
 			return nil, fmt.Errorf("invalid proxy URL %q: %w", telegramCfg.Proxy, parseErr)
 		}
-		opts = append(opts, telego.WithHTTPClient(&http.Client{
+		baseTransport = &http.Transport{Proxy: http.ProxyURL(proxyURL)}
-			Transport: &http.Transport{
-				Proxy: http.ProxyURL(proxyURL),
-			},
-		}))
 	} else if os.Getenv("HTTP_PROXY") != "" || os.Getenv("HTTPS_PROXY") != "" {
-		// Use environment proxy if configured
+		baseTransport = &http.Transport{Proxy: http.ProxyFromEnvironment}
-		opts = append(opts, telego.WithHTTPClient(&http.Client{
+	} else {
-			Transport: &http.Transport{
+		baseTransport = http.DefaultTransport
-				Proxy: http.ProxyFromEnvironment,
-			},
-		}))
 	}
 
+	ch := &TelegramChannel{
+		config:  cfg,
+		chatIDs: make(map[string]int64),
+		dedupe:  channels.NewMessageDeduplicator(1000),
+	}
+
+	transport := &resilientTransport{
+		base: baseTransport,
+		onFailure: func() {
+			logger.WarnC("telegram", "Polling connection lost, retrying...")
+		},
+		onRecover: func() {
+			logger.InfoC("telegram", "Polling connection recovered")
+			if chatID := ch.lastActiveChatID; chatID != "" {
+				go func() {
+					_ = ch.sendReconnectNotice(chatID)
+				}()
+			}
+		},
+	}
+
+	opts = append(opts, telego.WithHTTPClient(&http.Client{
+		Transport: transport,
+		Timeout:   telegramHTTPTimeout,
+	}))
+
 	if baseURL := strings.TrimRight(strings.TrimSpace(telegramCfg.BaseURL), "/"); baseURL != "" {
 		opts = append(opts, telego.WithAPIServer(baseURL))
 	}
@@ -82,12 +112,10 @@ func NewTelegramChannel(cfg *config.Config, bus *bus.MessageBus) (*TelegramChann
 		channels.WithReasoningChannelID(telegramCfg.ReasoningChannelID),
 	)
 
-	return &TelegramChannel{
+	ch.BaseChannel = base
-		BaseChannel: base,
+	ch.bot = bot
-		bot:         bot,
+
-		config:      cfg,
+	return ch, nil
-		chatIDs:     make(map[string]int64),
-	}, nil
 }
 
 func (c *TelegramChannel) Start(ctx context.Context) error {
@@ -505,6 +533,15 @@ func (c *TelegramChannel) handleMessage(ctx context.Context, message *telego.Mes
 		return nil
 	}
 
+	// Deduplicate: Telegram may redeliver the same update after a timeout.
+	msgID := fmt.Sprintf("%d", message.MessageID)
+	if !c.dedupe.MarkMessageProcessed(msgID) {
+		logger.DebugCF("telegram", "Skipping duplicate message", map[string]any{
+			"message_id": msgID,
+		})
+		return nil
+	}
+
 	chatID := message.Chat.ID
 	c.chatIDs[platformID] = chatID
 
@@ -617,6 +654,8 @@ func (c *TelegramChannel) handleMessage(ctx context.Context, message *telego.Mes
 		compositeChatID = fmt.Sprintf("%d/%d", chatID, threadID)
 	}
 
+	c.lastActiveChatID = compositeChatID
+
 	logger.DebugCF("telegram", "Received message", map[string]any{
 		"sender_id": sender.CanonicalID,
 		"chat_id":   compositeChatID,
@@ -818,6 +857,19 @@ func isBotCommandEntityForThisBot(entityText, botUsername string) bool {
 	return strings.EqualFold(mentionUsername, botUsername)
 }
 
+// sendReconnectNotice sends a short notification to the given chat after
+// the polling connection recovers from a failure.
+func (c *TelegramChannel) sendReconnectNotice(compositeChatID string) error {
+	cid, threadID, err := parseTelegramChatID(compositeChatID)
+	if err != nil {
+		return err
+	}
+	msg := tu.Message(tu.ID(cid), "[system] Connection recovered — messages during the outage may have been delayed.")
+	msg.MessageThreadID = threadID
+	_, err = c.bot.SendMessage(c.ctx, msg)
+	return err
+}
+
 // stripBotMention removes the @bot mention from the content.
 func (c *TelegramChannel) stripBotMention(content string) string {
 	botUsername := c.bot.Username()

pkg/channels/telegram/telegram_dispatch_test.go

@@ -15,6 +15,7 @@ func TestHandleMessage_DoesNotConsumeGenericCommandsLocally(t *testing.T) {
 	ch := &TelegramChannel{
 		BaseChannel: channels.NewBaseChannel("telegram", nil, messageBus, nil),
 		chatIDs:     make(map[string]int64),
+		dedupe:      channels.NewMessageDeduplicator(1000),
 		ctx:         context.Background(),
 	}
 

pkg/channels/telegram/telegram_group_command_filter_test.go

@@ -51,6 +51,7 @@ func newGroupMentionOnlyChannel(t *testing.T, botUsername string) (*TelegramChan
 		),
 		bot:     newTestTelegramBot(t, botUsername),
 		chatIDs: make(map[string]int64),
+		dedupe:  channels.NewMessageDeduplicator(1000),
 		ctx:     context.Background(),
 	}
 	return ch, messageBus

pkg/channels/telegram/telegram_test.go

@@ -524,6 +524,7 @@ func TestHandleMessage_ForumTopic_SetsMetadata(t *testing.T) {
 	ch := &TelegramChannel{
 		BaseChannel: channels.NewBaseChannel("telegram", nil, messageBus, nil),
 		chatIDs:     make(map[string]int64),
+		dedupe:      channels.NewMessageDeduplicator(1000),
 		ctx:         context.Background(),
 	}
 
@@ -565,6 +566,7 @@ func TestHandleMessage_NoForum_NoThreadMetadata(t *testing.T) {
 	ch := &TelegramChannel{
 		BaseChannel: channels.NewBaseChannel("telegram", nil, messageBus, nil),
 		chatIDs:     make(map[string]int64),
+		dedupe:      channels.NewMessageDeduplicator(1000),
 		ctx:         context.Background(),
 	}
 
@@ -604,6 +606,7 @@ func TestHandleMessage_ReplyThread_NonForum_NoIsolation(t *testing.T) {
 	ch := &TelegramChannel{
 		BaseChannel: channels.NewBaseChannel("telegram", nil, messageBus, nil),
 		chatIDs:     make(map[string]int64),
+		dedupe:      channels.NewMessageDeduplicator(1000),
 		ctx:         context.Background(),
 	}
 

pkg/channels/telegram/transport.go

@@ -0,0 +1,27 @@
+package telegram
+
+import (
+	"net/http"
+	"sync/atomic"
+)
+
+// resilientTransport wraps http.RoundTripper to detect polling failures
+// and notify on recovery.
+type resilientTransport struct {
+	base      http.RoundTripper
+	onFailure func() // called once when first failure detected
+	onRecover func() // called once when connection recovers after failure
+	failed    atomic.Bool
+}
+
+func (t *resilientTransport) RoundTrip(req *http.Request) (*http.Response, error) {
+	resp, err := t.base.RoundTrip(req)
+	if err != nil {
+		if !t.failed.Swap(true) && t.onFailure != nil {
+			t.onFailure()
+		}
+	} else if t.failed.Swap(false) && t.onRecover != nil {
+		t.onRecover()
+	}
+	return resp, err
+}

pkg/channels/wecom/app.go

@@ -38,7 +38,7 @@ type WeComAppChannel struct {
 	tokenMu       sync.RWMutex
 	ctx           context.Context
 	cancel        context.CancelFunc
-	processedMsgs *MessageDeduplicator
+	processedMsgs *channels.MessageDeduplicator
 }
 
 // WeComXMLMessage represents the XML message structure from WeCom
@@ -143,7 +143,7 @@ func NewWeComAppChannel(cfg config.WeComAppConfig, messageBus *bus.MessageBus) (
 		client:        &http.Client{Timeout: clientTimeout},
 		ctx:           ctx,
 		cancel:        cancel,
-		processedMsgs: NewMessageDeduplicator(wecomMaxProcessedMessages),
+		processedMsgs: channels.NewMessageDeduplicator(1000),
 	}, nil
 }
 

pkg/channels/wecom/bot.go

@@ -27,7 +27,7 @@ type WeComBotChannel struct {
 	client        *http.Client
 	ctx           context.Context
 	cancel        context.CancelFunc
-	processedMsgs *MessageDeduplicator
+	processedMsgs *channels.MessageDeduplicator
 }
 
 // WeComBotMessage represents the JSON message structure from WeCom Bot (AIBOT)
@@ -106,7 +106,7 @@ func NewWeComBotChannel(cfg config.WeComConfig, messageBus *bus.MessageBus) (*We
 		client:        &http.Client{Timeout: clientTimeout},
 		ctx:           ctx,
 		cancel:        cancel,
-		processedMsgs: NewMessageDeduplicator(wecomMaxProcessedMessages),
+		processedMsgs: channels.NewMessageDeduplicator(1000),
 	}, nil
 }