Cyrene_v2/internal/services/channelbridge/coalesce_admission_test.go

package channelbridge

// coalesce_admission_test.go 钉住 M23 的**窗口准入条件**（复审 r1 的 P2-1）：
//
//	窗口只收 chat/image/voice/file；**不收**系统通知 / action 帧 / 我们自己出站回执。
//
// 为什么这条要用例钉死（而不是靠注释）：准入判断必须发生在**窗口等待阶段**（车道头），
// 而不是等窗口到了、prepareInbound 跑完才发现「这条本来不该进批」。否则一条我们自己的
// `message_sent` 回执（它与群聊共用同一条车道键）就能把这一批的静默计时**一直续住**：
// 群友说了三句话，她却因为「撤回事件」「入群事件」「我们自己的回执」而迟迟不开口。
//
// 四条断言（每条都能被「准入失效」这一变异打红）：
//   - 窗口中途的 action 帧**不重置计时**（剩余等待仍是 500ms，发车时刻不变）；
//   - 系统通知**不进这一批**（批内 user 行只有聊天消息）；
//   - 单条出站回执**不建窗口**（定时器一次都没被请求）；
//   - 对照组：普通聊天消息**仍然**重置计时（否则上面三条可能只是「窗口根本不工作」）。

import (
	"testing"
	"time"

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

// coalesceActionFrame 是一条**平台帧**（action）：撤回 / 入群 / typing / 未知 post_type。
// 它的生产形状与本群聊天共用车道的键（ChannelType=group + GroupID），ContentType=action。
func coalesceActionFrame(platform, groupID, postType string) adapter.UnifiedMessage {
	return adapter.UnifiedMessage{
		Platform:    platform,
		ChannelID:   groupID,
		ChannelType: adapter.ChannelGroup,
		GroupID:     groupID,
		SenderID:    "9",
		Sender:      adapter.Actor{Role: adapter.RoleMember},
		ContentType: adapter.ContentAction,
		RawData:     []byte(`{"post_type":"` + postType + `"}`),
	}
}

// coalesceOutboundReceipt 是**我们自己那条消息的回执**（OneBot 的 message_sent）：
// 它是一条 ContentType=action、没有正文的群帧 —— 也正是「能把窗口续住」的那条。
func coalesceOutboundReceipt(platform, groupID string) adapter.UnifiedMessage {
	msg := coalesceActionFrame(platform, groupID, "message_sent")
	msg.SenderID = "100" // 她自己
	msg.Sender = adapter.Actor{Role: adapter.RoleOwner}
	return msg
}

// latestDuration 取请求时长序列里**最新**的一个。
//
// 用例关心的是「刚发的那条消息之后 worker 续的臂」，而不是历史第几条：臂的次数取决于
// worker 什么时候把消息收进本批（调度决定），按下标取会引入时序假设。
func latestDuration(d []time.Duration) time.Duration {
	if len(d) == 0 {
		return 0
	}
	return d[len(d)-1]
}

// waitTimerCalls 等到定时器工厂至少被请求 n 次，返回到目前为止的**请求时长序列**。
func waitTimerCalls(t *testing.T, f *coalesceBridge, n int) []time.Duration {
	t.Helper()
	deadline := time.Now().Add(3 * time.Second)
	for time.Now().Before(deadline) {
		if d := f.timers.durations(); len(d) >= n {
			return d
		}
		time.Sleep(time.Millisecond)
	}
	t.Fatalf("timer factory called %d times, want at least %d", f.timers.callCount(), n)
	return nil
}

// fireLatest 把注入时钟推进到**最近一次**请求的截止时刻并触发它。
//
// 为什么是最近一次：窗口被重置/续臂时，旧的定时器通道留在那里（再没人监听），
// 真正决定发车时刻的是**最近一次**请求的那个。
//
// 为什么按**截止时刻**推进而不是「再推一个请求时长」：worker 可能已经在这段时间里
// 重新续过臂（例如一条系统通知落在窗口中途），把时长再累加一次会让时钟越过绝对上限，
// 循环头于是走「到点直接发车、不 takeAll」的分支；按截止时刻推进只表示「时间走到了
// 它约定的那个点」。
//
// 为什么要重试：登记与触发之间隔着 worker 的一次循环 —— 万一它刚续了新臂，被触发的
// 那个通道就没人接（时钟已经推了，却没人醒）。这里在 done() 未成立、仍有待触发定时器时
// 继续触发，直到 worker 真的发车；断言口径由调用方守。
func (t *coalesceTimers) fireLatest(tb *testing.T, clock *laneClock, done func() bool) {
	tb.Helper()
	deadline := time.Now().Add(3 * time.Second)
	for time.Now().Before(deadline) {
		t.mu.Lock()
		if len(t.pending) == 0 {
			t.mu.Unlock()
			if done != nil && done() {
				return
			}
			time.Sleep(time.Millisecond)
			continue
		}
		timer := t.pending[len(t.pending)-1]
		t.pending = t.pending[:len(t.pending)-1]
		t.mu.Unlock()
		clock.advanceTo(timer.at)
		timer.ch <- timer.at
		if done != nil && done() {
			return
		}
		time.Sleep(time.Millisecond)
	}
	tb.Fatalf("fireLatest: 3s 内没有把窗口推到发车（calls=%d）", t.callCount())
}

// TestCoalesceActionFrameDoesNotResetWindow 断言**准入在窗口等待阶段就生效**：
// 一条平台帧（action）落在窗口中途，既不进批、也不重置静默计时 ——
// 剩余等待仍是 500ms，发车时刻仍是「m1 到达 + 2.0s」。
//
// 变异：让 laneWindowAdmits 恒真（= 修前行为）⇒ 第二次请求的时长变成 2000ms，本用例红。
func TestCoalesceActionFrameDoesNotResetWindow(t *testing.T) {
	t.Parallel()

	chat := &laneChatter{}
	f := newCoalesceBridge(t, Options{
		Loop:     chat,
		Coalesce: CoalesceOptions{For: coalesceFor(coalesceOn(2000, 6000, 24))},
	})
	// 投递用 sendSettled：它等到 worker 把这条收进本批、并**为它**续好臂才返回。
	// 这样「哪一次续臂属于哪条消息」是确定的 —— 只按「至少 N 次请求 + 取最新」仍有
	// 缝：worker 可能在帧之后才补上 m1 的续臂，最新那一次就是 2s（假红）。
	if d := f.sendSettledAndDuration(t, coalesceGroupMsg("telegram", "-20", "9", "张三", "m1")); d != 2000*time.Millisecond {
		// 第一条聊天消息 ⇒ 一个完整窗口（2.0s）。
		t.Fatalf("first window = %s, want 2s", d)
	}

	// 窗口走到 1.5s 时来了一条平台帧。
	f.clock.advance(1500 * time.Millisecond)
	if d := f.sendSettledAndDuration(t, coalesceActionFrame("telegram", "-20", "notice.group_recall")); d != 500*time.Millisecond {
		// 🔴 关键断言（状态，不是结果）：这条帧续出的窗口**剩余**仍是 500ms。
		// 若这条帧被当成聊天消息（修前），计时会重置成完整的 2000ms。
		t.Fatalf("re-armed window = %s, want 500ms (an action frame must not reset the quiet window)", d)
	}

	f.timers.fireLatest(t, f.clock, func() bool { return chat.runCount() >= 1 })
	f.waitRuns(t, 1)
	f.waitIdle(t)

	if n := chat.runCount(); n != 1 {
		t.Fatalf("turns started = %d, want 1 (the frame must not add a turn)", n)
	}
	if got := userTextsIn(chat.runAt(0).messages); len(got) != 1 || got[0] != "m1" {
		t.Fatalf("batch user messages = %v, want [m1] (the frame must not join the batch)", got)
	}
	stats := f.b.CoalesceStats()
	if stats.WindowsCreated != 1 || stats.BatchesTotal != 1 {
		t.Fatalf("stats = %+v, want one window and one batch", stats)
	}
}

// TestCoalesceSystemNoticeDoesNotJoinBatch 断言系统通知**不进这一批**：
// 两条聊天消息 + 中途一条通知 ⇒ 一轮、批内只有那两条聊天消息。
func TestCoalesceSystemNoticeDoesNotJoinBatch(t *testing.T) {
	t.Parallel()

	chat := &laneChatter{}
	f := newCoalesceBridge(t, Options{
		Loop:     chat,
		Coalesce: CoalesceOptions{For: coalesceFor(coalesceOn(2000, 6000, 24))},
	})
	// 每条都等「worker 收进本批 + 为它续好臂」再继续（见 sendSettledAndDuration）：
	// 只按「至少 N 次请求 + 取最新」有缝 —— worker 可能在通知之后才补上 m1/m2 的
	// 续臂，最新那一次就是 2s，用例假红。
	f.sendSettledAndDuration(t, coalesceGroupMsg("telegram", "-20", "9", "张三", "m1"))
	f.sendSettledAndDuration(t, coalesceGroupMsg("telegram", "-20", "9", "张三", "m2"))

	f.clock.advance(1500 * time.Millisecond)
	// 与 action 帧同理：通知也不许重置计时（剩余仍是 500ms）。
	// 变异（准入失效）⇒ 这里拿到 2s，本用例红。
	if d := f.sendSettledAndDuration(t, coalesceActionFrame("telegram", "-20", "notice.group_increase")); d != 500*time.Millisecond {
		t.Fatalf("re-armed window = %s, want 500ms (a system notice must not reset the quiet window)", d)
	}

	f.timers.fireLatest(t, f.clock, func() bool { return chat.runCount() >= 1 })
	f.waitRuns(t, 1)
	f.waitIdle(t)

	if n := chat.runCount(); n != 1 {
		t.Fatalf("turns started = %d, want 1 (the notice must not drive its own turn here)", n)
	}
	if got := userTextsIn(chat.runAt(0).messages); len(got) != 2 || got[0] != "m1" || got[1] != "m2" {
		t.Fatalf("batch user messages = %v, want [m1 m2] only", got)
	}
	if stats := f.b.CoalesceStats(); stats.BatchesTotal != 1 {
		t.Fatalf("batches = %d, want 1 (the notice is not a batch)", stats.BatchesTotal)
	}
}

// TestCoalesceOutboundReceiptCreatesNoWindow 断言**我们自己的出站回执**不建窗口：
// 单条回执到达 ⇒ 定时器工厂一次都没被请求、没有窗口、没有批、也没有回合。
//
// 变异：让 laneWindowAdmits 恒真 ⇒ 回执会建一个窗口（定时器被请求、windows_created=1），
// 本用例红 —— 这条正是「一条回执就能把窗口续住」的最小复现。
func TestCoalesceOutboundReceiptCreatesNoWindow(t *testing.T) {
	t.Parallel()

	chat := &laneChatter{}
	f := newCoalesceBridge(t, Options{
		Loop:     chat,
		Coalesce: CoalesceOptions{For: coalesceFor(coalesceOn(2000, 6000, 24))},
	})
	f.send(coalesceOutboundReceipt("telegram", "-20"))
	// 先看「有没有窗口」再等空闲：修前会建一个永不触发的窗口（calls=1），
	// 这里因此留下的是「窗口被建出来了」这条最直接的证据，而不是一句「车道没退休」。
	// 轮询到车道退休（或 300ms）是必要的：投递返回时 worker 可能还没起势，
	// 一次性读 callCount() 会让这条断言在修前也偶尔通过。
	deadline := time.Now().Add(300 * time.Millisecond)
	for time.Now().Before(deadline) {
		if calls := f.timers.callCount(); calls != 0 {
			t.Fatalf("coalesce timer factory called %d times for an outbound receipt, want 0", calls)
		}
		if f.b.LaneStats().ActiveLanes == 0 {
			break
		}
		time.Sleep(time.Millisecond)
	}
	stats := f.b.CoalesceStats()
	if stats.WindowsCreated != 0 || stats.BatchesTotal != 0 {
		t.Fatalf("stats = %+v, want no window and no batch for a lone receipt", stats)
	}
	if n := chat.runCount(); n != 0 {
		t.Fatalf("turns started = %d for a lone receipt, want 0", n)
	}
	f.waitIdle(t)
}

// TestCoalesceChatStillResetsWindow 是对照组：普通聊天消息**仍然**重置计时
// （否则上面三条可能只是「窗口本来就不工作」这种伪结论）。
func TestCoalesceChatStillResetsWindow(t *testing.T) {
	t.Parallel()

	chat := &laneChatter{}
	f := newCoalesceBridge(t, Options{
		Loop:     chat,
		Coalesce: CoalesceOptions{For: coalesceFor(coalesceOn(2000, 6000, 24))},
	})
	if d := f.sendSettledAndDuration(t, coalesceGroupMsg("telegram", "-20", "9", "张三", "m1")); d != 2000*time.Millisecond {
		t.Fatalf("first window = %s, want 2s", d)
	}
	f.clock.advance(1500 * time.Millisecond)
	// 同 action-frame 用例：读的是**这条消息**续出的臂（sendSettledAndDuration 保证归属）。
	if d := f.sendSettledAndDuration(t, coalesceGroupMsg("telegram", "-20", "9", "张三", "m2")); d != 2000*time.Millisecond {
		t.Fatalf("re-armed window = %s, want a full 2s (a chat message resets the window)", d)
	}
	f.timers.fireLatest(t, f.clock, func() bool { return chat.runCount() >= 1 })
	f.waitRuns(t, 1)
	if got := userTextsIn(chat.runAt(0).messages); len(got) != 2 {
		t.Fatalf("batch user messages = %v, want both chat messages", got)
	}
}