From 4d5d1d05095cc923a1645e8c125f8fe31700ff6d Mon Sep 17 00:00:00 2001 From: Dmitrii Balabanov Date: Mon, 23 Mar 2026 11:07:42 +0200 Subject: [PATCH] refactor(channels): Send returns ([]string, error), remove MessageIDsSender Replace the optional MessageIDsSender interface with a direct return value on Send. Channels that support delivery IDs (Telegram, Discord, Slack, QQ) return them from Send; all others return nil. Manager.sendWithRetry drops the type-assertion branch and calls Send uniformly. --- pkg/agent/loop_test.go | 4 +- pkg/channels/README.md | 48 +++++++++++-------- pkg/channels/README.zh.md | 48 +++++++++++-------- pkg/channels/base.go | 2 +- pkg/channels/dingtalk/dingtalk.go | 10 ++-- pkg/channels/discord/discord.go | 8 +--- pkg/channels/feishu/feishu_32.go | 4 +- pkg/channels/feishu/feishu_64.go | 16 +++---- pkg/channels/interfaces.go | 6 --- pkg/channels/irc/irc.go | 10 ++-- pkg/channels/line/line.go | 8 ++-- pkg/channels/maixcam/maixcam.go | 12 ++--- pkg/channels/manager.go | 11 ++--- pkg/channels/manager_test.go | 19 +++----- pkg/channels/matrix/matrix.go | 12 ++--- pkg/channels/onebot/onebot.go | 16 +++---- pkg/channels/pico/client.go | 8 ++-- pkg/channels/pico/client_test.go | 8 ++-- pkg/channels/pico/pico.go | 6 +-- pkg/channels/qq/qq.go | 8 +--- pkg/channels/slack/slack.go | 8 +--- pkg/channels/telegram/telegram.go | 8 +--- pkg/channels/telegram/telegram_test.go | 24 +++++----- pkg/channels/wecom/wecom.go | 16 +++---- pkg/channels/weixin/weixin.go | 16 +++---- pkg/channels/whatsapp/whatsapp.go | 14 +++--- .../whatsapp_native/whatsapp_native.go | 16 +++---- 27 files changed, 170 insertions(+), 196 deletions(-) diff --git a/pkg/agent/loop_test.go b/pkg/agent/loop_test.go index b61527993..cfa5cf0f2 100644 --- a/pkg/agent/loop_test.go +++ b/pkg/agent/loop_test.go @@ -28,7 +28,9 @@ type fakeChannel struct{ id string } func (f *fakeChannel) Name() string { return "fake" } func (f *fakeChannel) Start(ctx context.Context) error { return nil } func (f *fakeChannel) Stop(ctx context.Context) error { return nil } -func (f *fakeChannel) Send(ctx context.Context, msg bus.OutboundMessage) error { return nil } +func (f *fakeChannel) Send(ctx context.Context, msg bus.OutboundMessage) ([]string, error) { + return nil, nil +} func (f *fakeChannel) IsRunning() bool { return true } func (f *fakeChannel) IsAllowed(string) bool { return true } func (f *fakeChannel) IsAllowedSender(sender bus.SenderInfo) bool { return true } diff --git a/pkg/channels/README.md b/pkg/channels/README.md index e16d364c3..887e0ae0f 100644 --- a/pkg/channels/README.md +++ b/pkg/channels/README.md @@ -253,27 +253,27 @@ func (c *TelegramChannel) Stop(ctx context.Context) error { ```go // Old code: returns plain error -func (c *TelegramChannel) Send(ctx context.Context, msg bus.OutboundMessage) error { - if !c.running { return fmt.Errorf("not running") } +func (c *TelegramChannel) Send(ctx context.Context, msg bus.OutboundMessage) ([]string, error) { + if !c.running { return nil, fmt.Errorf("not running") } // ... - if err != nil { return err } + if err != nil { return nil, err } } // New code: must return sentinel errors for Manager to determine retry strategy -func (c *TelegramChannel) Send(ctx context.Context, msg bus.OutboundMessage) error { +func (c *TelegramChannel) Send(ctx context.Context, msg bus.OutboundMessage) ([]string, error) { if !c.IsRunning() { - return channels.ErrNotRunning // ← Manager will not retry + return nil, channels.ErrNotRunning // ← Manager will not retry } // ... if err != nil { // Use ClassifySendError to wrap error based on HTTP status code - return channels.ClassifySendError(statusCode, err) + return nil, channels.ClassifySendError(statusCode, err) // Or manually wrap: - // return fmt.Errorf("%w: %v", channels.ErrTemporary, err) - // return fmt.Errorf("%w: %v", channels.ErrRateLimit, err) - // return fmt.Errorf("%w: %v", channels.ErrSendFailed, err) + // return nil, fmt.Errorf("%w: %v", channels.ErrTemporary, err) + // return nil, fmt.Errorf("%w: %v", channels.ErrRateLimit, err) + // return nil, fmt.Errorf("%w: %v", channels.ErrSendFailed, err) } - return nil + return nil, nil } ``` @@ -301,6 +301,8 @@ sender := bus.SenderInfo{ CanonicalID: identity.BuildCanonicalID("telegram", strconv.FormatInt(from.ID, 10)), Username: from.Username, DisplayName: from.FirstName, + FirstName: from.FirstName, + LastName: from.LastName, } peer := bus.Peer{ @@ -502,10 +504,10 @@ func (c *MatrixChannel) Stop(ctx context.Context) error { return nil } -func (c *MatrixChannel) Send(ctx context.Context, msg bus.OutboundMessage) error { +func (c *MatrixChannel) Send(ctx context.Context, msg bus.OutboundMessage) ([]string, error) { // 1. Check running state if !c.IsRunning() { - return channels.ErrNotRunning + return nil, channels.ErrNotRunning } // 2. Send message to Matrix @@ -513,14 +515,14 @@ func (c *MatrixChannel) Send(ctx context.Context, msg bus.OutboundMessage) error if err != nil { // 3. Must use error classification wrapping // If you have an HTTP status code: - // return channels.ClassifySendError(statusCode, err) + // return nil, channels.ClassifySendError(statusCode, err) // If it's a network error: - // return channels.ClassifyNetError(err) + // return nil, channels.ClassifyNetError(err) // If manual classification is needed: - return fmt.Errorf("%w: %v", channels.ErrTemporary, err) + return nil, fmt.Errorf("%w: %v", channels.ErrTemporary, err) } - return nil + return nil, nil } // ========== Incoming Message Handling ========== @@ -868,7 +870,9 @@ type SenderInfo struct { PlatformID string `json:"platform_id,omitempty"` // Platform-native ID CanonicalID string `json:"canonical_id,omitempty"` // "platform:id" canonical format Username string `json:"username,omitempty"` - DisplayName string `json:"display_name,omitempty"` + DisplayName string `json:"display_name,omitempty"` // fallback display name when first/last not available + FirstName string `json:"first_name,omitempty"` // given name (preferred over DisplayName when set) + LastName string `json:"last_name,omitempty"` // family name } // Inbound message @@ -889,9 +893,11 @@ type InboundMessage struct { // Outbound text message type OutboundMessage struct { - Channel string - ChatID string - Content string + Channel string // Target channel name + ChatID string // Target chat/room ID + Content string // Message text + ReplyToMessageID string // Optional: reply to this platform message ID + OnDelivered func(msgIDs []string) // Optional: called with delivered platform message IDs (not serialized) } // Outbound media message @@ -1272,7 +1278,7 @@ type Channel interface { Name() string Start(ctx context.Context) error Stop(ctx context.Context) error - Send(ctx context.Context, msg bus.OutboundMessage) error + Send(ctx context.Context, msg bus.OutboundMessage) ([]string, error) IsRunning() bool IsAllowed(senderID string) bool IsAllowedSender(sender bus.SenderInfo) bool diff --git a/pkg/channels/README.zh.md b/pkg/channels/README.zh.md index 7e122ae1b..7af66f08f 100644 --- a/pkg/channels/README.zh.md +++ b/pkg/channels/README.zh.md @@ -253,27 +253,27 @@ func (c *TelegramChannel) Stop(ctx context.Context) error { ```go // 旧代码：返回普通 error -func (c *TelegramChannel) Send(ctx context.Context, msg bus.OutboundMessage) error { - if !c.running { return fmt.Errorf("not running") } +func (c *TelegramChannel) Send(ctx context.Context, msg bus.OutboundMessage) ([]string, error) { + if !c.running { return nil, fmt.Errorf("not running") } // ... - if err != nil { return err } + if err != nil { return nil, err } } // 新代码：必须返回哨兵错误，供 Manager 判断重试策略 -func (c *TelegramChannel) Send(ctx context.Context, msg bus.OutboundMessage) error { +func (c *TelegramChannel) Send(ctx context.Context, msg bus.OutboundMessage) ([]string, error) { if !c.IsRunning() { - return channels.ErrNotRunning // ← Manager 不会重试 + return nil, channels.ErrNotRunning // ← Manager 不会重试 } // ... if err != nil { // 使用 ClassifySendError 根据 HTTP 状态码包装错误 - return channels.ClassifySendError(statusCode, err) + return nil, channels.ClassifySendError(statusCode, err) // 或手动包装： - // return fmt.Errorf("%w: %v", channels.ErrTemporary, err) - // return fmt.Errorf("%w: %v", channels.ErrRateLimit, err) - // return fmt.Errorf("%w: %v", channels.ErrSendFailed, err) + // return nil, fmt.Errorf("%w: %v", channels.ErrTemporary, err) + // return nil, fmt.Errorf("%w: %v", channels.ErrRateLimit, err) + // return nil, fmt.Errorf("%w: %v", channels.ErrSendFailed, err) } - return nil + return nil, nil } ``` @@ -301,6 +301,8 @@ sender := bus.SenderInfo{ CanonicalID: identity.BuildCanonicalID("telegram", strconv.FormatInt(from.ID, 10)), Username: from.Username, DisplayName: from.FirstName, + FirstName: from.FirstName, + LastName: from.LastName, } peer := bus.Peer{ @@ -502,10 +504,10 @@ func (c *MatrixChannel) Stop(ctx context.Context) error { return nil } -func (c *MatrixChannel) Send(ctx context.Context, msg bus.OutboundMessage) error { +func (c *MatrixChannel) Send(ctx context.Context, msg bus.OutboundMessage) ([]string, error) { // 1. 检查运行状态 if !c.IsRunning() { - return channels.ErrNotRunning + return nil, channels.ErrNotRunning } // 2. 发送消息到 Matrix @@ -513,14 +515,14 @@ func (c *MatrixChannel) Send(ctx context.Context, msg bus.OutboundMessage) error if err != nil { // 3. 必须使用错误分类包装 // 如果你有 HTTP 状态码： - // return channels.ClassifySendError(statusCode, err) + // return nil, channels.ClassifySendError(statusCode, err) // 如果是网络错误： - // return channels.ClassifyNetError(err) + // return nil, channels.ClassifyNetError(err) // 如果需要手动分类： - return fmt.Errorf("%w: %v", channels.ErrTemporary, err) + return nil, fmt.Errorf("%w: %v", channels.ErrTemporary, err) } - return nil + return nil, nil } // ========== 消息接收处理 ========== @@ -867,7 +869,9 @@ type SenderInfo struct { PlatformID string `json:"platform_id,omitempty"` // 平台原始 ID CanonicalID string `json:"canonical_id,omitempty"` // "platform:id" 规范格式 Username string `json:"username,omitempty"` - DisplayName string `json:"display_name,omitempty"` + DisplayName string `json:"display_name,omitempty"` // 无 first/last 时的回退显示名 + FirstName string `json:"first_name,omitempty"` // 名（优先于 DisplayName） + LastName string `json:"last_name,omitempty"` // 姓 } // 入站消息 @@ -888,9 +892,11 @@ type InboundMessage struct { // 出站文本消息 type OutboundMessage struct { - Channel string - ChatID string - Content string + Channel string // 目标 channel 名称 + ChatID string // 目标聊天/房间 ID + Content string // 消息文本 + ReplyToMessageID string // 可选：回复的平台消息 ID + OnDelivered func(msgIDs []string) // 可选：投递成功后携带平台消息 ID 回调（不序列化） } // 出站媒体消息 @@ -1271,7 +1277,7 @@ type Channel interface { Name() string Start(ctx context.Context) error Stop(ctx context.Context) error - Send(ctx context.Context, msg bus.OutboundMessage) error + Send(ctx context.Context, msg bus.OutboundMessage) ([]string, error) IsRunning() bool IsAllowed(senderID string) bool IsAllowedSender(sender bus.SenderInfo) bool diff --git a/pkg/channels/base.go b/pkg/channels/base.go index f50e2abef..82b7fa92e 100644 --- a/pkg/channels/base.go +++ b/pkg/channels/base.go @@ -48,7 +48,7 @@ type Channel interface { Name() string Start(ctx context.Context) error Stop(ctx context.Context) error - Send(ctx context.Context, msg bus.OutboundMessage) error + Send(ctx context.Context, msg bus.OutboundMessage) ([]string, error) IsRunning() bool IsAllowed(senderID string) bool IsAllowedSender(sender bus.SenderInfo) bool diff --git a/pkg/channels/dingtalk/dingtalk.go b/pkg/channels/dingtalk/dingtalk.go index 7ac2c073f..023e07e97 100644 --- a/pkg/channels/dingtalk/dingtalk.go +++ b/pkg/channels/dingtalk/dingtalk.go @@ -103,20 +103,20 @@ func (c *DingTalkChannel) Stop(ctx context.Context) error { } // Send sends a message to DingTalk via the chatbot reply API -func (c *DingTalkChannel) Send(ctx context.Context, msg bus.OutboundMessage) error { +func (c *DingTalkChannel) Send(ctx context.Context, msg bus.OutboundMessage) ([]string, error) { if !c.IsRunning() { - return channels.ErrNotRunning + return nil, channels.ErrNotRunning } // Get session webhook from storage sessionWebhookRaw, ok := c.sessionWebhooks.Load(msg.ChatID) if !ok { - return fmt.Errorf("no session_webhook found for chat %s, cannot send message", msg.ChatID) + return nil, fmt.Errorf("no session_webhook found for chat %s, cannot send message", msg.ChatID) } sessionWebhook, ok := sessionWebhookRaw.(string) if !ok { - return fmt.Errorf("invalid session_webhook type for chat %s", msg.ChatID) + return nil, fmt.Errorf("invalid session_webhook type for chat %s", msg.ChatID) } logger.DebugCF("dingtalk", "Sending message", map[string]any{ @@ -125,7 +125,7 @@ func (c *DingTalkChannel) Send(ctx context.Context, msg bus.OutboundMessage) err }) // Use the session webhook to send the reply - return c.SendDirectReply(ctx, sessionWebhook, msg.Content) + return nil, c.SendDirectReply(ctx, sessionWebhook, msg.Content) } // onChatBotMessageReceived implements the IChatBotMessageHandler function signature diff --git a/pkg/channels/discord/discord.go b/pkg/channels/discord/discord.go index 6e8926d83..ba2f77179 100644 --- a/pkg/channels/discord/discord.go +++ b/pkg/channels/discord/discord.go @@ -128,13 +128,7 @@ func (c *DiscordChannel) Stop(ctx context.Context) error { return nil } -func (c *DiscordChannel) Send(ctx context.Context, msg bus.OutboundMessage) error { - _, err := c.SendMessageWithIDs(ctx, msg) - return err -} - -// SendMessageWithIDs implements channels.MessageIDsSender. -func (c *DiscordChannel) SendMessageWithIDs(ctx context.Context, msg bus.OutboundMessage) ([]string, error) { +func (c *DiscordChannel) Send(ctx context.Context, msg bus.OutboundMessage) ([]string, error) { if !c.IsRunning() { return nil, channels.ErrNotRunning } diff --git a/pkg/channels/feishu/feishu_32.go b/pkg/channels/feishu/feishu_32.go index f5e3aa224..7bb006bdd 100644 --- a/pkg/channels/feishu/feishu_32.go +++ b/pkg/channels/feishu/feishu_32.go @@ -36,8 +36,8 @@ func (c *FeishuChannel) Stop(ctx context.Context) error { } // Send is a stub method to satisfy the Channel interface -func (c *FeishuChannel) Send(ctx context.Context, msg bus.OutboundMessage) error { - return errUnsupported +func (c *FeishuChannel) Send(ctx context.Context, msg bus.OutboundMessage) ([]string, error) { + return nil, errUnsupported } // EditMessage is a stub method to satisfy MessageEditor diff --git a/pkg/channels/feishu/feishu_64.go b/pkg/channels/feishu/feishu_64.go index 76df988ad..714b16d60 100644 --- a/pkg/channels/feishu/feishu_64.go +++ b/pkg/channels/feishu/feishu_64.go @@ -131,26 +131,26 @@ func (c *FeishuChannel) Stop(ctx context.Context) error { // Send sends a message using Interactive Card format for markdown rendering. // Falls back to plain text message if card sending fails (e.g., table limit exceeded). -func (c *FeishuChannel) Send(ctx context.Context, msg bus.OutboundMessage) error { +func (c *FeishuChannel) Send(ctx context.Context, msg bus.OutboundMessage) ([]string, error) { if !c.IsRunning() { - return channels.ErrNotRunning + return nil, channels.ErrNotRunning } if msg.ChatID == "" { - return fmt.Errorf("chat ID is empty: %w", channels.ErrSendFailed) + return nil, fmt.Errorf("chat ID is empty: %w", channels.ErrSendFailed) } // Build interactive card with markdown content cardContent, err := buildMarkdownCard(msg.Content) if err != nil { // If card build fails, fall back to plain text - return c.sendText(ctx, msg.ChatID, msg.Content) + return nil, c.sendText(ctx, msg.ChatID, msg.Content) } // First attempt: try sending as interactive card err = c.sendCard(ctx, msg.ChatID, cardContent) if err == nil { - return nil + return nil, nil } // Check if error is due to card table limit (error code 11310) @@ -167,14 +167,14 @@ func (c *FeishuChannel) Send(ctx context.Context, msg bus.OutboundMessage) error // Second attempt: fall back to plain text message textErr := c.sendText(ctx, msg.ChatID, msg.Content) if textErr == nil { - return nil + return nil, nil } // If text also fails, return the text error - return textErr + return nil, textErr } // For other errors, return the original card error - return err + return nil, err } // EditMessage implements channels.MessageEditor. diff --git a/pkg/channels/interfaces.go b/pkg/channels/interfaces.go index e4388664a..0cfd435b0 100644 --- a/pkg/channels/interfaces.go +++ b/pkg/channels/interfaces.go @@ -62,12 +62,6 @@ type PlaceholderRecorder interface { RecordReactionUndo(channel, chatID string, undo func()) } -// MessageIDsSender is implemented by channels that can return the platform -// message IDs for a delivered outbound text message. -type MessageIDsSender interface { - SendMessageWithIDs(ctx context.Context, msg bus.OutboundMessage) (messageIDs []string, err error) -} - // CommandRegistrarCapable is implemented by channels that can register // command menus with their upstream platform (e.g. Telegram BotCommand). // Channels that do not support platform-level command menus can ignore it. diff --git a/pkg/channels/irc/irc.go b/pkg/channels/irc/irc.go index 289ce2c9b..0b0e8e0c8 100644 --- a/pkg/channels/irc/irc.go +++ b/pkg/channels/irc/irc.go @@ -130,18 +130,18 @@ func (c *IRCChannel) Stop(ctx context.Context) error { } // Send sends a message to an IRC channel or user. -func (c *IRCChannel) Send(ctx context.Context, msg bus.OutboundMessage) error { +func (c *IRCChannel) Send(ctx context.Context, msg bus.OutboundMessage) ([]string, error) { if !c.IsRunning() { - return channels.ErrNotRunning + return nil, channels.ErrNotRunning } target := msg.ChatID if target == "" { - return fmt.Errorf("chat ID is empty: %w", channels.ErrSendFailed) + return nil, fmt.Errorf("chat ID is empty: %w", channels.ErrSendFailed) } if strings.TrimSpace(msg.Content) == "" { - return nil + return nil, nil } // Send each line separately (IRC is line-oriented) @@ -158,7 +158,7 @@ func (c *IRCChannel) Send(ctx context.Context, msg bus.OutboundMessage) error { "target": target, "lines": len(lines), }) - return nil + return nil, nil } // StartTyping implements channels.TypingCapable using IRCv3 +typing client tag. diff --git a/pkg/channels/line/line.go b/pkg/channels/line/line.go index 4eaadae70..8fd21f0c0 100644 --- a/pkg/channels/line/line.go +++ b/pkg/channels/line/line.go @@ -496,9 +496,9 @@ func (c *LINEChannel) resolveChatID(source lineSource) string { // Send sends a message to LINE. It first tries the Reply API (free) // using a cached reply token, then falls back to the Push API. -func (c *LINEChannel) Send(ctx context.Context, msg bus.OutboundMessage) error { +func (c *LINEChannel) Send(ctx context.Context, msg bus.OutboundMessage) ([]string, error) { if !c.IsRunning() { - return channels.ErrNotRunning + return nil, channels.ErrNotRunning } // Load and consume quote token for this chat @@ -516,14 +516,14 @@ func (c *LINEChannel) Send(ctx context.Context, msg bus.OutboundMessage) error { "chat_id": msg.ChatID, "quoted": quoteToken != "", }) - return nil + return nil, nil } logger.DebugC("line", "Reply API failed, falling back to Push API") } } // Fall back to Push API - return c.sendPush(ctx, msg.ChatID, msg.Content, quoteToken) + return nil, c.sendPush(ctx, msg.ChatID, msg.Content, quoteToken) } // SendMedia implements the channels.MediaSender interface. diff --git a/pkg/channels/maixcam/maixcam.go b/pkg/channels/maixcam/maixcam.go index ff9a3ed1a..bbbf2da56 100644 --- a/pkg/channels/maixcam/maixcam.go +++ b/pkg/channels/maixcam/maixcam.go @@ -240,15 +240,15 @@ func (c *MaixCamChannel) Stop(ctx context.Context) error { return nil } -func (c *MaixCamChannel) Send(ctx context.Context, msg bus.OutboundMessage) error { +func (c *MaixCamChannel) Send(ctx context.Context, msg bus.OutboundMessage) ([]string, error) { if !c.IsRunning() { - return channels.ErrNotRunning + return nil, channels.ErrNotRunning } // Check ctx before entering write path select { case <-ctx.Done(): - return ctx.Err() + return nil, ctx.Err() default: } @@ -257,7 +257,7 @@ func (c *MaixCamChannel) Send(ctx context.Context, msg bus.OutboundMessage) erro if len(c.clients) == 0 { logger.WarnC("maixcam", "No MaixCam devices connected") - return fmt.Errorf("no connected MaixCam devices") + return nil, fmt.Errorf("no connected MaixCam devices") } response := map[string]any{ @@ -269,7 +269,7 @@ func (c *MaixCamChannel) Send(ctx context.Context, msg bus.OutboundMessage) erro data, err := json.Marshal(response) if err != nil { - return fmt.Errorf("failed to marshal response: %w", err) + return nil, fmt.Errorf("failed to marshal response: %w", err) } var sendErr error @@ -285,5 +285,5 @@ func (c *MaixCamChannel) Send(ctx context.Context, msg bus.OutboundMessage) erro _ = conn.SetWriteDeadline(time.Time{}) } - return sendErr + return nil, sendErr } diff --git a/pkg/channels/manager.go b/pkg/channels/manager.go index 8b2cc5cbd..3a768d7ca 100644 --- a/pkg/channels/manager.go +++ b/pkg/channels/manager.go @@ -705,14 +705,8 @@ func (m *Manager) sendWithRetry( var lastErr error var msgIDs []string - sender, hasMessageIDsSender := w.ch.(MessageIDsSender) for attempt := 0; attempt <= maxRetries; attempt++ { - msgIDs = nil - if hasMessageIDsSender { - msgIDs, lastErr = sender.SendMessageWithIDs(ctx, msg) - } else { - lastErr = w.ch.Send(ctx, msg) - } + msgIDs, lastErr = w.ch.Send(ctx, msg) if lastErr == nil { return msgIDs, true } @@ -1161,5 +1155,6 @@ func (m *Manager) SendToChannel(ctx context.Context, channelName, chatID, conten // Fallback: direct send (should not happen) channel, _ := m.channels[channelName] - return channel.Send(ctx, msg) + _, err := channel.Send(ctx, msg) + return err } diff --git a/pkg/channels/manager_test.go b/pkg/channels/manager_test.go index 5b3ee3bf8..98cee5821 100644 --- a/pkg/channels/manager_test.go +++ b/pkg/channels/manager_test.go @@ -26,20 +26,15 @@ type mockChannel struct { lastPlaceholderID string } -func (m *mockChannel) Send(ctx context.Context, msg bus.OutboundMessage) error { +func (m *mockChannel) Send(ctx context.Context, msg bus.OutboundMessage) ([]string, error) { m.sentMessages = append(m.sentMessages, msg) - return m.sendFn(ctx, msg) -} - -func (m *mockChannel) SendMessageWithIDs(ctx context.Context, msg bus.OutboundMessage) ([]string, error) { - m.sentMessages = append(m.sentMessages, msg) - if m.sendWithIDsFn == nil { - if m.sendFn == nil { - return nil, nil - } - return nil, m.sendFn(ctx, msg) + if m.sendWithIDsFn != nil { + return m.sendWithIDsFn(ctx, msg) } - return m.sendWithIDsFn(ctx, msg) + if m.sendFn == nil { + return nil, nil + } + return nil, m.sendFn(ctx, msg) } func (m *mockChannel) Start(ctx context.Context) error { return nil } diff --git a/pkg/channels/matrix/matrix.go b/pkg/channels/matrix/matrix.go index a5e768dc5..cd854d207 100644 --- a/pkg/channels/matrix/matrix.go +++ b/pkg/channels/matrix/matrix.go @@ -379,26 +379,26 @@ func markdownToHTML(md string) string { return strings.TrimSpace(string(markdown.ToHTML([]byte(md), p, renderer))) } -func (c *MatrixChannel) Send(ctx context.Context, msg bus.OutboundMessage) error { +func (c *MatrixChannel) Send(ctx context.Context, msg bus.OutboundMessage) ([]string, error) { if !c.IsRunning() { - return channels.ErrNotRunning + return nil, channels.ErrNotRunning } roomID := id.RoomID(strings.TrimSpace(msg.ChatID)) if roomID == "" { - return fmt.Errorf("matrix room ID is empty: %w", channels.ErrSendFailed) + return nil, fmt.Errorf("matrix room ID is empty: %w", channels.ErrSendFailed) } content := strings.TrimSpace(msg.Content) if content == "" { - return nil + return nil, nil } _, err := c.client.SendMessageEvent(ctx, roomID, event.EventMessage, c.messageContent(content)) if err != nil { - return fmt.Errorf("matrix send: %w", channels.ErrTemporary) + return nil, fmt.Errorf("matrix send: %w", channels.ErrTemporary) } - return nil + return nil, nil } func (c *MatrixChannel) messageContent(text string) *event.MessageEventContent { diff --git a/pkg/channels/onebot/onebot.go b/pkg/channels/onebot/onebot.go index 048be48eb..4e3b52842 100644 --- a/pkg/channels/onebot/onebot.go +++ b/pkg/channels/onebot/onebot.go @@ -391,15 +391,15 @@ func (c *OneBotChannel) Stop(ctx context.Context) error { return nil } -func (c *OneBotChannel) Send(ctx context.Context, msg bus.OutboundMessage) error { +func (c *OneBotChannel) Send(ctx context.Context, msg bus.OutboundMessage) ([]string, error) { if !c.IsRunning() { - return channels.ErrNotRunning + return nil, channels.ErrNotRunning } // Check ctx before entering write path select { case <-ctx.Done(): - return ctx.Err() + return nil, ctx.Err() default: } @@ -408,12 +408,12 @@ func (c *OneBotChannel) Send(ctx context.Context, msg bus.OutboundMessage) error c.mu.Unlock() if conn == nil { - return fmt.Errorf("OneBot WebSocket not connected") + return nil, fmt.Errorf("OneBot WebSocket not connected") } action, params, err := c.buildSendRequest(msg) if err != nil { - return err + return nil, err } echo := fmt.Sprintf("send_%d", atomic.AddInt64(&c.echoCounter, 1)) @@ -426,7 +426,7 @@ func (c *OneBotChannel) Send(ctx context.Context, msg bus.OutboundMessage) error data, err := json.Marshal(req) if err != nil { - return fmt.Errorf("failed to marshal OneBot request: %w", err) + return nil, fmt.Errorf("failed to marshal OneBot request: %w", err) } c.writeMu.Lock() @@ -439,10 +439,10 @@ func (c *OneBotChannel) Send(ctx context.Context, msg bus.OutboundMessage) error logger.ErrorCF("onebot", "Failed to send message", map[string]any{ "error": err.Error(), }) - return fmt.Errorf("onebot send: %w", channels.ErrTemporary) + return nil, fmt.Errorf("onebot send: %w", channels.ErrTemporary) } - return nil + return nil, nil } // SendMedia implements the channels.MediaSender interface. diff --git a/pkg/channels/pico/client.go b/pkg/channels/pico/client.go index 2c335050d..82fc7ca26 100644 --- a/pkg/channels/pico/client.go +++ b/pkg/channels/pico/client.go @@ -273,22 +273,22 @@ func (c *PicoClientChannel) handleServerMessage(pc *picoConn, msg PicoMessage) { } // Send sends a message to the remote server. -func (c *PicoClientChannel) Send(ctx context.Context, msg bus.OutboundMessage) error { +func (c *PicoClientChannel) Send(ctx context.Context, msg bus.OutboundMessage) ([]string, error) { if !c.IsRunning() { - return channels.ErrNotRunning + return nil, channels.ErrNotRunning } c.mu.Lock() pc := c.conn c.mu.Unlock() if pc == nil || pc.closed.Load() { - return channels.ErrSendFailed + return nil, channels.ErrSendFailed } outMsg := newMessage(TypeMessageSend, map[string]any{ "content": msg.Content, }) outMsg.SessionID = strings.TrimPrefix(msg.ChatID, "pico_client:") - return pc.writeJSON(outMsg) + return nil, pc.writeJSON(outMsg) } // StartTyping implements channels.TypingCapable. diff --git a/pkg/channels/pico/client_test.go b/pkg/channels/pico/client_test.go index 118c9abea..3b6b017ff 100644 --- a/pkg/channels/pico/client_test.go +++ b/pkg/channels/pico/client_test.go @@ -46,7 +46,7 @@ func TestSend_NotRunning(t *testing.T) { if err != nil { t.Fatal(err) } - err = ch.Send(context.Background(), bus.OutboundMessage{Content: "hi"}) + _, err = ch.Send(context.Background(), bus.OutboundMessage{Content: "hi"}) if !errors.Is(err, channels.ErrNotRunning) { t.Fatalf("expected ErrNotRunning, got %v", err) } @@ -124,7 +124,7 @@ func TestClientChannel_ConnectAndSend(t *testing.T) { defer ch.Stop(ctx) // Send a message - err = ch.Send(ctx, bus.OutboundMessage{ + _, err = ch.Send(ctx, bus.OutboundMessage{ ChatID: "pico_client:sess-1", Content: "hello", }) @@ -179,7 +179,7 @@ func TestClientChannel_ReceivesServerMessage(t *testing.T) { defer ch.Stop(ctx) // Send a message; the echo server replies with message.create - err = ch.Send(ctx, bus.OutboundMessage{ + _, err = ch.Send(ctx, bus.OutboundMessage{ ChatID: "pico_client:sess-echo", Content: "ping", }) @@ -252,7 +252,7 @@ func TestSend_ClosedConnection(t *testing.T) { ch.conn.close() ch.mu.Unlock() - err = ch.Send(ctx, bus.OutboundMessage{ + _, err = ch.Send(ctx, bus.OutboundMessage{ ChatID: "pico_client:sess-close", Content: "should fail", }) diff --git a/pkg/channels/pico/pico.go b/pkg/channels/pico/pico.go index 1aa1941cf..b0ea4e4c1 100644 --- a/pkg/channels/pico/pico.go +++ b/pkg/channels/pico/pico.go @@ -234,16 +234,16 @@ func (c *PicoChannel) ServeHTTP(w http.ResponseWriter, r *http.Request) { } // Send implements Channel — sends a message to the appropriate WebSocket connection. -func (c *PicoChannel) Send(ctx context.Context, msg bus.OutboundMessage) error { +func (c *PicoChannel) Send(ctx context.Context, msg bus.OutboundMessage) ([]string, error) { if !c.IsRunning() { - return channels.ErrNotRunning + return nil, channels.ErrNotRunning } outMsg := newMessage(TypeMessageCreate, map[string]any{ "content": msg.Content, }) - return c.broadcastToSession(msg.ChatID, outMsg) + return nil, c.broadcastToSession(msg.ChatID, outMsg) } // EditMessage implements channels.MessageEditor. diff --git a/pkg/channels/qq/qq.go b/pkg/channels/qq/qq.go index 38808238e..15349b090 100644 --- a/pkg/channels/qq/qq.go +++ b/pkg/channels/qq/qq.go @@ -200,13 +200,7 @@ func (c *QQChannel) getChatKind(chatID string) string { return "group" } -func (c *QQChannel) Send(ctx context.Context, msg bus.OutboundMessage) error { - _, err := c.SendMessageWithIDs(ctx, msg) - return err -} - -// SendMessageWithIDs implements channels.MessageIDsSender. -func (c *QQChannel) SendMessageWithIDs(ctx context.Context, msg bus.OutboundMessage) ([]string, error) { +func (c *QQChannel) Send(ctx context.Context, msg bus.OutboundMessage) ([]string, error) { if !c.IsRunning() { return nil, channels.ErrNotRunning } diff --git a/pkg/channels/slack/slack.go b/pkg/channels/slack/slack.go index 00061ced4..9ac95735f 100644 --- a/pkg/channels/slack/slack.go +++ b/pkg/channels/slack/slack.go @@ -108,13 +108,7 @@ func (c *SlackChannel) Stop(ctx context.Context) error { return nil } -func (c *SlackChannel) Send(ctx context.Context, msg bus.OutboundMessage) error { - _, err := c.SendMessageWithIDs(ctx, msg) - return err -} - -// SendMessageWithIDs implements channels.MessageIDsSender. -func (c *SlackChannel) SendMessageWithIDs(ctx context.Context, msg bus.OutboundMessage) ([]string, error) { +func (c *SlackChannel) Send(ctx context.Context, msg bus.OutboundMessage) ([]string, error) { if !c.IsRunning() { return nil, channels.ErrNotRunning } diff --git a/pkg/channels/telegram/telegram.go b/pkg/channels/telegram/telegram.go index 76c7ba24c..6941656d8 100644 --- a/pkg/channels/telegram/telegram.go +++ b/pkg/channels/telegram/telegram.go @@ -168,13 +168,7 @@ func (c *TelegramChannel) Stop(ctx context.Context) error { return nil } -func (c *TelegramChannel) Send(ctx context.Context, msg bus.OutboundMessage) error { - _, err := c.SendMessageWithIDs(ctx, msg) - return err -} - -// SendMessageWithIDs implements channels.MessageIDsSender. -func (c *TelegramChannel) SendMessageWithIDs(ctx context.Context, msg bus.OutboundMessage) ([]string, error) { +func (c *TelegramChannel) Send(ctx context.Context, msg bus.OutboundMessage) ([]string, error) { if !c.IsRunning() { return nil, channels.ErrNotRunning } diff --git a/pkg/channels/telegram/telegram_test.go b/pkg/channels/telegram/telegram_test.go index e12ad1f5d..4f6f78f44 100644 --- a/pkg/channels/telegram/telegram_test.go +++ b/pkg/channels/telegram/telegram_test.go @@ -236,7 +236,7 @@ func TestSend_EmptyContent(t *testing.T) { } ch := newTestChannel(t, caller) - err := ch.Send(context.Background(), bus.OutboundMessage{ + _, err := ch.Send(context.Background(), bus.OutboundMessage{ ChatID: "12345", Content: "", }) @@ -253,7 +253,7 @@ func TestSend_ShortMessage_SingleCall(t *testing.T) { } ch := newTestChannel(t, caller) - err := ch.Send(context.Background(), bus.OutboundMessage{ + _, err := ch.Send(context.Background(), bus.OutboundMessage{ ChatID: "12345", Content: "Hello, world!", }) @@ -276,7 +276,7 @@ func TestSend_LongMessage_SingleCall(t *testing.T) { longContent := strings.Repeat("a", 4000) - err := ch.Send(context.Background(), bus.OutboundMessage{ + _, err := ch.Send(context.Background(), bus.OutboundMessage{ ChatID: "12345", Content: longContent, }) @@ -295,7 +295,7 @@ func TestSendMessageWithIDs_ReturnsAllChunkIDsAfterHTMLResplit(t *testing.T) { chunk := "[x](https://example.com/" + strings.Repeat("a", 20) + ") " content := strings.Repeat(chunk, 120) - ids, err := ch.SendMessageWithIDs(context.Background(), bus.OutboundMessage{ + ids, err := ch.Send(context.Background(), bus.OutboundMessage{ ChatID: "12345", Content: content, }) @@ -320,7 +320,7 @@ func TestSend_HTMLFallback_PerChunk(t *testing.T) { } ch := newTestChannel(t, caller) - err := ch.Send(context.Background(), bus.OutboundMessage{ + _, err := ch.Send(context.Background(), bus.OutboundMessage{ ChatID: "12345", Content: "Hello **world**", }) @@ -338,7 +338,7 @@ func TestSend_HTMLFallback_BothFail(t *testing.T) { } ch := newTestChannel(t, caller) - err := ch.Send(context.Background(), bus.OutboundMessage{ + _, err := ch.Send(context.Background(), bus.OutboundMessage{ ChatID: "12345", Content: "Hello", }) @@ -360,7 +360,7 @@ func TestSend_LongMessage_HTMLFallback_StopsOnError(t *testing.T) { longContent := strings.Repeat("x", 4001) - err := ch.Send(context.Background(), bus.OutboundMessage{ + _, err := ch.Send(context.Background(), bus.OutboundMessage{ ChatID: "12345", Content: longContent, }) @@ -390,7 +390,7 @@ func TestSend_MarkdownShortButHTMLLong_MultipleCalls(t *testing.T) { "HTML expansion must exceed Telegram limit for this test to be meaningful", ) - err := ch.Send(context.Background(), bus.OutboundMessage{ + _, err := ch.Send(context.Background(), bus.OutboundMessage{ ChatID: "12345", Content: markdownContent, }) @@ -425,7 +425,7 @@ func TestSend_HTMLOverflow_WordBoundary(t *testing.T) { // Ensure the test content matches the intended boundary conditions. assert.LessOrEqual(t, len([]rune(content)), 4000, "markdown content must not exceed chunk size for this test") - err := ch.Send(context.Background(), bus.OutboundMessage{ + _, err := ch.Send(context.Background(), bus.OutboundMessage{ ChatID: "123456", Content: content, }) @@ -461,7 +461,7 @@ func TestSend_NotRunning(t *testing.T) { ch := newTestChannel(t, caller) ch.SetRunning(false) - err := ch.Send(context.Background(), bus.OutboundMessage{ + _, err := ch.Send(context.Background(), bus.OutboundMessage{ ChatID: "12345", Content: "Hello", }) @@ -479,7 +479,7 @@ func TestSend_InvalidChatID(t *testing.T) { } ch := newTestChannel(t, caller) - err := ch.Send(context.Background(), bus.OutboundMessage{ + _, err := ch.Send(context.Background(), bus.OutboundMessage{ ChatID: "not-a-number", Content: "Hello", }) @@ -536,7 +536,7 @@ func TestSend_WithForumThreadID(t *testing.T) { } ch := newTestChannel(t, caller) - err := ch.Send(context.Background(), bus.OutboundMessage{ + _, err := ch.Send(context.Background(), bus.OutboundMessage{ ChatID: "-1001234567890/42", Content: "Hello from topic", }) diff --git a/pkg/channels/wecom/wecom.go b/pkg/channels/wecom/wecom.go index 26e971921..a80e1d952 100644 --- a/pkg/channels/wecom/wecom.go +++ b/pkg/channels/wecom/wecom.go @@ -184,20 +184,20 @@ func (c *WeComChannel) BeginStream(_ context.Context, chatID string) (channels.S }, nil } -func (c *WeComChannel) Send(ctx context.Context, msg bus.OutboundMessage) error { +func (c *WeComChannel) Send(ctx context.Context, msg bus.OutboundMessage) ([]string, error) { if !c.IsRunning() { - return channels.ErrNotRunning + return nil, channels.ErrNotRunning } content := strings.TrimSpace(msg.Content) if content == "" { - return nil + return nil, nil } if turn, ok := c.getTurn(msg.ChatID); ok { if time.Since(turn.CreatedAt) <= wecomStreamMaxDuration { if err := c.sendStreamReply(turn, content); err == nil { c.consumeTurn(msg.ChatID, turn) - return nil + return nil, nil } } c.consumeTurn(msg.ChatID, turn) @@ -205,15 +205,15 @@ func (c *WeComChannel) Send(ctx context.Context, msg bus.OutboundMessage) error if route, ok := c.routes.Get(msg.ChatID); ok { if err := c.sendActivePush(route.ChatID, route.ChatType, content); err != nil { - return err + return nil, err } - return nil + return nil, nil } if err := c.sendActivePush(msg.ChatID, 0, content); err != nil { - return err + return nil, err } - return nil + return nil, nil } func (c *WeComChannel) SendMedia(ctx context.Context, msg bus.OutboundMediaMessage) error { diff --git a/pkg/channels/weixin/weixin.go b/pkg/channels/weixin/weixin.go index b9e821ef1..e627d5899 100644 --- a/pkg/channels/weixin/weixin.go +++ b/pkg/channels/weixin/weixin.go @@ -313,16 +313,16 @@ func (c *WeixinChannel) handleInboundMessage(ctx context.Context, msg WeixinMess } // Send implements channels.Channel by sending a text message to the WeChat user. -func (c *WeixinChannel) Send(ctx context.Context, msg bus.OutboundMessage) error { +func (c *WeixinChannel) Send(ctx context.Context, msg bus.OutboundMessage) ([]string, error) { if !c.IsRunning() { - return channels.ErrNotRunning + return nil, channels.ErrNotRunning } if err := c.ensureSessionActive(); err != nil { - return err + return nil, err } if msg.Content == "" { - return nil + return nil, nil } // We need a context_token to send a reply. It should be stored in the conversation metadata. @@ -341,7 +341,7 @@ func (c *WeixinChannel) Send(ctx context.Context, msg bus.OutboundMessage) error logger.ErrorCF("weixin", "Missing context token, cannot send message", map[string]any{ "to_user_id": toUserID, }) - return fmt.Errorf("weixin send: %w: missing context token for chat %s", channels.ErrSendFailed, toUserID) + return nil, fmt.Errorf("weixin send: %w: missing context token for chat %s", channels.ErrSendFailed, toUserID) } if err := c.sendTextMessage(ctx, toUserID, contextToken, msg.Content); err != nil { @@ -350,10 +350,10 @@ func (c *WeixinChannel) Send(ctx context.Context, msg bus.OutboundMessage) error "error": err.Error(), }) if c.remainingPause() > 0 { - return fmt.Errorf("weixin send: %w", channels.ErrSendFailed) + return nil, fmt.Errorf("weixin send: %w", channels.ErrSendFailed) } - return fmt.Errorf("weixin send: %w", channels.ErrTemporary) + return nil, fmt.Errorf("weixin send: %w", channels.ErrTemporary) } - return nil + return nil, nil } diff --git a/pkg/channels/whatsapp/whatsapp.go b/pkg/channels/whatsapp/whatsapp.go index 70b3e02bf..98622fe37 100644 --- a/pkg/channels/whatsapp/whatsapp.go +++ b/pkg/channels/whatsapp/whatsapp.go @@ -104,15 +104,15 @@ func (c *WhatsAppChannel) Stop(ctx context.Context) error { return nil } -func (c *WhatsAppChannel) Send(ctx context.Context, msg bus.OutboundMessage) error { +func (c *WhatsAppChannel) Send(ctx context.Context, msg bus.OutboundMessage) ([]string, error) { if !c.IsRunning() { - return channels.ErrNotRunning + return nil, channels.ErrNotRunning } // Check ctx before acquiring lock select { case <-ctx.Done(): - return ctx.Err() + return nil, ctx.Err() default: } @@ -120,7 +120,7 @@ func (c *WhatsAppChannel) Send(ctx context.Context, msg bus.OutboundMessage) err defer c.mu.Unlock() if c.conn == nil { - return fmt.Errorf("whatsapp connection not established: %w", channels.ErrTemporary) + return nil, fmt.Errorf("whatsapp connection not established: %w", channels.ErrTemporary) } payload := map[string]any{ @@ -131,17 +131,17 @@ func (c *WhatsAppChannel) Send(ctx context.Context, msg bus.OutboundMessage) err data, err := json.Marshal(payload) if err != nil { - return fmt.Errorf("failed to marshal message: %w", err) + return nil, fmt.Errorf("failed to marshal message: %w", err) } _ = c.conn.SetWriteDeadline(time.Now().Add(10 * time.Second)) if err := c.conn.WriteMessage(websocket.TextMessage, data); err != nil { _ = c.conn.SetWriteDeadline(time.Time{}) - return fmt.Errorf("whatsapp send: %w", channels.ErrTemporary) + return nil, fmt.Errorf("whatsapp send: %w", channels.ErrTemporary) } _ = c.conn.SetWriteDeadline(time.Time{}) - return nil + return nil, nil } func (c *WhatsAppChannel) listen() { diff --git a/pkg/channels/whatsapp_native/whatsapp_native.go b/pkg/channels/whatsapp_native/whatsapp_native.go index 188a7c8fa..d0a74a405 100644 --- a/pkg/channels/whatsapp_native/whatsapp_native.go +++ b/pkg/channels/whatsapp_native/whatsapp_native.go @@ -396,13 +396,13 @@ func (c *WhatsAppNativeChannel) handleIncoming(evt *events.Message) { c.HandleMessage(c.runCtx, peer, messageID, senderID, chatID, content, mediaPaths, metadata, sender) } -func (c *WhatsAppNativeChannel) Send(ctx context.Context, msg bus.OutboundMessage) error { +func (c *WhatsAppNativeChannel) Send(ctx context.Context, msg bus.OutboundMessage) ([]string, error) { if !c.IsRunning() { - return channels.ErrNotRunning + return nil, channels.ErrNotRunning } select { case <-ctx.Done(): - return ctx.Err() + return nil, ctx.Err() default: } @@ -411,18 +411,18 @@ func (c *WhatsAppNativeChannel) Send(ctx context.Context, msg bus.OutboundMessag c.mu.Unlock() if client == nil || !client.IsConnected() { - return fmt.Errorf("whatsapp connection not established: %w", channels.ErrTemporary) + return nil, fmt.Errorf("whatsapp connection not established: %w", channels.ErrTemporary) } // Detect unpaired state: the client is connected (to WhatsApp servers) // but has not completed QR-login yet, so sending would fail. if client.Store.ID == nil { - return fmt.Errorf("whatsapp not yet paired (QR login pending): %w", channels.ErrTemporary) + return nil, fmt.Errorf("whatsapp not yet paired (QR login pending): %w", channels.ErrTemporary) } to, err := parseJID(msg.ChatID) if err != nil { - return fmt.Errorf("invalid chat id %q: %w", msg.ChatID, err) + return nil, fmt.Errorf("invalid chat id %q: %w", msg.ChatID, err) } waMsg := &waE2E.Message{ @@ -430,9 +430,9 @@ func (c *WhatsAppNativeChannel) Send(ctx context.Context, msg bus.OutboundMessag } if _, err = client.SendMessage(ctx, to, waMsg); err != nil { - return fmt.Errorf("whatsapp send: %w", channels.ErrTemporary) + return nil, fmt.Errorf("whatsapp send: %w", channels.ErrTemporary) } - return nil + return nil, nil } // parseJID converts a chat ID (phone number or JID string) to types.JID.