Cyrene_v2/internal/services/channelbridge/coalesce_fixture_test.go

package channelbridge

// coalesce_fixture_test.go 是 M23（静默窗口合并 + 回合中续跑）全部用例共用的夹具：
//
//   - **注入时钟**（laneClock）：窗口与陈旧度共用同一个时间源（Options.Now），
//     因此「2.0s 静默」「6s 绝对上限」「30s 超龄」都能被确定性推进；
//   - **注入定时器工厂**（coalesceTimers）：它既让窗口等待可以被手动触发，
//     **又是「主人私聊从未建窗口」的证据面** —— calls 必须为 0；
//   - **群消息夹具**：非主人、带提及（⇒ 必然进回合）、同一群共用一个隔离会话。
//
// 夹具全部落在 *_test.go 里，不进生产二进制。

import (
	"context"
	"sync"
	"testing"
	"time"

	"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/session"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/config/runtimeconfig"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/user"
)

// coalesceOffFor 是「把静默窗口关掉」的注入面：lane_test.go 的 M22 语义用例用它，
// 使那些用例仍以毫秒级钉住车道本身（顺序/串行/背压/陈旧度）。
func coalesceOffFor(string) *runtimeconfig.CoalescePolicy {
	p := runtimeconfig.DefaultCoalescePolicy()
	p.Enabled = false
	return &p
}

// coalesceOn 返回一组「群聊与陌生人私聊都开窗口」的生效参数。
func coalesceOn(windowMs, maxWaitMs, maxBatch int) runtimeconfig.CoalescePolicy {
	p := runtimeconfig.DefaultCoalescePolicy()
	p.Enabled = true
	p.WindowMs = windowMs
	p.MaxWaitMs = maxWaitMs
	p.MaxBatchMessages = maxBatch
	p.ApplyToStrangers = true
	p.StrangerWindowMs = windowMs
	return p
}

// coalesceFor 把一组固定参数包成注入面（同一实例可被多条车道共用）。
func coalesceFor(p runtimeconfig.CoalescePolicy) func(string) *runtimeconfig.CoalescePolicy {
	return func(string) *runtimeconfig.CoalescePolicy {
		c := p
		return &c
	}
}

// newCoalesceClock 返回一个**本地正午**的注入时钟。
//
// 起点刻意避开免打扰时段（23-07）：本文件测的是窗口合并，而凌晨的注入时钟会让
// 非主人群消息先在「免打扰」那一档被静默掉 —— 那样每条用例都变成了空断言
// （测试要观察的只是「什么时候发车」，不是「说不说」）。
func newCoalesceClock() *laneClock {
	c := &laneClock{}
	c.t = time.Date(2026, 5, 1, 12, 0, 0, 0, time.Local)
	return c
}

// coalesceTimers 是注入的定时器工厂：记录每次请求的时长，交回一个由测试手动触发的
// 通道（带缓冲，测试的发送永远不会阻塞）。
//
// 🔴 它同时是红线用例的证据面：主人私聊**一次都不该**调用它
// （TestOwnerPrivateNeverCreatesWindow 断言 calls==0）。
//
// **触发语义按「截止时刻」而不是「先进先出队列」**：每次请求记下「注册时的注入时钟 + 时长」，
// 测试推钟时**只把时钟推进到当前最早的截止时刻**，然后触发到期的那一个。
// 为什么不能按队列逐个推（旧写法）：`coalesceWait` 每个循环迭代都会重新 `after(wait)`，
// 于是每条新到的消息都会留下一个**再没人监听**的旧定时器；旧写法把它们也逐个触发，
// 每次照样推一个「请求时长」——实测 12 条消息内攒到 5 个，时钟被推 +10s（窗口 2s、
// 绝对上限 6s）⇒ 时钟越过绝对上限 ⇒ 循环头在**没有 takeAll** 的分支上直接发车，
// 队列尾部留在下一批（「12 条 → 2 轮」）。按截止时刻推进让「推了多少时间」= 「真的过了
// 多少时间」，旧定时器过期后被触发也不会把时钟推过头。
type coalesceTimers struct {
	mu      sync.Mutex
	calls   []time.Duration
	pending []coalesceTimer
	now     func() time.Time
	// depth 返回「车道上还没被 worker 取走的条目数」（含 ch/spill/front）。
	// 推钟前必须等它归零，见 pump 的说明。
	depth func() int
	// seq 是单调请求序号：worker 每次重新挂定时器都会 +1，因此「比某个序号更大的请求」
	// 才可能是它此刻**真正在等**的那一个（旧的都已被它放弃）。
	seq int64
}

// coalesceTimer 是一次定时器请求：通道 + 请求时长 + **截止时刻**（注册时的注入时钟 + d）。
type coalesceTimer struct {
	ch  chan time.Time
	d   time.Duration
	at  time.Time
	seq int64
}

func (t *coalesceTimers) after(d time.Duration) <-chan time.Time {
	ch := make(chan time.Time, 1)
	at := time.Time{}
	if t.now != nil {
		at = t.now().Add(d)
	}
	t.mu.Lock()
	t.calls = append(t.calls, d)
	t.seq++
	t.pending = append(t.pending, coalesceTimer{ch: ch, d: d, at: at, seq: t.seq})
	t.mu.Unlock()
	return ch
}

// callCount 返回定时器被请求的总次数。
func (t *coalesceTimers) callCount() int {
	t.mu.Lock()
	defer t.mu.Unlock()
	return len(t.calls)
}

// durations 返回每次请求的时长（按请求顺序）。
func (t *coalesceTimers) durations() []time.Duration {
	t.mu.Lock()
	defer t.mu.Unlock()
	return append([]time.Duration(nil), t.calls...)
}

// nextDue 取走**截止最早**的那一个未触发定时器（没有就返回 false）。
func (t *coalesceTimers) nextDue() (coalesceTimer, bool) {
	t.mu.Lock()
	defer t.mu.Unlock()
	if len(t.pending) == 0 {
		return coalesceTimer{}, false
	}
	idx := 0
	for i := 1; i < len(t.pending); i++ {
		if t.pending[i].at.Before(t.pending[idx].at) {
			idx = i
		}
	}
	timer := t.pending[idx]
	t.pending = append(t.pending[:idx], t.pending[idx+1:]...)
	return timer, true
}

// waitRegistered 等到「至少有一次定时器请求挂着」（= worker 确实在等窗口）。
//
// 它只做**存在性检查**，不触发、也不消耗定时器：旧夹具里有调用方用 take 来证明
// 「它在等窗口」，却把那个定时器丢掉了 —— 按截止时刻推进之后那会丢掉唯一活着的那个。
// seqNow 返回当前请求序号（投递前记下它，之后用 waitFresh 等「这次投递之后新挂的」定时器）。
func (t *coalesceTimers) seqNow() int64 {
	t.mu.Lock()
	defer t.mu.Unlock()
	return t.seq
}

// waitFresh 等到「序号大于 after 的定时器已挂上」= worker 已经把这次投递收进本批并重新
// 开始等窗口。旧请求（已被 worker 放弃、再没人监听）不算。
func (t *coalesceTimers) waitFresh(tb *testing.T, after int64, timeout time.Duration) {
	tb.Helper()
	deadline := time.Now().Add(timeout)
	for time.Now().Before(deadline) {
		t.mu.Lock()
		fresh := false
		for _, tm := range t.pending {
			if tm.seq > after {
				fresh = true
				break
			}
		}
		t.mu.Unlock()
		if fresh {
			return
		}
		time.Sleep(time.Millisecond)
	}
	tb.Fatalf("等不到投递之后新挂的定时器（seq>%d, calls=%d）", after, t.callCount())
}

func (t *coalesceTimers) waitRegistered(tb *testing.T, timeout time.Duration) {
	tb.Helper()
	deadline := time.Now().Add(timeout)
	for time.Now().Before(deadline) {
		t.mu.Lock()
		n := len(t.pending)
		t.mu.Unlock()
		if n > 0 {
			return
		}
		time.Sleep(time.Millisecond)
	}
	tb.Fatalf("no coalesce timer registered within %s (calls=%d)", timeout, t.callCount())
}

// advanceAndFire 把注入时钟推进到**当前所有未触发请求里最晚的那个截止时刻**，
// 然后触发它们（= 「这些窗口全部到点」）。
//
// 推进量因此恒等于「真的过了多久」，不会再出现旧写法里「按每个请求时长累加」的推过头。
func (t *coalesceTimers) advanceAndFire(tb *testing.T, clock *laneClock, done func() bool) {
	tb.Helper()
	deadline := time.Now().Add(3 * time.Second)
	// 与 pump 同一条不变量：先把队列排空再推钟（否则循环头会按窗口到点发车，
	// 而那条分支不 takeAll，尾部消息会落到下一批）。
	for t.depth != nil && t.depth() > 0 && time.Now().Before(deadline) {
		time.Sleep(time.Millisecond)
	}
	for time.Now().Before(deadline) {
		due, ok := t.allDue()
		if ok {
			clock.advanceTo(due)
			// 为什么要重试：worker 可能在「登记」与「触发」之间续了新臂，
			// 被触发的旧通道于是没人接（时钟推了、没人醒）。done 成立即收手。
			if done == nil || done() {
				return
			}
			time.Sleep(time.Millisecond)
			continue
		}
		if done != nil && done() {
			return
		}
		time.Sleep(time.Millisecond)
	}
	if done != nil && done() {
		return
	}
	tb.Fatalf("advanceAndFire: 3s 内没有把窗口推到发车（calls=%d）", t.callCount())
}

// advanceAndFireUntil 推进并触发窗口，直到**回合数比调用时多一个**（或超时）。
func (f *coalesceBridge) advanceAndFireUntil(t *testing.T) {
	t.Helper()
	before := f.chat.runCount()
	f.timers.advanceAndFire(t, f.clock, func() bool { return f.chat.runCount() > before })
}

// allDue 取走全部未触发请求，把时钟推到其中最晚的截止时刻，并逐个触发。
// 返回被推到的时间点（没有任何请求时 ok=false）。
func (t *coalesceTimers) allDue() (time.Time, bool) {
	t.mu.Lock()
	pending := t.pending
	t.pending = nil
	t.mu.Unlock()
	if len(pending) == 0 {
		return time.Time{}, false
	}
	latest := pending[0]
	for _, tm := range pending[1:] {
		if tm.at.After(latest.at) {
			latest = tm
		}
	}
	for _, tm := range pending {
		tm.ch <- latest.at
	}
	return latest.at, true
}

// pump 反复「把时钟推进到最早截止时刻并触发它」，直到 done() 成立（或超时）。
//
// 为什么不是「按次数触发」：窗口到达的方式有两种 —— 静默到点（要定时器）与**绝对上限
// 已过**（循环头直接发车，**根本不建定时器**）。测试若假定「一定会有定时器」，
// 就会被调度时序（worker 什么时候启动）绊倒；pump 因此只看**结果条件**。
func (f *coalesceBridge) pump(t *testing.T, done func() bool) {
	t.Helper()
	deadline := time.Now().Add(3 * time.Second)
	for time.Now().Before(deadline) {
		if done() {
			return
		}
		if timer, ok := f.timers.nextDue(); ok {
			// 🔴 **推钟前必须等队列排空**：时钟一旦前进，`coalesceWait` 的循环头就可能
			// 直接按「窗口到点」发车（那条分支不 takeAll），还没被 worker 收进本批的
			// 尾部消息就会落到下一批 —— 「12 条 → 2 轮」正是这么来的。
			// 这里等的是**状态**（车道上没有待取条目），不是拿 sleep 掩盖时序：
			// worker 一旦看到队列有货就会收（select 的 l.ch 分支），所以排空必然发生。
			if !f.awaitQueueDrained(deadline) {
				break
			}
			f.clock.advanceTo(timer.at)
			timer.ch <- timer.at
			continue
		}
		time.Sleep(time.Millisecond)
	}
	t.Fatalf("condition never met within 3s (runs=%d, queued=%d, stats=%+v)",
		f.chat.runCount(), f.queueDepth(), f.b.CoalesceStats())
}

// coalesceGroupSessions 给**每个群**一个稳定的隔离会话：同一群的多个成员 = 同一个会话
// （真实 SessionManager 就是这么派生的），因此「合成一轮」的落库行落在同一处。
type coalesceGroupSessions struct{}

func (coalesceGroupSessions) GetOrCreate(_ context.Context, msg adapter.UnifiedMessage, _ *user.OwnerProfile) (session.Session, error) {
	return session.Session{
		Platform:    msg.Platform,
		ChannelID:   msg.ChannelID,
		SessionID:   "sess-" + msg.GroupID,
		IsGroup:     true,
		MemWritable: false,
	}, nil
}

// coalesceGroupMsg 是一条**非主人、指向她本人**的群消息：它因此必然进回合
// （不受最小打扰分级影响），是静默窗口最干净的夹具。
func coalesceGroupMsg(platform, groupID, sender, senderName, content string) adapter.UnifiedMessage {
	return adapter.UnifiedMessage{
		Platform:    platform,
		ChannelID:   groupID,
		ChannelType: adapter.ChannelGroup,
		GroupID:     groupID,
		SenderID:    sender,
		SenderName:  senderName,
		Sender:      adapter.Actor{Role: adapter.RoleMember},
		Mentions:    []adapter.Mention{{UserID: sender}},
		Content:     content,
		ContentType: adapter.ContentText,
	}
}

// coalesceBridge 是一个装好注入时钟与定时器工厂的桥接器（外加它的日志/历史/审计面）。
type coalesceBridge struct {
	b      *Bridge
	chat   *laneChatter
	ad     *laneAdapter
	clock  *laneClock
	timers *coalesceTimers
}

// newCoalesceBridge 构造夹具：opts.Loop 必须是 *laneChatter（用例要读回合记录），
// 时钟与定时器工厂由夹具注入（opts.Now / opts.Coalesce.After 若已给出则不覆盖）。
func newCoalesceBridge(t *testing.T, opts Options) *coalesceBridge {
	t.Helper()
	chat, ok := opts.Loop.(*laneChatter)
	if !ok {
		t.Fatalf("newCoalesceBridge needs a *laneChatter loop, got %T", opts.Loop)
	}
	clock := newCoalesceClock()
	timers := &coalesceTimers{now: clock.Now}
	_ = timers
	if opts.Now == nil {
		opts.Now = clock.Now
	}
	opts.Coalesce.After = timers.after
	if opts.Sessions == nil {
		opts.Sessions = coalesceGroupSessions{}
	}
	if opts.Owner == nil {
		opts.Owner = &user.OwnerProfile{}
	}
	b, err := New(opts)
	if err != nil {
		t.Fatalf("New: %v", err)
	}
	ad := &laneAdapter{name: "telegram"}
	b.Add(ad)
	timers.depth = func() int { return queueDepthOf(b) }
	return &coalesceBridge{b: b, chat: chat, ad: ad, clock: clock, timers: timers}
}

// snapshot 在锁内返回某个会话已落库的行（测试直接读 h.rows 会与车道 worker 的
// 写入构成数据竞争 —— -race 下必红，因此统一走这里）。
func (h *laneHistory) snapshot(sessionID string) []history.Message {
	h.mu.Lock()
	defer h.mu.Unlock()
	return append([]history.Message(nil), h.rows[sessionID]...)
}

// send 投递一条消息（走真实的 onMessage 入口：Validate → 流控闸 → 车道）。
//
// ctx 取 b.ctx（与生产的读循环一致：`readLoop` 传的就是桥自己的生命周期 context），
// 这样 Stop 取消的车道 worker 与用例投递的消息用的是同一棵树 —— 否则「停机诚实」
// 这条用例根本验证不到真实路径（b.ctx 为 nil 时退回 Background）。
func (f *coalesceBridge) send(msg adapter.UnifiedMessage) {
	ctx := f.b.ctx
	if ctx == nil {
		ctx = context.Background()
	}
	f.b.onMessage(ctx, msg, f.ad)
}

// userTextsIn 返回一次回合的模型可见消息里的 user 文本（按顺序）。
func userTextsIn(msgs []model.Message) []string {
	out := make([]string, 0, len(msgs))
	for _, m := range msgs {
		if m.Role != model.RoleUser {
			continue
		}
		out = append(out, lastText(m))
	}
	return out
}

// waitRuns 等至少 n 个回合开跑（封装 chatter 的同名方法，省得到处传超时）。
func (f *coalesceBridge) waitRuns(t *testing.T, n int) {
	t.Helper()
	f.chat.waitRuns(t, n, 3*time.Second)
}

// runTexts 返回每个已开跑回合的 user 文本（按开跑顺序）。只用于失败信息：
// 「12 条 → 2 轮」这类断言失守时，得先看清每条落到了哪一批，才能判断是夹具相位问题
// 还是合并真的拆了批。
func runTexts(c *laneChatter) [][]string {
	n := c.runCount()
	out := make([][]string, 0, n)
	for i := 0; i < n; i++ {
		out = append(out, userTextsIn(c.runAt(i).messages))
	}
	return out
}

// waitIdle 等到车道全部退休。
func (f *coalesceBridge) waitIdle(t *testing.T) {
	t.Helper()
	// 「等车道全部退休」必须在**推动窗口**的前提下等：worker 若正停在一个还没到点的
	// 窗口里，就永远走不到「队列空 ⇒ 退休」那一行。这里因此一边等一边把挂着的定时器
	// 按截止时刻触发（与 pump 同一条语义），判据仍是 waitLaneIdle 自己那套断言。
	deadline := time.Now().Add(2 * time.Second)
	for time.Now().Before(deadline) {
		if f.b.LaneStats().ActiveLanes == 0 {
			return
		}
		if timer, ok := f.timers.nextDue(); ok {
			f.clock.advanceTo(timer.at)
			timer.ch <- timer.at
			continue
		}
		time.Sleep(time.Millisecond)
	}
	waitLaneIdle(t, f.b, time.Millisecond)
}

// awaitQueueDrained 等到车道上没有待取条目（= worker 已把该收的都收进本批）。
// 返回 false 表示到 deadline 都没排空（由调用方的断言/超时报出来）。
func (f *coalesceBridge) awaitQueueDrained(deadline time.Time) bool {
	for time.Now().Before(deadline) {
		if f.queueDepth() == 0 {
			return true
		}
		time.Sleep(time.Millisecond)
	}
	return f.queueDepth() == 0
}

// queueDepth 返回所有车道里「还没被 worker 取走」的条目数（ch + spill + front）。
func (f *coalesceBridge) queueDepth() int { return queueDepthOf(f.b) }

func queueDepthOf(b *Bridge) int {
	if b == nil || b.lanes == nil {
		return 0
	}
	b.lanes.mu.Lock()
	defer b.lanes.mu.Unlock()
	n := 0
	for _, l := range b.lanes.lanes {
		n += len(l.ch) + len(l.spill) + len(l.front)
	}
	return n
}

// sendSettled 投递一条，并等 worker 把它收进本批（车道排空）后才返回。
//
// 给「投递与注入时钟交替推进」的用例用（如每隔 500ms 来一条的绝对上限用例）：
// 时钟只能表示「时间过去了」，所以必须保证每条都已经到了 worker 手里再走时间，
// 否则「到点发车」会漏掉还没被取走的尾部（实测批次 8/12）。
// 注意：**不要**在需要验证「队列能装下整批、不溢出」的用例里用它 —— 那里要的是
// 真的把消息压进队列。
func (f *coalesceBridge) sendSettled(t *testing.T, msg adapter.UnifiedMessage) {
	t.Helper()
	deadline := time.Now().Add(3 * time.Second)
	before := f.timers.seqNow()
	f.send(msg)
	if !f.awaitQueueDrained(deadline) {
		t.Fatalf("投递后车道没有排空（queued=%d）", f.queueDepth())
	}
	// 还要等 worker **为这次投递重新挂上定时器**（= 它已回到 select 里等窗口）。
	// 必须按序号判新鲜度：pending 里可能还躺着此前那些「再没人监听」的旧请求，
	// 拿它们当「已就绪」会让下一步推钟落在两轮循环之间（定时器换了新的、旧的被触发没人接）。
	f.timers.waitFresh(t, before, time.Until(deadline))
}

// waitRows 等到会话落了至少 want 行指定角色的记录（回合是异步跑完再落库的，
// 读历史前必须等它真的写完 —— 否则断言会读到「12 user + 0 assistant」的半成品）。
func (h *laneHistory) waitRows(tb *testing.T, sessionID, role string, want int, timeout time.Duration) {
	tb.Helper()
	deadline := time.Now().Add(timeout)
	for time.Now().Before(deadline) {
		n := 0
		for _, row := range h.snapshot(sessionID) {
			if row.Role == role {
				n++
			}
		}
		if n >= want {
			return
		}
		time.Sleep(time.Millisecond)
	}
	tb.Fatalf("等了 %s 仍只有 %d 行 %s（want %d）", timeout, len(h.snapshot(sessionID)), role, want)
}

// sendSettledAndDuration 投递一条并返回**这条消息之后 worker 续出的那次窗口时长**。
//
// 它把「投递 → worker 收进本批 → 为它续臂」做成一个确定的相位：续臂用请求序号认领
// （`seq > 投递前`），因此不会把此前那条消息的续臂误当成这一次的 —— 只按「至少 N 次
// 请求 + 取最新」在负载下仍会拿到上一条的时长（实测 -race -count=200 有 2/200 假红）。
func (f *coalesceBridge) sendSettledAndDuration(t *testing.T, msg adapter.UnifiedMessage) time.Duration {
	t.Helper()
	f.sendSettled(t, msg)
	return latestDuration(f.timers.durations())
}