Cyrene_v2/plugins/channels/qq_obv11/outbound_id_test.go

package qq_obv11

// 本文件验收 M12 的 A 层（出站消息 id 采集）三态：**带回 / 没带 / 没等到**。
// 逐条钉死两件事：① 取到就报出来（HTTP 从响应信封的 data.message_id，WS 从 echo 回执）；
// ② **取不到就是空**，绝不编造。另加一条「既有发送路径行为不变」的回归。

import (
	"context"
	"encoding/json"
	"net/http"
	"net/http/httptest"
	"strings"
	"sync"
	"testing"
	"time"

	"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter"
)

// fakeSendAPI 是只服务出站 action 的假 OneBot HTTP API：`data` 由用例给定（原样回），
// 用来驱动「回带 id / 不回带 id」两态。
type fakeSendAPI struct {
	data string
}

func (f *fakeSendAPI) handler() http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("Content-Type", "application/json")
		data := f.data
		if data == "" {
			data = "{}"
		}
		_, _ = w.Write([]byte(`{"status":"ok","retcode":0,"msg":"fake","data":` + data + `}`))
	})
}

// fakeOneBotWS 是本地假的 OneBot WS 服务端：记录收到的每一帧，并按 reply 决定是否回执。
// reply 返回 false 表示**不回**（「没等到」那一态就是这么造出来的）。
type fakeOneBotWS struct {
	mu       sync.Mutex
	received [][]byte
	reply    func(action, echo string, params json.RawMessage) ([]byte, bool)
}

func (f *fakeOneBotWS) count() int {
	f.mu.Lock()
	defer f.mu.Unlock()
	return len(f.received)
}

func (f *fakeOneBotWS) handler() http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		conn, err := upgradeWS(w, r)
		if err != nil {
			return
		}
		defer func() { _ = conn.Close() }()
		for {
			_, payload, err := conn.ReadMessage()
			if err != nil {
				return
			}
			f.mu.Lock()
			f.received = append(f.received, payload)
			replyFn := f.reply
			f.mu.Unlock()
			if replyFn == nil {
				continue
			}
			var req struct {
				Action string          `json:"action"`
				Echo   string          `json:"echo"`
				Params json.RawMessage `json:"params"`
			}
			if err := json.Unmarshal(payload, &req); err != nil {
				continue
			}
			frame, ok := replyFn(req.Action, req.Echo, req.Params)
			if !ok {
				continue
			}
			if err := conn.WriteMessage(opText, frame); err != nil {
				return
			}
		}
	})
}

// wsEchoFrame 构造一条 OneBot 动作回执帧：**空 post_type + echo**（就是 routeActionResponse
// 认的那种形态），data 原样拼进去。
func wsEchoFrame(echo, data string) []byte {
	return []byte(`{"status":"ok","retcode":0,"data":` + data + `,"echo":"` + echo + `"}`)
}

// echoOf 从请求体里取 echo，用来把回执发回给**发起它的那一次**发送。
func echoOf(t *testing.T, payload []byte) string {
	t.Helper()
	var req struct {
		Echo string `json:"echo"`
	}
	if err := json.Unmarshal(payload, &req); err != nil {
		t.Fatalf("decode outbound frame: %v", err)
	}
	if req.Echo == "" {
		t.Fatal("outbound send_msg must carry an echo for receipt correlation")
	}
	return req.Echo
}

// newWSReceiptAdapter 起一个 client 模式适配器，指向本地假 OneBot，并把等待回执的上限
// 收紧到 150ms（真实默认 5s 只覆盖异常，测试里不必真等）。
func newWSReceiptAdapter(t *testing.T, srv *httptest.Server) *Adapter {
	t.Helper()
	url := "ws://" + strings.TrimPrefix(srv.URL, "http://")
	a := NewAdapter(Config{Mode: modeClient, RemoteURL: url})
	a.sendReceiptTimeout = 150 * time.Millisecond
	t.Cleanup(func() { _ = a.Disconnect(context.Background()) })
	return a
}

func groupReply(text string) adapter.UnifiedResponse {
	return adapter.UnifiedResponse{Messages: []adapter.UnifiedMessage{
		{ChannelType: adapter.ChannelGroup, GroupID: "123", ContentType: adapter.ContentText, Content: text},
	}}
}

// TestHTTPOutboundIDReported：HTTP 分支从响应信封的 data.message_id 取到 id。
func TestHTTPOutboundIDReported(t *testing.T) {
	t.Parallel()

	srv := httptest.NewServer((&fakeSendAPI{data: `{"message_id":12345}`}).handler())
	defer srv.Close()

	a := NewAdapter(Config{Transport: transportHTTP, Mode: modeClient, APIBaseURL: srv.URL})
	if err := a.Connect(context.Background()); err != nil {
		t.Fatalf("Connect error: %v", err)
	}
	defer func() { _ = a.Disconnect(context.Background()) }()

	out, err := a.SendMessageWithReceipt(context.Background(), groupReply("pong"))
	if err != nil {
		t.Fatalf("SendMessageWithReceipt error: %v", err)
	}
	if len(out.Receipts) != 1 {
		t.Fatalf("Receipts len = %d, want 1", len(out.Receipts))
	}
	got := out.Receipts[0]
	if got.MessageID != "12345" || got.IDStatus != adapter.SendIDReported || !got.Reported() {
		t.Errorf("receipt = %+v, want reported 12345", got)
	}
}

// TestHTTPOutboundIDAbsent：回执到了但**没带** id ⇒ 空 id + Absent（可分辨，不编造）。
func TestHTTPOutboundIDAbsent(t *testing.T) {
	t.Parallel()

	srv := httptest.NewServer((&fakeSendAPI{data: `{}`}).handler())
	defer srv.Close()

	a := NewAdapter(Config{Transport: transportHTTP, Mode: modeClient, APIBaseURL: srv.URL})
	if err := a.Connect(context.Background()); err != nil {
		t.Fatalf("Connect error: %v", err)
	}
	defer func() { _ = a.Disconnect(context.Background()) }()

	out, err := a.SendMessageWithReceipt(context.Background(), groupReply("pong"))
	if err != nil {
		t.Fatalf("SendMessageWithReceipt error: %v", err)
	}
	got := out.Receipts[0]
	if got.MessageID != "" {
		t.Errorf("MessageID = %q, want empty (must never invent one)", got.MessageID)
	}
	if got.IDStatus != adapter.SendIDAbsent {
		t.Errorf("IDStatus = %q, want %q", got.IDStatus, adapter.SendIDAbsent)
	}
}

// TestHTTPOutboundIDAsString：实现对 message_id 回**字符串**时同样要取到（实现之间有出入）。
func TestHTTPOutboundIDAsString(t *testing.T) {
	t.Parallel()

	srv := httptest.NewServer((&fakeSendAPI{data: `{"message_id":"777888"}`}).handler())
	defer srv.Close()

	a := NewAdapter(Config{Transport: transportHTTP, Mode: modeClient, APIBaseURL: srv.URL})
	if err := a.Connect(context.Background()); err != nil {
		t.Fatalf("Connect error: %v", err)
	}
	defer func() { _ = a.Disconnect(context.Background()) }()

	out, err := a.SendMessageWithReceipt(context.Background(), groupReply("pong"))
	if err != nil {
		t.Fatalf("SendMessageWithReceipt error: %v", err)
	}
	if got := out.Receipts[0]; got.MessageID != "777888" || !got.Reported() {
		t.Errorf("receipt = %+v, want reported 777888", got)
	}
}

// TestWSOutboundIDReported：WS 分支登记 send_msg 的 echo 等待者，从回执里取出 id。
func TestWSOutboundIDReported(t *testing.T) {
	t.Parallel()

	fake := &fakeOneBotWS{reply: func(action, echo string, _ json.RawMessage) ([]byte, bool) {
		if action != "send_msg" {
			return nil, false
		}
		return wsEchoFrame(echo, `{"message_id":9001}`), true
	}}
	srv := httptest.NewServer(fake.handler())
	defer srv.Close()

	a := newWSReceiptAdapter(t, srv)
	if err := a.Connect(context.Background()); err != nil {
		t.Fatalf("Connect error: %v", err)
	}

	out, err := a.SendMessageWithReceipt(context.Background(), groupReply("pong"))
	if err != nil {
		t.Fatalf("SendMessageWithReceipt error: %v", err)
	}
	if got := out.Receipts[0]; got.MessageID != "9001" || !got.Reported() {
		t.Errorf("receipt = %+v, want reported 9001", got)
	}
}

// TestWSOutboundIDAbsent：回执到了但没带 id ⇒ 空 id + Absent。
func TestWSOutboundIDAbsent(t *testing.T) {
	t.Parallel()

	fake := &fakeOneBotWS{reply: func(action, echo string, _ json.RawMessage) ([]byte, bool) {
		if action != "send_msg" {
			return nil, false
		}
		return wsEchoFrame(echo, `{}`), true
	}}
	srv := httptest.NewServer(fake.handler())
	defer srv.Close()

	a := newWSReceiptAdapter(t, srv)
	if err := a.Connect(context.Background()); err != nil {
		t.Fatalf("Connect error: %v", err)
	}

	out, err := a.SendMessageWithReceipt(context.Background(), groupReply("pong"))
	if err != nil {
		t.Fatalf("SendMessageWithReceipt error: %v", err)
	}
	got := out.Receipts[0]
	if got.MessageID != "" {
		t.Errorf("MessageID = %q, want empty (must never invent one)", got.MessageID)
	}
	if got.IDStatus != adapter.SendIDAbsent {
		t.Errorf("IDStatus = %q, want %q", got.IDStatus, adapter.SendIDAbsent)
	}
}

// TestWSOutboundIDTimeoutUnknown：对端**根本不回**回执 ⇒ 超时收敛成「不知道」，
// 而且帧确实写出去了（发送成功与「拿不到 id」是两件事）。
func TestWSOutboundIDTimeoutUnknown(t *testing.T) {
	t.Parallel()

	fake := &fakeOneBotWS{} // reply 为 nil：永远不回
	srv := httptest.NewServer(fake.handler())
	defer srv.Close()

	a := newWSReceiptAdapter(t, srv)
	if err := a.Connect(context.Background()); err != nil {
		t.Fatalf("Connect error: %v", err)
	}

	start := time.Now()
	out, err := a.SendMessageWithReceipt(context.Background(), groupReply("pong"))
	elapsed := time.Since(start)
	if err != nil {
		t.Fatalf("SendMessageWithReceipt error: %v (写完帧不等于投递失败)", err)
	}
	if got := out.Receipts[0]; got.MessageID != "" || got.IDStatus != adapter.SendIDUnreported {
		t.Errorf("receipt = %+v, want empty id with unreported status", got)
	}
	if elapsed < a.sendReceiptTimeout {
		t.Errorf("returned after %v, want at least the %v receipt timeout", elapsed, a.sendReceiptTimeout)
	}
	if fake.count() != 1 {
		t.Errorf("fake OneBot received %d frames, want 1 (frame must be written regardless)", fake.count())
	}
}

// TestSendMessageDoesNotAwaitEcho 是**既有发送路径行为不变**的回归：SendMessage 不等
// 回执（对端不回也立刻返回），帧照旧写出去。等待只发生在显式的凭据路径上。
func TestSendMessageDoesNotAwaitEcho(t *testing.T) {
	t.Parallel()

	fake := &fakeOneBotWS{}
	srv := httptest.NewServer(fake.handler())
	defer srv.Close()

	a := newWSReceiptAdapter(t, srv)
	// 故意把等待上限拉到 3s：若 SendMessage 误走等待路径，这条用例会明显变慢。
	a.sendReceiptTimeout = 3 * time.Second
	if err := a.Connect(context.Background()); err != nil {
		t.Fatalf("Connect error: %v", err)
	}

	start := time.Now()
	if err := a.SendMessage(context.Background(), groupReply("pong")); err != nil {
		t.Fatalf("SendMessage error: %v", err)
	}
	if elapsed := time.Since(start); elapsed > time.Second {
		t.Errorf("SendMessage took %v, want no receipt wait on the legacy path", elapsed)
	}
	if !waitUntil(2*time.Second, func() bool { return fake.count() == 1 }) {
		t.Errorf("fake OneBot received %d frames, want the send_msg frame", fake.count())
	}
}

// TestWSSendReceiptsCorrelateByEcho：并发的多次发送必须各拿各的 id（echo 关联键唯一，
// **不靠时钟**）。假 OneBot 故意**乱序**回执：后发的先回。
func TestWSSendReceiptsCorrelateByEcho(t *testing.T) {
	t.Parallel()

	contents := []string{"m1", "m2", "m3"}
	delay := map[string]time.Duration{"m1": 120 * time.Millisecond, "m2": 60 * time.Millisecond, "m3": 0}
	fake := &fakeOneBotWS{reply: func(action, echo string, params json.RawMessage) ([]byte, bool) {
		if action != "send_msg" {
			return nil, false
		}
		var p struct {
			Message []struct {
				Data struct {
					Text string `json:"text"`
				} `json:"data"`
			} `json:"message"`
		}
		if err := json.Unmarshal(params, &p); err != nil || len(p.Message) == 0 {
			return nil, false
		}
		text := p.Message[0].Data.Text
		time.Sleep(delay[text])
		return wsEchoFrame(echo, `{"message_id":`+strings.TrimPrefix(text, "m")+`}`), true
	}}
	srv := httptest.NewServer(fake.handler())
	defer srv.Close()

	a := newWSReceiptAdapter(t, srv)
	a.sendReceiptTimeout = 2 * time.Second
	if err := a.Connect(context.Background()); err != nil {
		t.Fatalf("Connect error: %v", err)
	}

	var wg sync.WaitGroup
	got := make(map[string]string, len(contents))
	var mu sync.Mutex
	for _, text := range contents {
		wg.Add(1)
		go func(text string) {
			defer wg.Done()
			out, err := a.SendMessageWithReceipt(context.Background(), groupReply(text))
			if err != nil || len(out.Receipts) != 1 {
				t.Errorf("send %q: out=%+v err=%v", text, out, err)
				return
			}
			mu.Lock()
			got[text] = out.Receipts[0].MessageID
			mu.Unlock()
		}(text)
	}
	wg.Wait()

	for _, text := range contents {
		want := strings.TrimPrefix(text, "m")
		if got[text] != want {
			t.Errorf("send %q got id %q, want %q (echo correlation broken)", text, got[text], want)
		}
	}
}

// TestPlatformMessageIDForms 钉住 id 的**形态判据**：只有纯十进制数字串能当 id，其余
// 一律空（= 平台没给可用 id），绝不把原始 JSON 片段当 id 用。
func TestPlatformMessageIDForms(t *testing.T) {
	t.Parallel()

	cases := []struct {
		name string
		raw  string
		want string
	}{
		{"整数", `12345`, "12345"},
		{"数字字符串", `"12345"`, "12345"},
		{"带前导零的数字串", `"0012"`, "0012"},
		{"null", `null`, ""},
		{"缺失字段", ``, ""},
		{"空字符串", `""`, ""},
		{"浮点", `12.5`, ""},
		{"负数", `-1`, ""},
		{"对象", `{"a":1}`, ""},
		{"数组", `[1]`, ""},
		{"夹带字母", `"12ab"`, ""},
		{"夹带空格", `" 12"`, ""},
		{"超长", `"` + strings.Repeat("9", maxPlatformIDLen+1) + `"`, ""},
	}
	for _, tc := range cases {
		if got := platformMessageID(json.RawMessage(tc.raw)); got != tc.want {
			t.Errorf("platformMessageID(%s) = %q, want %q", tc.name, got, tc.want)
		}
	}
}

// TestSendReceiptForensicsLog：出站 id 的采集必须在 **DEBUG 里可取证** —— 「生产里平台
// 到底带不带 id」是下一片（M13 的环形表）能否成立的前提。日志只许记状态与 id，绝不记正文。
func TestSendReceiptForensicsLog(t *testing.T) {
	t.Parallel()

	buf, log := logCapture()
	srv := httptest.NewServer((&fakeSendAPI{data: `{"message_id":4242}`}).handler())
	defer srv.Close()

	a := NewAdapter(Config{Transport: transportHTTP, Mode: modeClient, APIBaseURL: srv.URL, Logger: log})
	if err := a.Connect(context.Background()); err != nil {
		t.Fatalf("Connect error: %v", err)
	}
	defer func() { _ = a.Disconnect(context.Background()) }()

	if _, err := a.SendMessageWithReceipt(context.Background(), groupReply("pong")); err != nil {
		t.Fatalf("SendMessageWithReceipt error: %v", err)
	}

	line := buf.String()
	if !strings.Contains(line, `msg="send_msg receipt captured"`) {
		t.Errorf("missing send receipt forensics line; log = %s", line)
	}
	if !strings.Contains(line, "message_id=4242") {
		t.Errorf("forensics line must carry the captured id; log = %s", line)
	}
	if strings.Contains(line, "pong") {
		t.Errorf("outbound text must never reach the log; log = %s", line)
	}
}

// TestMessageSentEchoCarriesPlatformID 证明**适配器侧**已经把 message_sent 回声的平台
// message_id 投影进了结构（SystemNotice.RefMessage）—— 也就是「她自己那条消息的 id」
// 在归一化之后并未丢失，丢它的是下游的档位判定（见交付说明）。
func TestMessageSentEchoCarriesPlatformID(t *testing.T) {
	t.Parallel()

	a := NewAdapter(Config{})
	notice, ok := a.NormalizeSystemNotice(adapter.UnifiedMessage{RawData: []byte(messageSentPrivate)})
	if !ok {
		t.Fatal("message_sent must normalize into a system notice (nil ok=false)")
	}
	if notice.Category != adapter.NoticeCategoryMessageSent {
		t.Fatalf("category = %q, want %q", notice.Category, adapter.NoticeCategoryMessageSent)
	}
	if notice.RefMessage != "77" {
		t.Errorf("RefMessage = %q, want 77 (platform message_id of her own echo)", notice.RefMessage)
	}
}