Cyrene_v2/internal/channel/adapter/flowgate_session_test.go

package adapter

// 本文件覆盖裁决 06.4.3：节流（群聊滑窗与待处理上限）的计数键是**会话级**
// （ConversationKey = platform+channelID+groupID/peerID），不在适配器顶层汇总——
// 两个群并发互不挤占。

import (
	"context"
	"testing"
	"time"
)

// groupMsgIn 构造指定群的群聊消息（platform kook），用于验证会话级计数。
func groupMsgIn(groupID string, i int) UnifiedMessage {
	return UnifiedMessage{
		Platform:    "kook",
		ChannelID:   groupID,
		ChannelType: ChannelGroup,
		GroupID:     groupID,
		SenderID:    "u" + itoa(i),
		Content:     "msg " + itoa(i),
		ContentType: ContentText,
	}
}

// directMsgTo 构造指定对端的私聊消息，用于验证私聊额度独立。
func directMsgTo(peer string, i int) UnifiedMessage {
	return UnifiedMessage{
		Platform:    "telegram",
		ChannelID:   peer,
		ChannelType: ChannelDirect,
		SenderID:    peer,
		Content:     "dm " + itoa(i),
		ContentType: ContentText,
	}
}

// TestConversationKeySessionScoped 固定会话键口径（裁决 06.4.3）：群聊按 (平台, 群)
// 独立、私聊按 (平台, 对端) 独立，不同会话的键互不相等。
func TestConversationKeySessionScoped(t *testing.T) {
	t.Parallel()

	if ConversationKey(groupMsgIn("-100", 0)) != ConversationKey(groupMsgIn("-100", 1)) {
		t.Error("同一个群的两条消息应落在同一个会话键")
	}
	if ConversationKey(groupMsgIn("-100", 0)) == ConversationKey(groupMsgIn("-200", 0)) {
		t.Error("两个群共用了同一个会话键（会话级计数失效）")
	}
	if ConversationKey(directMsgTo("peer-1", 0)) == ConversationKey(directMsgTo("peer-2", 0)) {
		t.Error("两个私聊对端共用了同一个会话键")
	}
	if ConversationKey(directMsgTo("peer-1", 0)) == ConversationKey(groupMsgIn("-100", 0)) {
		t.Error("私聊与群聊共用了同一个会话键")
	}

	// 群聊缺 group id 时退回 channel id：仍按群独立，不得退化成按发送者分桶
	// （否则限流会被发言者数量放大而失效）。
	bare := UnifiedMessage{Platform: "kook", ChannelType: ChannelGroup, ChannelID: "-100"}
	withSender := UnifiedMessage{
		Platform: "kook", ChannelType: ChannelGroup, ChannelID: "-100", SenderID: "u9",
	}
	if ConversationKey(bare) != ConversationKey(withSender) {
		t.Error("群聊缺 group id 时退化成按发送者分桶")
	}
}

// TestFlowGate_GroupRateLimitIsPerConversation 是裁决 06.4.3 的核心断言：群聊滑窗按
// 每个群独立计数——A 群用尽额度不挤占 B 群（两个群并发互不挤占）。
func TestFlowGate_GroupRateLimitIsPerConversation(t *testing.T) {
	t.Parallel()

	now, clock := fixedClock(time.Unix(1000, 0))
	g := NewFlowGate(FlowGateOptions{GroupMsgPerMin: 2, Window: time.Minute, Now: now})
	ctx := context.Background()

	for i := 0; i < 2; i++ {
		if v := g.Admit(ctx, groupMsgIn("-100", i)); !v.Proceed {
			t.Fatalf("A 群第 %d 条被拒，want admitted: %+v", i+1, v)
		}
	}
	if v := g.Admit(ctx, groupMsgIn("-100", 2)); v.Proceed {
		t.Fatal("A 群超出自身额度仍被放行")
	}

	for i := 0; i < 2; i++ {
		if v := g.Admit(ctx, groupMsgIn("-200", i)); !v.Proceed {
			t.Fatalf("B 群第 %d 条被 A 群额度挤占: %+v", i+1, v)
		}
	}
	if v := g.Admit(ctx, groupMsgIn("-200", 9)); v.Proceed {
		t.Fatal("B 群超出自身额度仍被放行")
	}

	// 私聊有自己的额度，不参与群聊滑窗。
	if v := g.Admit(ctx, directMsgTo("peer-1", 0)); !v.Proceed {
		t.Fatalf("私聊被群聊额度挤占: %+v", v)
	}

	// 窗口滚过之后 A 群额度恢复。
	*clock = clock.Add(61 * time.Second)
	if v := g.Admit(ctx, groupMsgIn("-100", 3)); !v.Proceed {
		t.Fatalf("窗口滚过后 A 群应恢复放行: %+v", v)
	}
}

// TestFlowGate_CapIsPerConversation 覆盖策略四同样按会话独立：一个群堆满待处理消息
// 触发的是本群的 FIFO 淘汰，不会把另一个群的待处理消息挤掉。
func TestFlowGate_CapIsPerConversation(t *testing.T) {
	t.Parallel()

	g := NewFlowGate(FlowGateOptions{MaxMessages: 2})
	ctx := context.Background()

	for i := 0; i < 2; i++ {
		if v := g.Admit(ctx, groupMsgIn("-100", i)); !v.Proceed {
			t.Fatalf("A 群第 %d 条被拒: %+v", i+1, v)
		}
	}
	v := g.Admit(ctx, groupMsgIn("-100", 2))
	if !v.Proceed {
		t.Fatalf("A 群新消息在 FIFO 淘汰后应放行: %+v", v)
	}
	if len(v.Pruned) != 1 || v.Pruned[0].ChannelID != "-100" {
		t.Fatalf("A 群淘汰的应是本群消息, got %+v", v.Pruned)
	}

	for i := 0; i < 2; i++ {
		if v := g.Admit(ctx, groupMsgIn("-200", i)); !v.Proceed || len(v.Pruned) != 0 {
			t.Fatalf("B 群被 A 群的待处理队列挤占: %+v", v)
		}
	}
}