Files
larksuite-cli/shortcuts/im/convert_lib/reactions_test.go
sammi-bytedance 893555a1b1 perf(im): parallelize reactions, thread_replies, and merge_forward fetches (#1146)
Follow-up to #1095. The reactions auto-enrichment shipped, but on busy chats the strictly-serial per-resource fetches in EnrichReactions, ExpandThreadReplies, and merge_forward expansion stretched the command's wall time above 14s — enough that wrapper agents (30–60s wall-clock budgets) saw timeouts even though the CLI itself never errored. This PR parallelizes all three with the same bounded-concurrency pattern, batches the follow-up contact-API sender resolution so it doesn't fan back out into a serial stall, and fixes two correctness bugs that surfaced during review. Scoped to convert_lib/{reactions,thread,merge,content_convert}.go + tests + the 4 shortcut Execute hooks + the reference doc.

Change-Id: I0206d10ad204382170bd42aec67f82578923736e
2026-05-28 19:25:11 +08:00

411 lines
14 KiB
Go

// Copyright (c) 2026 Lark Technologies Pte. Ltd.
// SPDX-License-Identifier: MIT
package convertlib
import (
"encoding/json"
"fmt"
"io"
"net/http"
"reflect"
"sort"
"strings"
"sync"
"testing"
)
// TestEnrichReactions_Success exercises the basic happy path: messages that
// carry reactions get a "reactions" field, messages without reactions stay
// untouched.
func TestEnrichReactions_Success(t *testing.T) {
runtime := newBotConvertlibRuntime(t, convertlibRoundTripFunc(func(req *http.Request) (*http.Response, error) {
if !strings.Contains(req.URL.Path, "/open-apis/im/v1/messages/reactions/batch_query") {
return nil, fmt.Errorf("unexpected path: %s", req.URL.Path)
}
var payload map[string]interface{}
body, _ := io.ReadAll(req.Body)
_ = json.Unmarshal(body, &payload)
queries, _ := payload["queries"].([]interface{})
if len(queries) != 2 {
t.Fatalf("queries size = %d, want 2", len(queries))
}
return convertlibJSONResponse(200, map[string]interface{}{
"code": 0,
"data": map[string]interface{}{
"success_msg_reaction_counts": []interface{}{
map[string]interface{}{
"message_id": "om_a",
"reaction_count": []interface{}{
map[string]interface{}{"reaction_type": "SMILE", "count": 3},
},
},
},
"success_msg_reaction_details": []interface{}{
map[string]interface{}{
"message_id": "om_a",
"message_reaction_items": []interface{}{
map[string]interface{}{
"reaction_id": "react_1",
"emoji_type": "SMILE",
"operator": map[string]interface{}{"operator_id": "ou_x", "operator_type": "user"},
"action_time": "1710600000",
},
},
},
},
"fail_msg_reaction_details": []interface{}{},
},
}), nil
}))
messages := []map[string]interface{}{
{"message_id": "om_a"},
{"message_id": "om_b"},
}
EnrichReactions(runtime, messages)
reactionsA, ok := messages[0]["reactions"].(map[string]interface{})
if !ok {
t.Fatalf("message om_a missing reactions field: %#v", messages[0])
}
counts, _ := reactionsA["counts"].([]interface{})
if len(counts) != 1 {
t.Fatalf("om_a counts = %d, want 1", len(counts))
}
details, _ := reactionsA["details"].([]interface{})
if len(details) != 1 {
t.Fatalf("om_a details = %d, want 1", len(details))
}
if _, ok := messages[1]["reactions"]; ok {
t.Fatalf("message om_b should not have reactions field (none in response): %#v", messages[1])
}
}
// TestEnrichReactions_BatchSize splits queries into batches of 20 (server-side
// max for batch_query). Multi-batch dispatch is concurrent (bounded fan-out),
// so callers must tolerate any ordering of batch arrivals at the transport.
func TestEnrichReactions_BatchSize(t *testing.T) {
var mu sync.Mutex
var observedBatchSizes []int
runtime := newBotConvertlibRuntime(t, convertlibRoundTripFunc(func(req *http.Request) (*http.Response, error) {
body, _ := io.ReadAll(req.Body)
var payload map[string]interface{}
_ = json.Unmarshal(body, &payload)
queries, _ := payload["queries"].([]interface{})
mu.Lock()
observedBatchSizes = append(observedBatchSizes, len(queries))
mu.Unlock()
return convertlibJSONResponse(200, map[string]interface{}{
"code": 0,
"data": map[string]interface{}{},
}), nil
}))
messages := make([]map[string]interface{}, 25)
for i := range messages {
messages[i] = map[string]interface{}{"message_id": fmt.Sprintf("om_%02d", i)}
}
EnrichReactions(runtime, messages)
sort.Ints(observedBatchSizes)
if want := []int{5, 20}; !reflect.DeepEqual(observedBatchSizes, want) {
t.Fatalf("batch sizes (sorted) = %v, want %v", observedBatchSizes, want)
}
}
// TestEnrichReactions_MultiBatchCorrectness exercises the bounded-concurrency
// multi-batch path: every message across all batches must receive its own
// reactions block regardless of which goroutine the batch ran on. A race or a
// cross-batch index mix-up would manifest as missing or duplicated blocks.
func TestEnrichReactions_MultiBatchCorrectness(t *testing.T) {
var mu sync.Mutex
var batchCalls int
runtime := newBotConvertlibRuntime(t, convertlibRoundTripFunc(func(req *http.Request) (*http.Response, error) {
body, _ := io.ReadAll(req.Body)
var payload map[string]interface{}
_ = json.Unmarshal(body, &payload)
queries, _ := payload["queries"].([]interface{})
counts := make([]interface{}, 0, len(queries))
for _, q := range queries {
qm, _ := q.(map[string]interface{})
id, _ := qm["message_id"].(string)
counts = append(counts, map[string]interface{}{
"message_id": id,
"reaction_count": []interface{}{
map[string]interface{}{"reaction_type": "SMILE", "count": 1},
},
})
}
mu.Lock()
batchCalls++
mu.Unlock()
return convertlibJSONResponse(200, map[string]interface{}{
"code": 0,
"data": map[string]interface{}{
"success_msg_reaction_counts": counts,
},
}), nil
}))
// 65 messages -> 4 batches (20+20+20+5), enough to actually exercise the
// bounded fan-out (concurrency cap = 4) rather than degenerate to 1-2 calls.
messages := make([]map[string]interface{}, 65)
for i := range messages {
messages[i] = map[string]interface{}{"message_id": fmt.Sprintf("om_%03d", i)}
}
EnrichReactions(runtime, messages)
if batchCalls != 4 {
t.Fatalf("expected 4 batched calls, got %d", batchCalls)
}
for i, m := range messages {
if _, ok := m["reactions"]; !ok {
t.Fatalf("message %d (%s) missing reactions after multi-batch run", i, m["message_id"])
}
}
}
// TestEnrichReactions_APIFailure: when the API call fails, messages stay
// without a reactions field but get marked with reactions_error=true so
// downstream consumers can distinguish "fetch failed" from "no reactions".
// Mirrors the thread_replies_error pattern in thread.go.
func TestEnrichReactions_APIFailure(t *testing.T) {
runtime := newBotConvertlibRuntime(t, convertlibRoundTripFunc(func(req *http.Request) (*http.Response, error) {
return nil, fmt.Errorf("simulated network error")
}))
messages := []map[string]interface{}{
{"message_id": "om_a"},
{"message_id": "om_b"},
}
EnrichReactions(runtime, messages)
for _, m := range messages {
if _, ok := m["reactions"]; ok {
t.Fatalf("message %v should have no reactions after API failure", m["message_id"])
}
if v, _ := m["reactions_error"].(bool); !v {
t.Fatalf("message %v should have reactions_error=true after API failure, got = %#v",
m["message_id"], m["reactions_error"])
}
}
}
// TestEnrichReactions_PartialFailure: when batch_query returns a fail entry
// for one ID, that message gets reactions_error=true while the rest stay
// clean (no error flag) and keep their normal reactions block.
func TestEnrichReactions_PartialFailure(t *testing.T) {
runtime := newBotConvertlibRuntime(t, convertlibRoundTripFunc(func(req *http.Request) (*http.Response, error) {
return convertlibJSONResponse(200, map[string]interface{}{
"code": 0,
"data": map[string]interface{}{
"success_msg_reaction_counts": []interface{}{
map[string]interface{}{
"message_id": "om_ok",
"reaction_count": []interface{}{
map[string]interface{}{"reaction_type": "SMILE", "count": 1},
},
},
},
"fail_msg_reaction_details": []interface{}{
map[string]interface{}{"message_id": "om_bad"},
},
},
}), nil
}))
ok := map[string]interface{}{"message_id": "om_ok"}
bad := map[string]interface{}{"message_id": "om_bad"}
EnrichReactions(runtime, []map[string]interface{}{ok, bad})
if _, has := ok["reactions"]; !has {
t.Fatalf("om_ok should have reactions: %#v", ok)
}
if v, _ := ok["reactions_error"].(bool); v {
t.Fatalf("om_ok must not carry reactions_error: %#v", ok)
}
if _, has := bad["reactions"]; has {
t.Fatalf("om_bad should have no reactions block: %#v", bad)
}
if v, _ := bad["reactions_error"].(bool); !v {
t.Fatalf("om_bad should have reactions_error=true, got = %#v", bad["reactions_error"])
}
}
// TestEnrichReactions_EmptyMessages: no messages -> no API call at all.
func TestEnrichReactions_EmptyMessages(t *testing.T) {
called := false
runtime := newBotConvertlibRuntime(t, convertlibRoundTripFunc(func(req *http.Request) (*http.Response, error) {
called = true
return convertlibJSONResponse(200, map[string]interface{}{"code": 0, "data": map[string]interface{}{}}), nil
}))
EnrichReactions(runtime, nil)
EnrichReactions(runtime, []map[string]interface{}{})
if called {
t.Fatalf("API should not be called when messages list is empty")
}
}
// TestEnrichReactions_SkipsMessagesWithoutID: messages missing message_id
// (defensive) should not crash and not be sent in queries.
func TestEnrichReactions_SkipsMessagesWithoutID(t *testing.T) {
var sentIDs []string
runtime := newBotConvertlibRuntime(t, convertlibRoundTripFunc(func(req *http.Request) (*http.Response, error) {
body, _ := io.ReadAll(req.Body)
var payload map[string]interface{}
_ = json.Unmarshal(body, &payload)
queries, _ := payload["queries"].([]interface{})
for _, q := range queries {
qm, _ := q.(map[string]interface{})
id, _ := qm["message_id"].(string)
sentIDs = append(sentIDs, id)
}
return convertlibJSONResponse(200, map[string]interface{}{"code": 0, "data": map[string]interface{}{}}), nil
}))
messages := []map[string]interface{}{
{"message_id": "om_a"},
{}, // no message_id
{"message_id": ""},
{"message_id": "om_b"},
}
EnrichReactions(runtime, messages)
if want := []string{"om_a", "om_b"}; !reflect.DeepEqual(sentIDs, want) {
t.Fatalf("sent IDs = %v, want %v", sentIDs, want)
}
}
// TestEnrichReactions_WalksThreadReplies: thread_replies nested under a parent
// message must also be enriched, in the same batch_query call as the parent —
// otherwise the parent gets reactions but its replies don't, leaving the output
// inconsistent.
func TestEnrichReactions_WalksThreadReplies(t *testing.T) {
var observedQueriedIDs []string
var observedCallCount int
runtime := newBotConvertlibRuntime(t, convertlibRoundTripFunc(func(req *http.Request) (*http.Response, error) {
observedCallCount++
body, _ := io.ReadAll(req.Body)
var payload map[string]interface{}
_ = json.Unmarshal(body, &payload)
queries, _ := payload["queries"].([]interface{})
for _, q := range queries {
qm, _ := q.(map[string]interface{})
id, _ := qm["message_id"].(string)
observedQueriedIDs = append(observedQueriedIDs, id)
}
return convertlibJSONResponse(200, map[string]interface{}{
"code": 0,
"data": map[string]interface{}{
"success_msg_reaction_counts": []interface{}{
map[string]interface{}{
"message_id": "om_top",
"reaction_count": []interface{}{
map[string]interface{}{"reaction_type": "SMILE", "count": 1},
},
},
map[string]interface{}{
"message_id": "om_reply1",
"reaction_count": []interface{}{
map[string]interface{}{"reaction_type": "THUMBSUP", "count": 2},
},
},
map[string]interface{}{
"message_id": "om_reply2",
"reaction_count": []interface{}{
map[string]interface{}{"reaction_type": "HEART", "count": 3},
},
},
},
},
}), nil
}))
reply1 := map[string]interface{}{"message_id": "om_reply1"}
reply2 := map[string]interface{}{"message_id": "om_reply2"}
top := map[string]interface{}{
"message_id": "om_top",
"thread_replies": []map[string]interface{}{reply1, reply2},
}
messages := []map[string]interface{}{top}
EnrichReactions(runtime, messages)
if observedCallCount != 1 {
t.Fatalf("expected 1 batched API call, got %d", observedCallCount)
}
sort.Strings(observedQueriedIDs)
if want := []string{"om_reply1", "om_reply2", "om_top"}; !reflect.DeepEqual(observedQueriedIDs, want) {
t.Fatalf("queried IDs = %v, want %v (top + thread_replies)", observedQueriedIDs, want)
}
if _, ok := top["reactions"]; !ok {
t.Fatalf("top message missing reactions")
}
if _, ok := reply1["reactions"]; !ok {
t.Fatalf("reply1 missing reactions — thread_replies were not walked")
}
if _, ok := reply2["reactions"]; !ok {
t.Fatalf("reply2 missing reactions — thread_replies were not walked")
}
}
// TestEnrichReactions_DuplicateMessageID: when the caller passes two distinct
// message maps that share the same message_id (e.g. mget --message-ids om_a,om_a),
// both maps must receive the same reactions block, and the API must be queried
// for the id only once.
func TestEnrichReactions_DuplicateMessageID(t *testing.T) {
var observedQueriesPerCall []int
runtime := newBotConvertlibRuntime(t, convertlibRoundTripFunc(func(req *http.Request) (*http.Response, error) {
body, _ := io.ReadAll(req.Body)
var payload map[string]interface{}
_ = json.Unmarshal(body, &payload)
queries, _ := payload["queries"].([]interface{})
observedQueriesPerCall = append(observedQueriesPerCall, len(queries))
return convertlibJSONResponse(200, map[string]interface{}{
"code": 0,
"data": map[string]interface{}{
"success_msg_reaction_counts": []interface{}{
map[string]interface{}{
"message_id": "om_a",
"reaction_count": []interface{}{
map[string]interface{}{"reaction_type": "SMILE", "count": 2},
},
},
},
},
}), nil
}))
first := map[string]interface{}{"message_id": "om_a"}
second := map[string]interface{}{"message_id": "om_a"}
other := map[string]interface{}{"message_id": "om_b"}
messages := []map[string]interface{}{first, other, second}
EnrichReactions(runtime, messages)
if want := []int{2}; !reflect.DeepEqual(observedQueriesPerCall, want) {
t.Fatalf("queries-per-call = %v, want %v (each id once, no dup fetch)", observedQueriesPerCall, want)
}
firstReactions, firstOK := first["reactions"]
secondReactions, secondOK := second["reactions"]
if !firstOK {
t.Fatalf("first om_a entry missing reactions")
}
if !secondOK {
t.Fatalf("second om_a entry missing reactions — dup msg_id was dropped")
}
if !reflect.DeepEqual(firstReactions, secondReactions) {
t.Fatalf("dup entries reactions differ: %#v vs %#v", firstReactions, secondReactions)
}
}