Cyrene_v2/plugins/channels/qq_obv11/client_retry_test.go

package qq_obv11

// 本文件测 **ws + client** 的「首连失败不致命 + 后台重试」行为：用可注入的
// DialContextFunc 构造确定性场景（不靠 sleep 抢时序），断言重试次数、connected 状态、
// 派发通路、Disconnect 可打断、日志可观测。server / http 两端的行为不在此改动。

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"math"
	"net"
	"net/http"
	"net/http/httptest"
	"runtime"
	"strings"
	"sync"
	"testing"
	"time"

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

// testReconnectPolicy 是毫秒级退避策略：测试不必真等秒级退避，抖动置零使断言确定。
func testReconnectPolicy() reconnectPolicy {
	return reconnectPolicy{
		baseDelay:    time.Millisecond,
		maxDelay:     5 * time.Millisecond,
		factor:       reconnectBackoffFactor,
		maxDoublings: reconnectMaxDoublings,
		jitterRatio:  0,
	}
}

// countingDialer 是可注入 DialContextFunc 的测试替身：先失败 failTimes 次；之后若给了
// gate 则等 gate 放行（或 ctx 取消），再把真实拨号委托给 delegate。它记录每次尝试，
// 因此「重试有没有发生 / 停了没有」都是可断言的确定性事实。
type countingDialer struct {
	mu        sync.Mutex
	attempts  int
	failTimes int
	gate      chan struct{}
	delegate  DialContextFunc
}

func (d *countingDialer) dial(ctx context.Context, network, address string) (net.Conn, error) {
	d.mu.Lock()
	d.attempts++
	n := d.attempts
	d.mu.Unlock()

	if n <= d.failTimes {
		return nil, errors.New("test: onebot peer not up yet")
	}
	if d.gate != nil {
		select {
		case <-d.gate:
		case <-ctx.Done():
			return nil, ctx.Err()
		}
	}
	if d.delegate == nil {
		return nil, errors.New("test: no delegate configured")
	}
	return d.delegate(ctx, network, address)
}

func (d *countingDialer) count() int {
	d.mu.Lock()
	defer d.mu.Unlock()
	return d.attempts
}

// testLogStore / testLogHandler 捕获 slog 记录，用于断言 WARN/INFO 的内容（来源、
// 重试间隔）以及 access_token 不落日志。
type testLogStore struct {
	mu      sync.Mutex
	records []testLogRecord
}

type testLogRecord struct {
	level slog.Level
	msg   string
	attrs map[string]any
}

func (s *testLogStore) add(r testLogRecord) {
	s.mu.Lock()
	s.records = append(s.records, r)
	s.mu.Unlock()
}

func (s *testLogStore) all() []testLogRecord {
	s.mu.Lock()
	defer s.mu.Unlock()
	return append([]testLogRecord(nil), s.records...)
}

func (s *testLogStore) byLevel(level slog.Level) []testLogRecord {
	var out []testLogRecord
	for _, r := range s.all() {
		if r.level == level {
			out = append(out, r)
		}
	}
	return out
}

type testLogHandler struct {
	store *testLogStore
	base  []slog.Attr
}

func (h *testLogHandler) Enabled(context.Context, slog.Level) bool { return true }

func (h *testLogHandler) Handle(_ context.Context, r slog.Record) error {
	rec := testLogRecord{level: r.Level, msg: r.Message, attrs: map[string]any{}}
	for _, a := range h.base {
		rec.attrs[a.Key] = a.Value.Any()
	}
	r.Attrs(func(a slog.Attr) bool {
		rec.attrs[a.Key] = a.Value.Any()
		return true
	})
	h.store.add(rec)
	return nil
}

func (h *testLogHandler) WithAttrs(attrs []slog.Attr) slog.Handler {
	return &testLogHandler{store: h.store, base: append(append([]slog.Attr{}, h.base...), attrs...)}
}

func (h *testLogHandler) WithGroup(string) slog.Handler { return h }

// TestClientConnectFirstFailureIsNotFatal 场景①：对面不在时 Connect 必须成功返回、
// 状态如实为「等待连接」、后台持续重试，且日志带来源与下次重试间隔。
func TestClientConnectFirstFailureIsNotFatal(t *testing.T) {
	t.Parallel()

	const token = "super-secret-token"
	dial := &countingDialer{failTimes: math.MaxInt}
	store := &testLogStore{}
	a := NewAdapter(Config{
		Mode:        modeClient,
		RemoteURL:   "ws://127.0.0.1:1/",
		AccessToken: token,
		DialContext: dial.dial,
		Logger:      slog.New(&testLogHandler{store: store}),
	})
	a.policy = testReconnectPolicy()

	start := time.Now()
	if err := a.Connect(context.Background()); err != nil {
		t.Fatalf("Connect() error = %v, want nil（首连失败必须非致命）", err)
	}
	t.Cleanup(func() { _ = a.Disconnect(context.Background()) })
	if elapsed := time.Since(start); elapsed > time.Second {
		t.Errorf("Connect() 耗时 %v，首连失败应立即返回，重试必须在后台", elapsed)
	}
	if a.IsConnected() {
		t.Error("IsConnected() = true，首连失败时必须是 false（状态不许撒谎）")
	}
	if err := a.HealthCheck(context.Background()); !errors.Is(err, adapter.ErrNotConnected) {
		t.Errorf("HealthCheck() = %v, want ErrNotConnected（未连上不得报健康）", err)
	}
	if !waitUntil(3*time.Second, func() bool { return dial.count() >= 4 }) {
		t.Fatalf("拨号次数 = %d, want >= 4（没有进入后台重试循环）", dial.count())
	}
	if a.IsConnected() {
		t.Error("IsConnected() = true，但每次拨号都失败")
	}

	warns := store.byLevel(slog.LevelWarn)
	if len(warns) == 0 {
		t.Fatal("没有任何 WARN 记录：首连失败必须可观测")
	}
	first := warns[0]
	if first.attrs["source"] != "plugins/channels/qq_obv11" {
		t.Errorf("WARN source = %v, want plugins/channels/qq_obv11", first.attrs["source"])
	}
	if first.attrs["err"] == nil {
		t.Errorf("WARN 缺少 err 属性: %+v", first.attrs)
	}
	// 首次失败 → delayFor(1) = base × factor；测试策略抖动为 0，取值确定。
	if got, ok := first.attrs["retryIn"].(time.Duration); !ok || got != 2*time.Millisecond {
		t.Errorf("WARN retryIn = %v, want 2ms（必须写明下次重试间隔）", first.attrs["retryIn"])
	}
	assertNoTokenLeak(t, store, token)
}

// assertNoTokenLeak 断言所有日志记录都不含 access_token 明文。
func assertNoTokenLeak(t *testing.T, store *testLogStore, token string) {
	t.Helper()

	for _, r := range store.all() {
		if strings.Contains(r.msg, token) {
			t.Errorf("日志泄露 access_token: %q", r.msg)
		}
		for k, v := range r.attrs {
			if strings.Contains(fmt.Sprint(v), token) {
				t.Errorf("日志属性 %s 泄露 access_token: %v", k, v)
			}
		}
	}
}

// TestClientConnectRetriesUntilPeerComesUp 场景②：前 N 次失败、第 N+1 次成功时自动
// 连上（connected=true），并且入站派发与出站回消息都照常工作。
func TestClientConnectRetriesUntilPeerComesUp(t *testing.T) {
	t.Parallel()

	srv, gotCh := newFakeOneBotServer(t)
	defer srv.Close()

	const failTimes = 3
	const token = "retry-secret-token"
	gate := make(chan struct{})
	dial := &countingDialer{failTimes: failTimes, gate: gate, delegate: (&net.Dialer{}).DialContext}
	store := &testLogStore{}
	a := NewAdapter(Config{
		Mode:        modeClient,
		RemoteURL:   "ws://" + strings.TrimPrefix(srv.URL, "http://"),
		AccessToken: token,
		DialContext: dial.dial,
		Logger:      slog.New(&testLogHandler{store: store}),
	})
	a.policy = testReconnectPolicy()
	msgCh := make(chan adapter.UnifiedMessage, 1)
	a.SetMessageHandler(func(ctx context.Context, m adapter.UnifiedMessage) { msgCh <- m })

	if err := a.Connect(context.Background()); err != nil {
		t.Fatalf("Connect() error = %v, want nil（对面没起来不算致命）", err)
	}
	t.Cleanup(func() { _ = a.Disconnect(context.Background()) })
	if a.IsConnected() {
		t.Error("IsConnected() = true，但前 3 次拨号必然失败")
	}
	// 第 4 次尝试会卡在 gate 上：等它出现，说明失败确实被重试了。
	if !waitUntil(3*time.Second, func() bool { return dial.count() >= failTimes+1 }) {
		t.Fatalf("拨号次数 = %d, want >= %d", dial.count(), failTimes+1)
	}
	close(gate) // 对面起来了

	if !waitUntil(3*time.Second, a.IsConnected) {
		t.Fatalf("第 %d 次拨号成功却没有自动连上（attempts=%d）", failTimes+1, dial.count())
	}
	if got := dial.count(); got != failTimes+1 {
		t.Errorf("拨号次数 = %d, want %d（失败都重试了，成功即不再拨）", got, failTimes+1)
	}
	if err := a.HealthCheck(context.Background()); err != nil {
		t.Errorf("HealthCheck() = %v, want nil（已连上）", err)
	}
	assertRetryObservability(t, store, failTimes, token)
	assertClientDelivery(t, a, msgCh, gotCh)
}

// newFakeOneBotServer 起一个假的 OneBot 实现：握手成功后推一条群消息，并把收到的出站
// 帧送到返回的 channel。
func newFakeOneBotServer(t *testing.T) (*httptest.Server, chan []byte) {
	t.Helper()

	gotCh := make(chan []byte, 4)
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		conn, err := upgradeWS(w, r)
		if err != nil {
			t.Errorf("fake onebot upgrade error: %v", err)
			return
		}
		defer conn.Close()
		evt := `{"post_type":"message","message_type":"group","time":1700000001,"self_id":100,"message_id":10,"user_id":42,"group_id":123,"message":[{"type":"text","data":{"text":"hi"}}],"sender":{"user_id":42,"nickname":"Bob","role":"member"}}`
		_ = conn.WriteMessage(opText, []byte(evt))
		for {
			_, payload, rerr := conn.ReadMessage()
			if rerr != nil {
				return
			}
			gotCh <- payload
		}
	}))
	return srv, gotCh
}

// assertRetryObservability 断言重试留痕：每次失败一条 WARN、连上一条 INFO "connected"，
// 且 access_token 不落日志。
func assertRetryObservability(t *testing.T, store *testLogStore, minWarns int, token string) {
	t.Helper()

	if got := len(store.byLevel(slog.LevelWarn)); got < minWarns {
		t.Errorf("WARN 记录 = %d, want >= %d（每次失败都要留痕）", got, minWarns)
	}
	sawConnected := false
	for _, r := range store.byLevel(slog.LevelInfo) {
		if strings.Contains(r.msg, "connected") {
			sawConnected = true
		}
	}
	if !sawConnected {
		t.Error("连上后没有 INFO connected 记录")
	}
	assertNoTokenLeak(t, store, token)
}

// assertClientDelivery 断言连上之后通道真的好用（重试成功 ≠ 收发可用）：入站派发与
// 出站 send_msg 各走一遍。
func assertClientDelivery(t *testing.T, a *Adapter, msgCh <-chan adapter.UnifiedMessage, gotCh <-chan []byte) {
	t.Helper()

	select {
	case m := <-msgCh:
		if m.ChannelType != adapter.ChannelGroup || m.GroupID != "123" || m.Content != "hi" {
			t.Errorf("连上后收到消息 = %+v, want group/123/hi", m)
		}
	case <-time.After(3 * time.Second):
		t.Fatal("重连成功后没有派发入站消息")
	}

	if err := a.SendMessage(context.Background(), adapter.UnifiedResponse{
		Messages: []adapter.UnifiedMessage{{ChannelType: adapter.ChannelGroup, GroupID: "123", Content: "pong"}},
	}); err != nil {
		t.Fatalf("SendMessage() error = %v", err)
	}
	select {
	case payload := <-gotCh:
		assertSendMsg(t, payload)
	case <-time.After(3 * time.Second):
		t.Fatal("重连成功后 fake OneBot 没收到 send_msg")
	}
}

// TestClientRetryStopsOnDisconnect 场景③：Disconnect 必须立即打断重试等待——不再有
// 新的拨号尝试，且不泄漏 goroutine。本用例断言进程 goroutine 数，故不并行。
func TestClientRetryStopsOnDisconnect(t *testing.T) {
	dial := &countingDialer{failTimes: math.MaxInt}
	a := NewAdapter(Config{
		Mode:        modeClient,
		RemoteURL:   "ws://127.0.0.1:1/",
		DialContext: dial.dial,
	})
	a.policy = testReconnectPolicy()

	base := runtime.NumGoroutine()
	if err := a.Connect(context.Background()); err != nil {
		t.Fatalf("Connect() error = %v", err)
	}
	if !waitUntil(3*time.Second, func() bool { return dial.count() >= 2 }) {
		t.Fatalf("拨号次数 = %d, want >= 2（没有进入重试循环）", dial.count())
	}
	if err := a.Disconnect(context.Background()); err != nil {
		t.Fatalf("Disconnect() error = %v", err)
	}
	stopped := dial.count()
	// 取消与拨号之间存在一个无法完全消除的竞态窗口：取消那一刻可能已有一次拨号
	// 「起跑」。所以先让它落地，再要求此后彻底安静——退避是毫秒级，重试若没停，
	// 这两个窗口里必然出现远超一次的拨号。
	time.Sleep(50 * time.Millisecond)
	settled := dial.count()
	if settled > stopped+1 {
		t.Fatalf("Disconnect 后仍在重试：拨号次数 %d → %d（还允许至多一次已起跑的拨号）", stopped, settled)
	}
	deadline := time.Now().Add(250 * time.Millisecond)
	for time.Now().Before(deadline) {
		if n := dial.count(); n != settled {
			t.Fatalf("Disconnect 后仍在重试：拨号次数 %d → %d", settled, n)
		}
		time.Sleep(5 * time.Millisecond)
	}
	if !waitUntil(3*time.Second, func() bool { return runtime.NumGoroutine() <= base }) {
		t.Errorf("goroutines = %d, want <= %d：重试循环泄漏了", runtime.NumGoroutine(), base)
	}
}

// TestClientConnectWithoutRemoteURLIsFatal 守住闸门口径（mode==client ⇒ remote_url
// 非空）：配置缺失不重试，直接失败——重试治不好配置错。
func TestClientConnectWithoutRemoteURLIsFatal(t *testing.T) {
	t.Parallel()

	dial := &countingDialer{failTimes: math.MaxInt}
	a := NewAdapter(Config{Mode: modeClient, DialContext: dial.dial})
	a.policy = testReconnectPolicy()

	if err := a.Connect(context.Background()); err == nil {
		t.Fatal("缺 remote_url 的 Connect() = nil, want error")
	}
	if n := dial.count(); n != 0 {
		t.Errorf("配置错也拨号了 %d 次，want 0", n)
	}
	if a.IsConnected() {
		t.Error("配置错的适配器报告已连接")
	}
}

// TestServerModeUnaffectedByClientRetry 守住「不动别人」：server 模式既不拨号、也不
// 因客户未连而伪造状态；缺 listen_addr 仍然致命。
func TestServerModeUnaffectedByClientRetry(t *testing.T) {
	t.Parallel()

	dial := &countingDialer{failTimes: math.MaxInt}
	a := NewAdapter(Config{
		Mode:        modeServer,
		ListenAddr:  "127.0.0.1:0",
		DialContext: dial.dial,
	})
	if err := a.Connect(context.Background()); err != nil {
		t.Fatalf("server Connect() error = %v", err)
	}
	defer func() { _ = a.Disconnect(context.Background()) }()
	if n := dial.count(); n != 0 {
		t.Errorf("server 模式拨号 %d 次，want 0", n)
	}
	if a.IsConnected() {
		t.Error("server 模式在无 OneBot 反连时报已连接")
	}

	client, err := dialWS(context.Background(), nil, "ws://"+a.Addr()+"/", "")
	if err != nil {
		t.Fatalf("dial server error: %v", err)
	}
	defer client.Close()
	if !waitUntil(3*time.Second, a.IsConnected) {
		t.Fatal("OneBot 反连后 server 模式仍未报告已连接")
	}

	missing := NewAdapter(Config{Mode: modeServer})
	if err := missing.Connect(context.Background()); err == nil {
		t.Error("缺 listen_addr 的 server Connect() = nil, want error")
	}
}