项目文件夹

文件
wehub-resource-sync bf9395e022
CI / license-header (push) Has been skipped
CI / e2e-dry-run (push) Has been skipped
CI / fast-gate (push) Failing after 0s
Test PR Label Logic / test-pr-labels (push) Failing after 1s
Skill Format Check / check-format (push) Failing after 2s
CI / security (push) Failing after 5s
CI / unit-test (push) Has been skipped
CI / lint (push) Has been skipped
CI / script-test (push) Has been skipped
CI / deterministic-gate (push) Has been skipped
CI / coverage (push) Has been skipped
CI / results (push) Has been cancelled
CI / deadcode (push) Has been cancelled
CI / e2e-live (push) Has been cancelled
chore: import upstream snapshot with attribution
2026-07-13 12:22:54 +08:00

363 行
14 KiB
Go

此文件含有模棱两可的 Unicode 字符
此文件含有可能会与其他字符混淆的 Unicode 字符。 如果您是想特意这样的,可以安全地忽略该警告。 使用 Escape 按钮显示他们。
// Copyright (c) 2026 Lark Technologies Pte. Ltd.
// SPDX-License-Identifier: MIT
package convertlib
import (
"fmt"
"net/http"
"sort"
"strings"
"sync"
"time"
"github.com/larksuite/cli/internal/validate"
"github.com/larksuite/cli/shortcuts/common"
larkcore "github.com/larksuite/oapi-sdk-go/v3/core"
)
// mergeForwardPrefetchConcurrency caps in-flight merge_forward sub-message
// fetches when a shortcut pre-scans a page for merge_forward messages and
// prefetches their children concurrently. Each call is one ~700ms-1s
// GET /open-apis/im/v1/messages/{id} per merge_forward — strictly serial in
// FormatMessageItem before this change, which turned page-size 50 + 5
// merge_forward messages into ~8.5s of stall (measured on a real chat).
// GET /open-apis/im/v1/messages/{id} has no published per-app rate-limit at
// these levels, so we set this higher than the reactions batch_query cap
// (which sits at 4 to stay well under the gateway-layer 50/s + 1000/min
// explicit ceiling on the reactions endpoint).
const mergeForwardPrefetchConcurrency = 8
type mergeForwardConverter struct{}
// Convert expands merge_forward sub-messages into a tree when runtime is
// available (or a pre-fetched cache was supplied), otherwise falls back to a
// summary string.
//
// When ctx.MergeForwardSubItems is non-nil (set by callers that pre-fetched
// the page's merge_forward children concurrently via
// PrefetchMergeForwardSubItems), Convert uses the cached items and skips the
// HTTP fetch entirely — this is how the shortcut layer turns N serial
// per-merge_forward GETs into one bounded-concurrency fan-out before the
// FormatMessageItem loop runs.
func (mergeForwardConverter) Convert(ctx *ConvertContext) string {
// Fast path: caller pre-fetched this merge_forward's sub-tree.
if ctx.MergeForwardSubItems != nil && ctx.MessageID != "" {
if cached, ok := ctx.MergeForwardSubItems[ctx.MessageID]; ok {
return renderMergeForwardTree(ctx, cached)
}
}
// Slow path: no pre-fetch; fall back to a per-merge_forward GET. Kept so
// callers that don't pre-fetch (e.g. event subscribers, ad-hoc Convert
// invocations in tests) still produce correct output, just serially.
// merge_forward body.content is typically a plain-text placeholder, not
// JSON with create_message_ids, so we must rely on the API to get actual
// sub-messages.
if ctx.Runtime != nil && ctx.MessageID != "" {
subItems, err := fetchMergeForwardSubMessages(ctx.MessageID, ctx.Runtime)
if err != nil {
return fmt.Sprintf("[Merged forward: fetch failed: %s]", err)
}
if len(subItems) > 0 {
return renderMergeForwardTree(ctx, subItems)
}
}
// Final fallback: try to extract message IDs from content (some older formats include them)
ids := ParseMergeForwardIDs(ctx.RawContent)
if len(ids) > 0 {
return fmt.Sprintf("[Merged forward: %d messages]", len(ids))
}
return "[Merged forward]"
}
// renderMergeForwardTree resolves sender names for the supplied sub-items and
// produces the formatted forwarded-messages tree. Shared by the prefetch fast
// path and the inline fetch fallback so both produce identical output.
func renderMergeForwardTree(ctx *ConvertContext, subItems []map[string]interface{}) string {
nameMap := ResolveSenderNames(ctx.Runtime, subItems, ctx.SenderNames)
AttachSenderNames(subItems, nameMap)
childrenMap := BuildMergeForwardChildrenMap(subItems, ctx.MessageID)
return FormatMergeForwardSubTree(ctx.MessageID, childrenMap)
}
// PrefetchMergeForwardSubItems scans rawItems for merge_forward messages,
// concurrently fetches each one's flat sub-message list, and returns a map
// keyed by the merge_forward message_id. Callers thread the returned map
// through FormatMessageItemWithMergePrefetch (or directly into a
// ConvertContext.MergeForwardSubItems) so the per-item conversion loop can
// reuse cached sub-trees instead of issuing its own serial GET.
//
// Each fetch is independent (different message_id, different sub-tree), so
// concurrent goroutines never contend on shared mutable state — the result
// map is written under a mutex purely to make the map safe for concurrent
// inserts.
//
// On fetch failure: emit a stderr warning and intentionally do NOT insert
// the failed id into the result map. The downstream
// mergeForwardConverter.Convert path keys off "is this id present in the
// prefetch?" — by leaving the key absent on failure, Convert falls through
// to its inline-fetch slow path, which (a) gets a second attempt at the
// GET, and (b) if that ALSO fails, surfaces the real "[Merged forward:
// fetch failed: ...]" string the user used to see in stdout. Inserting nil
// would have silently produced an empty <forwarded_messages> tree instead,
// dropping the failure signal from the user-visible output.
//
// When nameCache is non-nil, this function also runs one batched
// ResolveSenderNames across every sub-item it fetched, populating the cache
// before returning. Without this step, each per-merge_forward render in the
// caller's loop would issue its own contact API request for any uncached
// sender, re-introducing an N × ~400ms serial stall (measured at 5
// merge_forwards × ~400ms = ~2s in production traces). Pre-populating the
// cache makes those per-render ResolveSenderNames calls effective no-ops.
func PrefetchMergeForwardSubItems(runtime *common.RuntimeContext, rawItems []interface{}, nameCache map[string]string) map[string][]map[string]interface{} {
if runtime == nil || len(rawItems) == 0 {
return nil
}
var ids []string
for _, item := range rawItems {
m, _ := item.(map[string]interface{})
if m == nil {
continue
}
if mt, _ := m["msg_type"].(string); mt != "merge_forward" {
continue
}
id, _ := m["message_id"].(string)
if id == "" {
continue
}
ids = append(ids, id)
}
if len(ids) == 0 {
return nil
}
result := make(map[string][]map[string]interface{}, len(ids))
if len(ids) == 1 {
// Single-message fast path: no goroutine overhead. Matches the
// pre-existing serial behavior bit-for-bit when only one
// merge_forward is present.
items, err := fetchMergeForwardSubMessages(ids[0], runtime)
if err != nil {
fmt.Fprintf(runtime.IO().ErrOut, "warning: merge_forward_prefetch_failed: %s: %v\n", ids[0], err)
// Leave the key absent so Convert falls back to its inline GET
// path and surfaces "[Merged forward: fetch failed: ...]" if
// the retry also fails. See function godoc.
} else {
result[ids[0]] = items
}
batchResolveMergeForwardSenders(runtime, result, nameCache)
return result
}
var mu sync.Mutex
sem := make(chan struct{}, mergeForwardPrefetchConcurrency)
var wg sync.WaitGroup
for _, id := range ids {
// Add before the semaphore acquire — sync.WaitGroup godoc
// recommends Add precede the goroutine-spawning event.
wg.Add(1)
sem <- struct{}{}
go func() {
defer wg.Done()
defer func() { <-sem }()
items, err := fetchMergeForwardSubMessages(id, runtime)
mu.Lock()
if err != nil {
fmt.Fprintf(runtime.IO().ErrOut, "warning: merge_forward_prefetch_failed: %s: %v\n", id, err)
// Leave the key absent — see fast-path comment above.
} else {
result[id] = items
}
mu.Unlock()
}()
}
wg.Wait()
batchResolveMergeForwardSenders(runtime, result, nameCache)
return result
}
// batchResolveMergeForwardSenders gathers every sub-item across every
// prefetched merge_forward and runs a single ResolveSenderNames call against
// nameCache. No-op when nameCache is nil (callers that pre-fetched without
// caring about sender resolution, e.g. event subscribers that render on the
// fly) or when nothing was fetched.
func batchResolveMergeForwardSenders(runtime *common.RuntimeContext, prefetch map[string][]map[string]interface{}, nameCache map[string]string) {
if nameCache == nil || len(prefetch) == 0 {
return
}
var allSubItems []map[string]interface{}
for _, items := range prefetch {
allSubItems = append(allSubItems, items...)
}
if len(allSubItems) == 0 {
return
}
ResolveSenderNames(runtime, allSubItems, nameCache)
}
// fetchMergeForwardSubMessages fetches all sub-messages in a merge_forward
// container via a single API call. Returns a flat list of raw message items
// with upper_message_id for tree reconstruction.
//
// Uses DoAPIJSONTyped so the response envelope's code/msg are checked and surfaced
// — earlier this used the low-level DoAPI and reported every non-zero code
// as a generic "empty data" error, hiding the real failure (e.g. a server
// "code: 2200 Internal Error" with its log_id would show up as just "empty
// data" in the output).
func fetchMergeForwardSubMessages(messageID string, runtime *common.RuntimeContext) ([]map[string]interface{}, error) {
data, err := runtime.DoAPIJSONTyped(http.MethodGet, mergeForwardMessagesPath(messageID), larkcore.QueryParams{
"user_id_type": []string{"open_id"},
"card_msg_content_type": []string{"raw_card_content"},
}, nil)
if err != nil {
return nil, err
}
// DoAPIJSONTyped returns the envelope's `data` field; when the server's JSON
// has `code: 0` but omits `data` entirely, that field comes back as nil.
// Reading from a nil map in Go is safe (returns the zero value, never
// panics), but guarding explicitly makes the "successful empty
// response" path obvious and keeps a future signature change from
// silently introducing nil-deref hazards.
if data == nil {
return []map[string]interface{}{}, nil
}
rawItems, _ := data["items"].([]interface{})
items := make([]map[string]interface{}, 0, len(rawItems))
for _, raw := range rawItems {
if m, ok := raw.(map[string]interface{}); ok {
items = append(items, m)
}
}
return items, nil
}
func mergeForwardMessagesPath(messageID string) string {
return fmt.Sprintf("/open-apis/im/v1/messages/%s", validate.EncodePathSegment(messageID))
}
// ParseMergeForwardIDs extracts message IDs from a merge_forward content JSON.
func ParseMergeForwardIDs(raw string) []string {
parsed, err := ParseJSONObject(raw)
if err != nil {
return nil
}
rawIds, _ := parsed["create_message_ids"].([]interface{})
ids := make([]string, 0, len(rawIds))
for _, id := range rawIds {
if s, ok := id.(string); ok {
ids = append(ids, s)
}
}
return ids
}
// BuildMergeForwardChildrenMap builds a parent→children map from a flat items list.
// Items without upper_message_id are treated as direct children of rootMessageID.
// The root container message itself is skipped.
func BuildMergeForwardChildrenMap(items []map[string]interface{}, rootMessageID string) map[string][]map[string]interface{} {
result := make(map[string][]map[string]interface{})
for _, item := range items {
msgID, _ := item["message_id"].(string)
upperID, _ := item["upper_message_id"].(string)
// Skip the root container itself
if msgID == rootMessageID && upperID == "" {
continue
}
parentID := upperID
if parentID == "" {
parentID = rootMessageID
}
result[parentID] = append(result[parentID], item)
}
// Sort each group by create_time ascending
for _, children := range result {
sort.Slice(children, func(i, j int) bool {
return mergeForwardItemTimestamp(children[i]) < mergeForwardItemTimestamp(children[j])
})
}
return result
}
// FormatMergeForwardSubTree recursively formats a sub-tree rooted at parentID.
// For merge_forward children it recurses via the tree (no extra API calls).
// For other types it delegates to the provided convert callback.
func FormatMergeForwardSubTree(parentID string, childrenMap map[string][]map[string]interface{}) string {
children := childrenMap[parentID]
if len(children) == 0 {
return "<forwarded_messages/>"
}
var parts []string
for _, item := range children {
msgType, _ := item["msg_type"].(string)
if msgType == "" {
msgType = "text"
}
senderID := "unknown"
if senderMap, ok := item["sender"].(map[string]interface{}); ok {
if name, _ := senderMap["name"].(string); name != "" {
senderID = name
} else if id, _ := senderMap["id"].(string); id != "" {
senderID = id
}
}
tsStr, _ := item["create_time"].(string)
timestamp := FormatMergeForwardTimestamp(tsStr)
var content string
msgID, _ := item["message_id"].(string)
if msgType == "merge_forward" && msgID != "" {
content = FormatMergeForwardSubTree(msgID, childrenMap)
} else {
rawContent := ""
if body, ok := item["body"].(map[string]interface{}); ok {
rawContent, _ = body["content"].(string)
}
mentions, _ := item["mentions"].([]interface{})
content = ConvertBodyContent(msgType, &ConvertContext{
RawContent: rawContent,
MentionMap: BuildMentionKeyMap(mentions),
Mentions: mentions,
})
}
parts = append(parts, fmt.Sprintf("[%s] %s:\n%s", timestamp, senderID, IndentLines(content, " ")))
}
if len(parts) == 0 {
return "<forwarded_messages/>"
}
return "<forwarded_messages>\n" + strings.Join(parts, "\n") + "\n</forwarded_messages>"
}
// FormatMergeForwardTimestamp formats a millisecond timestamp string to local RFC3339 with offset.
func FormatMergeForwardTimestamp(tsStr string) string {
var ms int64
fmt.Sscanf(tsStr, "%d", &ms)
if ms == 0 {
return "unknown"
}
t := time.Unix(ms/1000, (ms%1000)*int64(time.Millisecond)).In(time.Local)
return t.Format(time.RFC3339)
}
// IndentLines prefixes every line of text with the given indent string.
func IndentLines(text, indent string) string {
lines := strings.Split(text, "\n")
for i, line := range lines {
lines[i] = indent + line
}
return strings.Join(lines, "\n")
}
// mergeForwardItemTimestamp returns the create_time as int64 milliseconds.
func mergeForwardItemTimestamp(item map[string]interface{}) int64 {
ts, _ := item["create_time"].(string)
var n int64
fmt.Sscanf(ts, "%d", &n)
return n
}