opencode/picoclaw
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

refactor(channels): Send returns ([]string, error), remove MessageIDsSender

Browse source 
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.
This commit is contained in:
Dmitrii Balabanov 2026-03-23 11:07:42 +02:00
parent 7f91fa90f2
commit 4d5d1d0509
27 changed files with 170 additions and 196 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
pkg/agent/loop_test.go
View file

@ -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 }

48
pkg/channels/README.md
View file

@ -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

48
pkg/channels/README.zh.md
View file

@ -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

2
pkg/channels/base.go
View file

@ -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

10
pkg/channels/dingtalk/dingtalk.go
View file

@ -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

8
pkg/channels/discord/discord.go
View file

@ -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
}

4
pkg/channels/feishu/feishu_32.go
View file

@ -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

16
pkg/channels/feishu/feishu_64.go
View file

@ -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.

6
pkg/channels/interfaces.go
View file

@ -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.

10
pkg/channels/irc/irc.go
View file

@ -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.

8
pkg/channels/line/line.go
View file

@ -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.

12
pkg/channels/maixcam/maixcam.go
View file

@ -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
}

11
pkg/channels/manager.go
View file

@ -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
}

11
pkg/channels/manager_test.go
View file

@ -26,21 +26,16 @@ 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)
if m.sendWithIDsFn != nil {
return m.sendWithIDsFn(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)
}
return m.sendWithIDsFn(ctx, msg)
}
func (m *mockChannel) Start(ctx context.Context) error { return nil }
func (m *mockChannel) Stop(ctx context.Context) error { return nil }

12
pkg/channels/matrix/matrix.go
View file

@ -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 {

16
pkg/channels/onebot/onebot.go
View file

@ -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.

8
pkg/channels/pico/client.go
View file

@ -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.

8
pkg/channels/pico/client_test.go
View file

@ -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",
})

6
pkg/channels/pico/pico.go
View file

@ -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.

8
pkg/channels/qq/qq.go
View file

@ -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
}

8
pkg/channels/slack/slack.go
View file

@ -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
}

8
pkg/channels/telegram/telegram.go
View file

@ -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
}

24
pkg/channels/telegram/telegram_test.go
View file

@ -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",
})

16
pkg/channels/wecom/wecom.go
View file

@ -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 {

16
pkg/channels/weixin/weixin.go
View file

@ -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
}

14
pkg/channels/whatsapp/whatsapp.go
View file

@ -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() {

16
pkg/channels/whatsapp_native/whatsapp_native.go
View file

@ -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.