From bf272d8dbff4def7c05b26fa0dfefe0233f60b64 Mon Sep 17 00:00:00 2001 From: viettranx Date: Sat, 11 Apr 2026 21:06:34 +0700 Subject: [PATCH] fix(feishu): route thread replies via Lark reply endpoint Bot responses to messages inside Lark topic threads were dropped outside the thread because outbound Send always used the new-message endpoint. This change: - Adds LarkClient.ReplyMessage() that POSTs to /open-apis/im/v1/messages/{id}/reply with reply_in_thread=true - Parses thread_id from im.message.receive_v1 events (distinct from root_id which fires on any quote reply) and stamps feishu_reply_target_id into the message metadata - Propagates the key through cmd/gateway_consumer_normal.go and the new package-level routingMetaKeys var in internal/channels/events.go so block replies and retry notifications also land in thread - Routes sendText, sendMarkdownCard, sendImage, sendFile, and sendMediaAttachment via a new deliverMessage helper that falls back to SendMessage with a warning log on reply endpoint errors (e.g. thread root deleted) - Adds 10 unit tests covering routing, fallback, content double-encoding, and the thread_id gate that prevents plain quote replies from being silently promoted to threads Closes #818 --- cmd/gateway_consumer_normal.go | 2 +- docs/05-channels-messaging.md | 9 + docs/17-changelog.md | 7 + internal/channels/events.go | 21 +- internal/channels/feishu/bot.go | 21 +- internal/channels/feishu/bot_parse.go | 2 + internal/channels/feishu/bot_policy.go | 2 +- internal/channels/feishu/feishu.go | 59 +++- .../channels/feishu/larkclient_messaging.go | 36 +++ .../feishu/larkclient_messaging_test.go | 90 +++++++ internal/channels/feishu/larkevents.go | 6 + internal/channels/feishu/media.go | 30 ++- internal/channels/feishu/send_routing_test.go | 254 ++++++++++++++++++ 13 files changed, 504 insertions(+), 35 deletions(-) create mode 100644 internal/channels/feishu/larkclient_messaging_test.go create mode 100644 internal/channels/feishu/send_routing_test.go diff --git a/cmd/gateway_consumer_normal.go b/cmd/gateway_consumer_normal.go index aa1609a0..0180f080 100644 --- a/cmd/gateway_consumer_normal.go +++ b/cmd/gateway_consumer_normal.go @@ -220,7 +220,7 @@ func processNormalMessage( outMeta["reply_to_message_id"] = mid } } - for _, k := range []string{tools.MetaMessageThreadID, "local_key", "placeholder_key", "group_id"} { + for _, k := range []string{tools.MetaMessageThreadID, "local_key", "placeholder_key", "group_id", "feishu_reply_target_id"} { if v := msg.Metadata[k]; v != "" { outMeta[k] = v } diff --git a/docs/05-channels-messaging.md b/docs/05-channels-messaging.md index d3c8607f..a1dd9ce8 100644 --- a/docs/05-channels-messaging.md +++ b/docs/05-channels-messaging.md @@ -367,6 +367,15 @@ When enabled, each thread gets an isolated session: - Different threads within the same group maintain separate conversation histories - Disabled by default +### Thread Reply Routing + +When a message is sent inside a Lark thread (detected via the `thread_id` field in the inbound event), the inbound handler stamps `metadata["feishu_reply_target_id"]` with the triggering message ID. During outbound delivery, the channel routes responses back to the same thread using `LarkClient.ReplyMessage()` which POSTs to `/open-apis/im/v1/messages/{message_id}/reply` with `reply_in_thread: true`. + +- **Automatic thread detection**: No configuration needed; replies are routed based on inbound `thread_id` +- **Metadata propagation**: The `feishu_reply_target_id` key is included in the `routingMetaKeys` allowlist so replies, block replies, and placeholder updates all land in the correct thread +- **Graceful fallback**: If the reply endpoint fails (e.g., thread root deleted), the channel falls back to `SendMessage()` for the regular chat +- **Applies to**: Text, card, image, and file messages + --- ## 7. Discord diff --git a/docs/17-changelog.md b/docs/17-changelog.md index 9e495545..4bed53df 100644 --- a/docs/17-changelog.md +++ b/docs/17-changelog.md @@ -32,6 +32,13 @@ All notable changes to GoClaw Gateway are documented here. Format follows [Keep ## [Unreleased] +### Fixed + +#### Feishu/Lark Thread Reply Routing — Issue #818 (2026-04-11) +- **Thread detection**: Inbound messages with `thread_id` now properly route responses back to the same Feishu thread via `/open-apis/im/v1/messages/{id}/reply` endpoint +- **Metadata propagation**: New `feishu_reply_target_id` metadata key added to `routingMetaKeys` allowlist so all outbound messages (text, cards, files, reactions) land in the correct thread +- **Graceful fallback**: If thread root is deleted, channel falls back to regular `SendMessage()` for robustness + ### Testing #### Test Coverage Improvement — Wave 1-3 (2026-04-11) diff --git a/internal/channels/events.go b/internal/channels/events.go index 0b10c144..127714fd 100644 --- a/internal/channels/events.go +++ b/internal/channels/events.go @@ -263,10 +263,12 @@ func (m *Manager) HandleAgentEvent(eventType, runID string, payload any) { // Build outbound metadata: copy routing fields but strip reply_to_message_id // (block replies are standalone) and placeholder_key (reserve for final message). + // feishu_reply_target_id MUST be preserved so intermediate block replies for + // threaded Lark messages also land inside the same thread. var outMeta map[string]string if rc.Metadata != nil { outMeta = make(map[string]string) - for _, k := range []string{"message_thread_id", "local_key", "group_id"} { + for _, k := range routingMetaKeys { if v := rc.Metadata[k]; v != "" { outMeta[k] = v } @@ -344,11 +346,22 @@ func extractPayloadString(payload any, key string) string { return "" } -// copyRoutingMeta copies channel routing metadata (thread_id, local_key, group_id) -// from RunContext.Metadata into a new map suitable for outbound messages. +// routingMetaKeys enumerates the metadata keys that must survive the hop from +// inbound RunContext.Metadata into outbound OutboundMessage.Metadata so that +// replies, block replies, retries, and placeholder updates all land in the +// correct thread / topic / subgroup routing bucket on each channel. +var routingMetaKeys = []string{ + "message_thread_id", // telegram forum topics + "local_key", // composite chat-id suffix + "group_id", // legacy group identifier + "feishu_reply_target_id", // feishu/lark thread reply routing (issue #818) +} + +// copyRoutingMeta copies channel routing metadata from RunContext.Metadata +// into a new map suitable for outbound messages. func copyRoutingMeta(src map[string]string) map[string]string { out := make(map[string]string) - for _, k := range []string{"message_thread_id", "local_key", "group_id"} { + for _, k := range routingMetaKeys { if v := src[k]; v != "" { out[k] = v } diff --git a/internal/channels/feishu/bot.go b/internal/channels/feishu/bot.go index 4020c6a3..1b219b4f 100644 --- a/internal/channels/feishu/bot.go +++ b/internal/channels/feishu/bot.go @@ -20,8 +20,9 @@ type messageContext struct { Content string ContentType string // "text", "post", "image", etc. MentionedBot bool - RootID string // thread root message ID - ParentID string // parent message ID + RootID string // reply-chain root (populated on ANY reply, incl. plain quote reply) + ParentID string // direct parent in reply chain + ThreadID string // set ONLY when message is inside an actual topic thread Mentions []mentionInfo } @@ -174,6 +175,22 @@ func (c *Channel) handleMessageEvent(ctx context.Context, event *MessageEvent) { "platform": channels.TypeFeishu, } + // Thread routing: stamp the triggering message ID ONLY when the inbound + // message is inside an actual topic thread (thread_id present per Lark + // docs). We deliberately do NOT fire on mc.RootID — Lark populates root_id + // on every reply including plain quote replies outside any thread, and + // routing those through the reply endpoint would silently promote them to + // new threads. thread_id is the definitive signal. + // + // Outbound Send() reads this key and, when non-empty, routes to the Lark + // reply endpoint with reply_in_thread=true so the bot response lands + // inside the same thread. Absent on non-thread messages — preserves + // existing new-message endpoint behavior for DMs, plain groups, and quote + // replies. + if mc.ThreadID != "" { + metadata["feishu_reply_target_id"] = messageID + } + if sender != nil { metadata["sender_open_id"] = sender.SenderID.OpenID } diff --git a/internal/channels/feishu/bot_parse.go b/internal/channels/feishu/bot_parse.go index 5b8fea62..ee3dfbda 100644 --- a/internal/channels/feishu/bot_parse.go +++ b/internal/channels/feishu/bot_parse.go @@ -16,6 +16,7 @@ func (c *Channel) parseMessageEvent(event *MessageEvent) *messageContext { contentType := msg.MessageType rootID := msg.RootID parentID := msg.ParentID + threadID := msg.ThreadID senderID := "" if sender != nil { @@ -58,6 +59,7 @@ func (c *Channel) parseMessageEvent(event *MessageEvent) *messageContext { MentionedBot: mentionedBot, RootID: rootID, ParentID: parentID, + ThreadID: threadID, Mentions: mentions, } } diff --git a/internal/channels/feishu/bot_policy.go b/internal/channels/feishu/bot_policy.go index ee581dc3..151f787f 100644 --- a/internal/channels/feishu/bot_policy.go +++ b/internal/channels/feishu/bot_policy.go @@ -127,7 +127,7 @@ func (c *Channel) sendPairingReply(ctx context.Context, senderID, chatID string) ) receiveIDType := resolveReceiveIDType(chatID) - if err := c.sendText(context.Background(), chatID, receiveIDType, replyText); err != nil { + if err := c.sendText(context.Background(), chatID, receiveIDType, replyText, ""); err != nil { slog.Warn("failed to send feishu pairing reply", "error", err) } else { c.MarkPairingNotifSent(senderID) diff --git a/internal/channels/feishu/feishu.go b/internal/channels/feishu/feishu.go index d218eb07..48760ef3 100644 --- a/internal/channels/feishu/feishu.go +++ b/internal/channels/feishu/feishu.go @@ -168,6 +168,13 @@ func (c *Channel) Send(ctx context.Context, msg bus.OutboundMessage) error { // Determine receive_id_type receiveIDType := resolveReceiveIDType(chatID) + // Thread reply: when the inbound message was inside a Lark thread, the + // Feishu inbound handler stamps metadata["feishu_reply_target_id"] with + // the triggering message ID so responses land back inside the same thread + // via POST /open-apis/im/v1/messages/{id}/reply with reply_in_thread=true. + // Absent on non-thread messages — Send falls back to the new-message path. + replyTargetID := msg.Metadata["feishu_reply_target_id"] + // Send text content text := msg.Content if text != "" { @@ -191,19 +198,19 @@ func (c *Channel) Send(ctx context.Context, msg bus.OutboundMessage) error { } if useCard { - if err := c.sendMarkdownCard(ctx, chatID, receiveIDType, text, nil); err != nil { + if err := c.sendMarkdownCard(ctx, chatID, receiveIDType, text, replyTargetID, nil); err != nil { return err } } else { - if err := c.sendChunkedText(ctx, chatID, receiveIDType, text, chunkLimit); err != nil { + if err := c.sendChunkedText(ctx, chatID, receiveIDType, text, chunkLimit, replyTargetID); err != nil { return err } } } - // Send media attachments + // Send media attachments — same thread routing applies as text. for _, media := range msg.Media { - if err := c.sendMediaAttachment(ctx, chatID, receiveIDType, media); err != nil { + if err := c.sendMediaAttachment(ctx, chatID, receiveIDType, media, replyTargetID); err != nil { slog.Warn("feishu send media failed", "url", media.URL, "error", err) } } @@ -323,34 +330,58 @@ func (c *Channel) probeBotInfo(ctx context.Context) error { // --- Send helpers --- -func (c *Channel) sendChunkedText(ctx context.Context, chatID, receiveIDType, text string, chunkLimit int) error { +func (c *Channel) sendChunkedText(ctx context.Context, chatID, receiveIDType, text string, chunkLimit int, replyTargetID string) error { for _, chunk := range channels.ChunkMarkdown(text, chunkLimit) { - if err := c.sendText(ctx, chatID, receiveIDType, chunk); err != nil { + if err := c.sendText(ctx, chatID, receiveIDType, chunk, replyTargetID); err != nil { return err } } return nil } -func (c *Channel) sendText(ctx context.Context, chatID, receiveIDType, text string) error { - content := buildPostContent(text) +// deliverMessage routes a message either through the Lark reply endpoint +// (when replyTargetID is non-empty) or the new-message endpoint. On reply +// endpoint failure — typically because the original thread-root message was +// deleted — it falls back to the new-message endpoint so the user still +// receives the response even if thread placement is lost. The fallback path +// logs a warning so operators can diagnose stale thread references. +func (c *Channel) deliverMessage(ctx context.Context, chatID, receiveIDType, replyTargetID, msgType, content string) error { + if replyTargetID != "" { + if _, err := c.client.ReplyMessage(ctx, replyTargetID, msgType, content, true); err == nil { + return nil + } else { + slog.Warn("feishu.reply_failed_fallback_send", + "reply_target_id", replyTargetID, + "msg_type", msgType, + "error", err, + ) + // Fall through to new-message endpoint. + } + } + if _, err := c.client.SendMessage(ctx, receiveIDType, chatID, msgType, content); err != nil { + return err + } + return nil +} - _, err := c.client.SendMessage(ctx, receiveIDType, chatID, "post", content) - if err != nil { +// sendText sends a Lark "post" message. When replyTargetID is non-empty, the +// message is routed through the reply endpoint with reply_in_thread=true so it +// stays nested inside the original thread. +func (c *Channel) sendText(ctx context.Context, chatID, receiveIDType, text, replyTargetID string) error { + content := buildPostContent(text) + if err := c.deliverMessage(ctx, chatID, receiveIDType, replyTargetID, "post", content); err != nil { return fmt.Errorf("feishu send text: %w", err) } return nil } -func (c *Channel) sendMarkdownCard(ctx context.Context, chatID, receiveIDType, text string, metadata map[string]string) error { +func (c *Channel) sendMarkdownCard(ctx context.Context, chatID, receiveIDType, text, replyTargetID string, metadata map[string]string) error { card := buildMarkdownCard(text) cardJSON, err := json.Marshal(card) if err != nil { return fmt.Errorf("marshal card: %w", err) } - - _, err = c.client.SendMessage(ctx, receiveIDType, chatID, "interactive", string(cardJSON)) - if err != nil { + if err := c.deliverMessage(ctx, chatID, receiveIDType, replyTargetID, "interactive", string(cardJSON)); err != nil { return fmt.Errorf("feishu send card: %w", err) } return nil diff --git a/internal/channels/feishu/larkclient_messaging.go b/internal/channels/feishu/larkclient_messaging.go index 3146ff5e..06da51c0 100644 --- a/internal/channels/feishu/larkclient_messaging.go +++ b/internal/channels/feishu/larkclient_messaging.go @@ -37,6 +37,42 @@ func (c *LarkClient) SendMessage(ctx context.Context, receiveIDType, receiveID, return &data, nil } +// ReplyMessage posts a reply to an existing Lark message via +// POST /open-apis/im/v1/messages/{message_id}/reply. +// +// When replyInThread is true and the target message is inside a thread, +// the reply stays nested in that thread. When the target is not threaded, +// the reply renders as an inline quote at chat level. +// +// NOTE: `content` must be a double-encoded JSON string (e.g. `{"text":"hi"}`), +// not a Go struct. The Lark API rejects object-typed `content`. +func (c *LarkClient) ReplyMessage(ctx context.Context, rootMessageID, msgType, content string, replyInThread bool) (*SendMessageResp, error) { + if rootMessageID == "" { + return nil, fmt.Errorf("reply message: empty root message id") + } + // Defensive escaping: Lark message IDs are currently `om_` + alphanumeric, + // but any future schema change that permits `/` or reserved chars would + // otherwise silently break URL routing. + path := fmt.Sprintf("/open-apis/im/v1/messages/%s/reply", url.PathEscape(rootMessageID)) + body := map[string]any{ + "msg_type": msgType, + "content": content, + "reply_in_thread": replyInThread, + } + resp, err := c.doJSON(ctx, "POST", path, body) + if err != nil { + return nil, err + } + if resp.Code != 0 { + return nil, fmt.Errorf("reply message: code=%d msg=%s", resp.Code, resp.Msg) + } + var data SendMessageResp + if err := json.Unmarshal(resp.Data, &data); err != nil { + return nil, fmt.Errorf("unmarshal response: %w", err) + } + return &data, nil +} + // --- IM API: Images --- func (c *LarkClient) DownloadImage(ctx context.Context, imageKey string) ([]byte, error) { diff --git a/internal/channels/feishu/larkclient_messaging_test.go b/internal/channels/feishu/larkclient_messaging_test.go new file mode 100644 index 00000000..dddf7f4f --- /dev/null +++ b/internal/channels/feishu/larkclient_messaging_test.go @@ -0,0 +1,90 @@ +package feishu + +import ( + "context" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" +) + +// TestReplyMessage_InThread verifies the client hits the correct reply endpoint +// with reply_in_thread=true and a double-encoded JSON content string. +func TestReplyMessage_InThread(t *testing.T) { + const rootMsgID = "om_root_1234567890" + const wantContent = `{"text":"hello from thread"}` + + var gotMethod, gotPath, gotAuth, gotContentType string + var gotBody map[string]any + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == tokenEndpoint { + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"code":0,"msg":"ok","tenant_access_token":"fake-token","expire":7200}`)) + return + } + gotMethod = r.Method + gotPath = r.URL.Path + gotAuth = r.Header.Get("Authorization") + gotContentType = r.Header.Get("Content-Type") + raw, _ := io.ReadAll(r.Body) + _ = json.Unmarshal(raw, &gotBody) + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"code":0,"msg":"","data":{"message_id":"om_reply_abc"}}`)) + })) + defer srv.Close() + + client := NewLarkClient("fake-app", "fake-secret", srv.URL) + resp, err := client.ReplyMessage(context.Background(), rootMsgID, "text", wantContent, true) + if err != nil { + t.Fatalf("ReplyMessage returned error: %v", err) + } + if resp == nil || resp.MessageID != "om_reply_abc" { + t.Fatalf("unexpected response: %+v", resp) + } + + // Path must be the reply endpoint, not the new-message endpoint. + wantPath := "/open-apis/im/v1/messages/" + rootMsgID + "/reply" + if gotPath != wantPath { + t.Errorf("path: got %q, want %q", gotPath, wantPath) + } + if gotMethod != http.MethodPost { + t.Errorf("method: got %q, want POST", gotMethod) + } + if !strings.HasPrefix(gotAuth, "Bearer fake-token") { + t.Errorf("authorization header missing or wrong: %q", gotAuth) + } + if !strings.HasPrefix(gotContentType, "application/json") { + t.Errorf("content-type: got %q", gotContentType) + } + + // Body assertions: content must be a JSON STRING (double-encoded), not an object. + if gotContent, _ := gotBody["content"].(string); gotContent != wantContent { + t.Errorf("content field: got %v (type %T), want string %q", gotBody["content"], gotBody["content"], wantContent) + } + if msgType, _ := gotBody["msg_type"].(string); msgType != "text" { + t.Errorf("msg_type: got %v, want %q", gotBody["msg_type"], "text") + } + if rit, ok := gotBody["reply_in_thread"].(bool); !ok || !rit { + t.Errorf("reply_in_thread: got %v, want true", gotBody["reply_in_thread"]) + } +} + +// TestReplyMessage_APIError verifies that a non-zero Lark code surfaces as an error. +func TestReplyMessage_APIError(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == tokenEndpoint { + _, _ = w.Write([]byte(`{"code":0,"msg":"ok","tenant_access_token":"t","expire":7200}`)) + return + } + _, _ = w.Write([]byte(`{"code":230002,"msg":"message not found","data":{}}`)) + })) + defer srv.Close() + + client := NewLarkClient("a", "s", srv.URL) + if _, err := client.ReplyMessage(context.Background(), "om_missing", "text", `{"text":"x"}`, true); err == nil { + t.Fatal("expected error on non-zero code, got nil") + } +} diff --git a/internal/channels/feishu/larkevents.go b/internal/channels/feishu/larkevents.go index ccb3e9d8..9b1f0f2b 100644 --- a/internal/channels/feishu/larkevents.go +++ b/internal/channels/feishu/larkevents.go @@ -45,6 +45,12 @@ type EventMessage struct { MessageID string `json:"message_id"` RootID string `json:"root_id"` ParentID string `json:"parent_id"` + // ThreadID is the definitive "this message lives inside a thread" signal + // per Lark docs. Unlike RootID (which is populated on ANY reply — including + // plain quote replies), ThreadID is only present when the message is in an + // actual topic thread. Used to decide whether to reply via the reply + // endpoint with reply_in_thread=true. + ThreadID string `json:"thread_id"` ChatID string `json:"chat_id"` ChatType string `json:"chat_type"` MessageType string `json:"message_type"` diff --git a/internal/channels/feishu/media.go b/internal/channels/feishu/media.go index 8615a8dc..2eb99187 100644 --- a/internal/channels/feishu/media.go +++ b/internal/channels/feishu/media.go @@ -34,22 +34,25 @@ func (c *Channel) uploadFile(ctx context.Context, data io.Reader, fileName, file // --- Send media --- -// sendImage sends an image message using an image_key. -func (c *Channel) sendImage(ctx context.Context, chatID, receiveIDType, imageKey string) error { +// sendImage sends an image message using an image_key. When replyTargetID is +// non-empty, the image is posted through the reply endpoint so it lands inside +// the same thread as the triggering message. Falls back to the new-message +// endpoint on reply errors (see deliverMessage). +func (c *Channel) sendImage(ctx context.Context, chatID, receiveIDType, imageKey, replyTargetID string) error { contentBytes, err := json.Marshal(map[string]string{"image_key": imageKey}) if err != nil { return fmt.Errorf("marshal image content: %w", err) } - _, err = c.client.SendMessage(ctx, receiveIDType, chatID, "image", string(contentBytes)) - if err != nil { + if err := c.deliverMessage(ctx, chatID, receiveIDType, replyTargetID, "image", string(contentBytes)); err != nil { return fmt.Errorf("feishu send image: %w", err) } return nil } -// sendFile sends a file message using a file_key. -// msgType: "file" for documents, "media" for audio/video. -func (c *Channel) sendFile(ctx context.Context, chatID, receiveIDType, fileKey, msgType string) error { +// sendFile sends a file message using a file_key. When replyTargetID is +// non-empty, the file is posted through the reply endpoint to stay inside the +// thread. msgType: "file" for documents, "media" for audio/video. +func (c *Channel) sendFile(ctx context.Context, chatID, receiveIDType, fileKey, msgType, replyTargetID string) error { if msgType == "" { msgType = "file" } @@ -57,8 +60,7 @@ func (c *Channel) sendFile(ctx context.Context, chatID, receiveIDType, fileKey, if err != nil { return fmt.Errorf("marshal file content: %w", err) } - _, err = c.client.SendMessage(ctx, receiveIDType, chatID, msgType, string(contentBytes)) - if err != nil { + if err := c.deliverMessage(ctx, chatID, receiveIDType, replyTargetID, msgType, string(contentBytes)); err != nil { return fmt.Errorf("feishu send file: %w", err) } return nil @@ -68,7 +70,9 @@ func (c *Channel) sendFile(ctx context.Context, chatID, receiveIDType, fileKey, // sendMediaAttachment uploads and sends a media attachment routed by MIME type. // Images → image message, audio/video → media message (inline playable), others → file message. -func (c *Channel) sendMediaAttachment(ctx context.Context, chatID, receiveIDType string, att bus.MediaAttachment) error { +// When replyTargetID is non-empty, the message is posted through the Lark +// reply endpoint so it lands inside the same thread as the triggering message. +func (c *Channel) sendMediaAttachment(ctx context.Context, chatID, receiveIDType string, att bus.MediaAttachment, replyTargetID string) error { filePath := att.URL if filePath == "" { return nil @@ -88,7 +92,7 @@ func (c *Channel) sendMediaAttachment(ctx context.Context, chatID, receiveIDType if err != nil { return fmt.Errorf("upload image: %w", err) } - return c.sendImage(ctx, chatID, receiveIDType, imageKey) + return c.sendImage(ctx, chatID, receiveIDType, imageKey, replyTargetID) case strings.HasPrefix(ct, "video/"), strings.HasPrefix(ct, "audio/"): // Lark "media" message type plays audio/video inline. @@ -98,7 +102,7 @@ func (c *Channel) sendMediaAttachment(ctx context.Context, chatID, receiveIDType if err != nil { return fmt.Errorf("upload media: %w", err) } - return c.sendFile(ctx, chatID, receiveIDType, fileKey, "media") + return c.sendFile(ctx, chatID, receiveIDType, fileKey, "media", replyTargetID) default: fileName := filepath.Base(filePath) @@ -107,7 +111,7 @@ func (c *Channel) sendMediaAttachment(ctx context.Context, chatID, receiveIDType if err != nil { return fmt.Errorf("upload file: %w", err) } - return c.sendFile(ctx, chatID, receiveIDType, fileKey, "file") + return c.sendFile(ctx, chatID, receiveIDType, fileKey, "file", replyTargetID) } } diff --git a/internal/channels/feishu/send_routing_test.go b/internal/channels/feishu/send_routing_test.go new file mode 100644 index 00000000..51f880d2 --- /dev/null +++ b/internal/channels/feishu/send_routing_test.go @@ -0,0 +1,254 @@ +package feishu + +import ( + "context" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" +) + +// recordedRequest captures what the helper sent to the mocked Lark server +// so the test can assert which endpoint was hit. +type recordedRequest struct { + method string + path string + body map[string]any +} + +// newMockLarkServer returns an httptest server that records the first non-token +// request it receives. Subsequent requests still succeed with code=0 but are +// not recorded. +func newMockLarkServer(t *testing.T) (*httptest.Server, *recordedRequest) { + t.Helper() + rec := &recordedRequest{} + var recorded bool + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == tokenEndpoint { + _, _ = w.Write([]byte(`{"code":0,"msg":"ok","tenant_access_token":"tok","expire":7200}`)) + return + } + if !recorded { + rec.method = r.Method + rec.path = r.URL.Path + raw, _ := io.ReadAll(r.Body) + _ = json.Unmarshal(raw, &rec.body) + recorded = true + } + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"code":0,"msg":"","data":{"message_id":"om_out"}}`)) + })) + t.Cleanup(srv.Close) + return srv, rec +} + +// TestSendText_InThread_RoutesToReplyEndpoint verifies that when replyTargetID +// is non-empty, sendText uses the reply endpoint with reply_in_thread=true. +func TestSendText_InThread_RoutesToReplyEndpoint(t *testing.T) { + srv, rec := newMockLarkServer(t) + + ch := &Channel{client: NewLarkClient("app", "secret", srv.URL)} + err := ch.sendText(context.Background(), "oc_chat_1", "chat_id", "hello in thread", "om_trigger_42") + if err != nil { + t.Fatalf("sendText returned error: %v", err) + } + + wantPath := "/open-apis/im/v1/messages/om_trigger_42/reply" + if rec.path != wantPath { + t.Errorf("path: got %q, want %q", rec.path, wantPath) + } + if rec.method != http.MethodPost { + t.Errorf("method: got %q, want POST", rec.method) + } + if rit, _ := rec.body["reply_in_thread"].(bool); !rit { + t.Errorf("reply_in_thread: got %v, want true", rec.body["reply_in_thread"]) + } + if mt, _ := rec.body["msg_type"].(string); mt != "post" { + t.Errorf("msg_type: got %v, want %q", rec.body["msg_type"], "post") + } + // content is the Lark "post" structure encoded as a JSON string (double-encoded). + content, ok := rec.body["content"].(string) + if !ok || content == "" { + t.Errorf("content: got %v, want non-empty string", rec.body["content"]) + } + if !strings.Contains(content, "hello in thread") { + t.Errorf("content missing text: %q", content) + } +} + +// TestSendText_NoThread_RoutesToNewMessageEndpoint verifies that when +// replyTargetID is empty, sendText falls through to the original SendMessage +// path (preserving existing UX for DMs and non-thread group messages). +func TestSendText_NoThread_RoutesToNewMessageEndpoint(t *testing.T) { + srv, rec := newMockLarkServer(t) + + ch := &Channel{client: NewLarkClient("app", "secret", srv.URL)} + err := ch.sendText(context.Background(), "oc_chat_1", "chat_id", "plain msg", "") + if err != nil { + t.Fatalf("sendText returned error: %v", err) + } + + // New-message path — path is /open-apis/im/v1/messages (receive_id_type is a query param). + wantPath := "/open-apis/im/v1/messages" + if rec.path != wantPath { + t.Errorf("path: got %q, want %q", rec.path, wantPath) + } + // No reply_in_thread field expected on new-message endpoint. + if _, present := rec.body["reply_in_thread"]; present { + t.Errorf("reply_in_thread should be absent on new-message endpoint, got %v", rec.body["reply_in_thread"]) + } + if rid, _ := rec.body["receive_id"].(string); rid != "oc_chat_1" { + t.Errorf("receive_id: got %v, want %q", rec.body["receive_id"], "oc_chat_1") + } +} + +// TestSendImage_InThread_RoutesToReplyEndpoint verifies image sends route to +// the reply endpoint when the message originated inside a thread. +func TestSendImage_InThread_RoutesToReplyEndpoint(t *testing.T) { + srv, rec := newMockLarkServer(t) + + ch := &Channel{client: NewLarkClient("app", "secret", srv.URL)} + err := ch.sendImage(context.Background(), "oc_chat_1", "chat_id", "img_key_123", "om_trigger_img") + if err != nil { + t.Fatalf("sendImage returned error: %v", err) + } + + wantPath := "/open-apis/im/v1/messages/om_trigger_img/reply" + if rec.path != wantPath { + t.Errorf("path: got %q, want %q", rec.path, wantPath) + } + if mt, _ := rec.body["msg_type"].(string); mt != "image" { + t.Errorf("msg_type: got %v, want %q", rec.body["msg_type"], "image") + } + if rit, _ := rec.body["reply_in_thread"].(bool); !rit { + t.Errorf("reply_in_thread: got %v, want true", rec.body["reply_in_thread"]) + } + if !strings.Contains(rec.body["content"].(string), "img_key_123") { + t.Errorf("content missing image key: %v", rec.body["content"]) + } +} + +// TestSendFile_InThread_RoutesToReplyEndpoint verifies file sends (documents + +// audio/video via "media" msg_type) route to reply endpoint inside threads. +func TestSendFile_InThread_RoutesToReplyEndpoint(t *testing.T) { + srv, rec := newMockLarkServer(t) + + ch := &Channel{client: NewLarkClient("app", "secret", srv.URL)} + err := ch.sendFile(context.Background(), "oc_chat_1", "chat_id", "file_key_abc", "file", "om_trigger_file") + if err != nil { + t.Fatalf("sendFile returned error: %v", err) + } + + wantPath := "/open-apis/im/v1/messages/om_trigger_file/reply" + if rec.path != wantPath { + t.Errorf("path: got %q, want %q", rec.path, wantPath) + } + if mt, _ := rec.body["msg_type"].(string); mt != "file" { + t.Errorf("msg_type: got %v, want %q", rec.body["msg_type"], "file") + } +} + +// TestSendMarkdownCard_InThread_RoutesToReplyEndpoint verifies card sends also +// respect thread routing. +func TestSendMarkdownCard_InThread_RoutesToReplyEndpoint(t *testing.T) { + srv, rec := newMockLarkServer(t) + + ch := &Channel{client: NewLarkClient("app", "secret", srv.URL)} + err := ch.sendMarkdownCard(context.Background(), "oc_chat_1", "chat_id", "**bold**", "om_trigger_99", nil) + if err != nil { + t.Fatalf("sendMarkdownCard returned error: %v", err) + } + + wantPath := "/open-apis/im/v1/messages/om_trigger_99/reply" + if rec.path != wantPath { + t.Errorf("path: got %q, want %q", rec.path, wantPath) + } + if mt, _ := rec.body["msg_type"].(string); mt != "interactive" { + t.Errorf("msg_type: got %v, want %q", rec.body["msg_type"], "interactive") + } + if rit, _ := rec.body["reply_in_thread"].(bool); !rit { + t.Errorf("reply_in_thread: got %v, want true", rec.body["reply_in_thread"]) + } +} + +// TestDeliverMessage_ReplyFailure_FallsBackToSendMessage verifies that when +// the reply endpoint returns a Lark error (e.g. thread root deleted), the +// helper falls through to the new-message endpoint so the user still receives +// the response even though thread placement is lost. +func TestDeliverMessage_ReplyFailure_FallsBackToSendMessage(t *testing.T) { + var replyHits, sendHits int + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == tokenEndpoint { + _, _ = w.Write([]byte(`{"code":0,"msg":"ok","tenant_access_token":"tok","expire":7200}`)) + return + } + w.Header().Set("Content-Type", "application/json") + if strings.Contains(r.URL.Path, "/reply") { + replyHits++ + // 230002 = message not found per research report. + _, _ = w.Write([]byte(`{"code":230002,"msg":"message not found","data":{}}`)) + return + } + sendHits++ + _, _ = w.Write([]byte(`{"code":0,"msg":"","data":{"message_id":"om_ok"}}`)) + })) + defer srv.Close() + + ch := &Channel{client: NewLarkClient("app", "secret", srv.URL)} + err := ch.deliverMessage(context.Background(), "oc_chat_1", "chat_id", "om_deleted_root", "text", `{"text":"hi"}`) + if err != nil { + t.Fatalf("deliverMessage should have succeeded via fallback, got: %v", err) + } + if replyHits != 1 { + t.Errorf("reply hits: got %d, want 1", replyHits) + } + if sendHits != 1 { + t.Errorf("send hits: got %d, want 1 (fallback)", sendHits) + } +} + +// TestParseMessageEvent_ThreadIDOnlyStampedForActualThreads guards H1: plain +// quote replies MUST NOT be stamped as thread replies. Only messages where +// Lark's event payload carries a non-empty `thread_id` should flow through +// the reply endpoint. This test asserts parseMessageEvent preserves the +// distinction between root_id (populated on any reply) and thread_id (only +// inside topic threads). +func TestParseMessageEvent_ThreadIDOnlyStampedForActualThreads(t *testing.T) { + ch := &Channel{} // botOpenID empty — treat mentions as bot for test simplicity + cases := []struct { + name string + rootID string + threadID string + wantThread string + wantRoot string + }{ + {"plain standalone message", "", "", "", ""}, + {"plain quote reply (root set, no thread)", "om_old_msg", "", "", "om_old_msg"}, + {"actual thread message", "om_thread_root", "om_thread_root", "om_thread_root", "om_thread_root"}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + ev := &MessageEvent{} + ev.Event.Message.MessageID = "om_current" + ev.Event.Message.ChatID = "oc_chat" + ev.Event.Message.ChatType = "group" + ev.Event.Message.MessageType = "text" + ev.Event.Message.Content = `{"text":"hi"}` + ev.Event.Message.RootID = tc.rootID + ev.Event.Message.ThreadID = tc.threadID + + mc := ch.parseMessageEvent(ev) + if mc == nil { + t.Fatal("parseMessageEvent returned nil") + } + if mc.ThreadID != tc.wantThread { + t.Errorf("ThreadID: got %q, want %q", mc.ThreadID, tc.wantThread) + } + if mc.RootID != tc.wantRoot { + t.Errorf("RootID: got %q, want %q", mc.RootID, tc.wantRoot) + } + }) + } +}