Cyrene_v2/internal/channel/adapter/flowgate_test.go

package adapter

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

// fixedClock returns a Now func pinned to a base time so tests can advance time by
// controlling the returned value.
func fixedClock(base time.Time) (func() time.Time, *time.Time) {
	cur := base
	return func() time.Time { return cur }, &cur
}

func groupMsg(i int) UnifiedMessage {
	return UnifiedMessage{
		Platform:    "kook",
		ChannelID:   "-100",
		ChannelType: ChannelGroup,
		GroupID:     "-100",
		SenderID:    "u" + itoa(i),
		Content:     "msg " + itoa(i),
		ContentType: ContentText,
	}
}

func directMsg(i int) UnifiedMessage {
	return UnifiedMessage{
		Platform:    "telegram",
		ChannelID:   "100",
		ChannelType: ChannelDirect,
		SenderID:    "u" + itoa(i),
		Content:     "dm " + itoa(i),
		ContentType: ContentText,
	}
}

func itoa(n int) string {
	if n == 0 {
		return "0"
	}
	var b [20]byte
	i := len(b)
	for n > 0 {
		i--
		b[i] = byte('0' + n%10)
		n /= 10
	}
	return string(b[i:])
}

func TestFlowGate_AllZeroAllowsEverything(t *testing.T) {
	t.Parallel()
	g := NewFlowGate(FlowGateOptions{})
	for i := 0; i < 20; i++ {
		if v := g.Admit(context.Background(), groupMsg(i)); !v.Proceed {
			t.Fatalf("message %d dropped by zero-option gate: %+v", i, v)
		}
	}
}

func TestFlowGate_GroupRateLimitDowngradesOverLimit(t *testing.T) {
	t.Parallel()
	now, clock := fixedClock(time.Unix(1000, 0))
	g := NewFlowGate(FlowGateOptions{
		GroupMsgPerMin: 6,
		Window:         time.Minute,
		Now:            now,
	})

	// Six messages within the window are admitted; the seventh is downgraded.
	for i := 0; i < 6; i++ {
		if v := g.Admit(context.Background(), groupMsg(i)); !v.Proceed {
			t.Fatalf("message %d should be admitted within limit", i)
		}
	}
	v := g.Admit(context.Background(), groupMsg(6))
	if v.Proceed {
		t.Fatal("7th group message in the window must be rate limited")
	}
	if v.Reason != "rate_limited" {
		t.Fatalf("reason = %q, want rate_limited", v.Reason)
	}
	if v.DroppedMsg == nil || v.DroppedMsg.Content != "msg 6" {
		t.Fatalf("dropped msg = %+v, want msg 6", v.DroppedMsg)
	}

	// Advancing past the window frees capacity again.
	*clock = clock.Add(61 * time.Second)
	if v := g.Admit(context.Background(), groupMsg(7)); !v.Proceed {
		t.Fatalf("message after window roll should be admitted, got %+v", v)
	}
}

func TestFlowGate_GroupRateLimitDoesNotTouchDirect(t *testing.T) {
	t.Parallel()
	g := NewFlowGate(FlowGateOptions{GroupMsgPerMin: 1, Window: time.Minute})
	// Direct/private messages bypass the group rate limit.
	for i := 0; i < 5; i++ {
		if v := g.Admit(context.Background(), directMsg(i)); !v.Proceed {
			t.Fatalf("direct message %d wrongly rate limited: %+v", i, v)
		}
	}
}

func TestFlowGate_MsgPerMinCapsAllNonOwner(t *testing.T) {
	t.Parallel()
	g := NewFlowGate(FlowGateOptions{MsgPerMin: 2, Window: time.Minute})
	if v := g.Admit(context.Background(), directMsg(0)); !v.Proceed {
		t.Fatal("first dm should pass")
	}
	if v := g.Admit(context.Background(), directMsg(1)); !v.Proceed {
		t.Fatal("second dm should pass")
	}
	if v := g.Admit(context.Background(), directMsg(2)); v.Proceed {
		t.Fatal("third dm should exceed the total per-minute cap")
	}
}

// TestFlowGate_ReviewPrunesBacklogAfterAdmission 固定策略四的新口径：审查**不在准入
// 路径上**。越阈那一条消息照常放行（它的到达只负责派发审查），审查完成后只修剪
// **它当时看到的那批**待处理条目，并以 "review_pruned" 入审计。
//
// 变异判据：把审查改回在 Admit 里同步执行并按结论丢「刚到的那条」⇒ 本用例的
// `v.Proceed` 断言红。
func TestFlowGate_ReviewPrunesBacklogAfterAdmission(t *testing.T) {
	t.Parallel()
	now, _ := fixedClock(time.Unix(1000, 0))
	audit := &auditRecorder{}
	g := NewFlowGate(FlowGateOptions{
		MaxMessages: 10,
		Window:      time.Minute,
		Now:         now,
		Review: func(_ context.Context, msgs []UnifiedMessage) ([]int, error) {
			// Keep only the first two messages; drop the rest as noise.
			return []int{0, 1}, nil
		},
		OnPruned: audit.hook(),
	})

	// Seven messages stay below the 80% threshold (reviewAt = 8), no review fires.
	for i := 0; i < 7; i++ {
		if v := g.Admit(context.Background(), groupMsg(i)); !v.Proceed {
			t.Fatalf("message %d should be admitted before threshold: %+v", i, v)
		}
	}
	waitReviewIdle(t, g)
	if n := len(audit.snapshot()); n != 0 {
		t.Fatalf("review fired below threshold: %d audits", n)
	}

	// The 8th message crosses the 80% threshold → the review is dispatched in the
	// background; the message itself is still admitted (admission never waits).
	v := g.Admit(context.Background(), groupMsg(7))
	if !v.Proceed {
		t.Fatalf("the message that crosses the threshold must still be admitted, got %+v", v)
	}
	if v.Reason != "" {
		t.Fatalf("reason = %q, want empty", v.Reason)
	}
	waitReviewIdle(t, g)

	// The review keeps 0 and 1, so msg2..7 are pruned from the pending set and audited.
	if n := len(audit.snapshot()); n != 6 {
		t.Fatalf("audit count = %d, want 6 (msg2..msg7)", n)
	}
	for _, e := range audit.snapshot() {
		if e.reason != "review_pruned" {
			t.Fatalf("audit reason = %q, want review_pruned", e.reason)
		}
	}
	if n := pendingLen(g, ConversationKey(groupMsg(0))); n != 2 {
		t.Fatalf("pending set = %d, want 2 (only the kept entries)", n)
	}
}

// TestFlowGate_ReviewThrottledWithinWindow 固定「每会话每窗口至多一次审查」。
// 审查者保留全部条目，使待处理集合恒在阈值上 —— 于是节流（而不是队列长度）是唯一的
// 抑制因素；这正是修复前失守的那条判据（lastReview 只在成功分支置位）。
func TestFlowGate_ReviewThrottledWithinWindow(t *testing.T) {
	t.Parallel()
	now, clock := fixedClock(time.Unix(1000, 0))
	var mu sync.Mutex
	reviews := 0
	g := NewFlowGate(FlowGateOptions{
		MaxMessages: 3,
		Window:      time.Minute,
		Now:         now,
		Review: func(ctx context.Context, msgs []UnifiedMessage) ([]int, error) {
			mu.Lock()
			reviews++
			mu.Unlock()
			return keepAll(ctx, msgs)
		},
	})
	count := func() int {
		mu.Lock()
		defer mu.Unlock()
		return reviews
	}

	// First threshold crossing triggers a review.
	for i := 0; i < 3; i++ {
		g.Admit(context.Background(), groupMsg(i))
	}
	waitReviewIdle(t, g)
	if got := count(); got != 1 {
		t.Fatalf("reviews after crossing threshold = %d, want 1", got)
	}

	// Within the same window the gate must not re-review even when it refills.
	for i := 3; i < 9; i++ {
		g.Admit(context.Background(), groupMsg(i))
	}
	waitReviewIdle(t, g)
	if got := count(); got != 1 {
		t.Fatalf("reviews within window = %d, want 1 (throttled)", got)
	}

	// A fresh window allows a review again.
	*clock = clock.Add(2 * time.Minute)
	for i := 9; i < 12; i++ {
		g.Admit(context.Background(), groupMsg(i))
	}
	waitReviewIdle(t, g)
	if got := count(); got != 2 {
		t.Fatalf("reviews after window roll = %d, want 2", got)
	}
}

func TestFlowGate_FifoCapWhenNoReview(t *testing.T) {
	t.Parallel()
	var audited []string
	g := NewFlowGate(FlowGateOptions{
		MaxMessages: 3,
		OnPruned: func(_ UnifiedMessage, reason string) {
			audited = append(audited, reason)
		},
	})

	for i := 0; i < 3; i++ {
		if v := g.Admit(context.Background(), groupMsg(i)); !v.Proceed {
			t.Fatalf("message %d should be admitted at cap: %+v", i, v)
		}
	}
	// The 4th evicts the oldest (msg0) to honor the cap; the new message is admitted.
	v := g.Admit(context.Background(), groupMsg(3))
	if !v.Proceed {
		t.Fatalf("new message should be admitted after FIFO eviction, got %+v", v)
	}
	if len(v.Pruned) != 1 || v.Pruned[0].Content != "msg 0" {
		t.Fatalf("pruned = %+v, want [msg 0]", v.Pruned)
	}
	if len(audited) != 1 || audited[0] != "capped" {
		t.Fatalf("audit = %v, want [capped]", audited)
	}
}

func TestFlowGate_Defaults(t *testing.T) {
	t.Parallel()
	if FlowLimitDefault != 6 {
		t.Fatalf("FlowLimitDefault = %d, want 6", FlowLimitDefault)
	}
	if FlowCapDefault != 50 {
		t.Fatalf("FlowCapDefault = %d, want 50", FlowCapDefault)
	}
	g := NewFlowGate(FlowGateOptions{})
	if g.window != time.Minute {
		t.Fatalf("default window = %v, want 1m", g.window)
	}
	if g.reviewAt != 0 || g.maxMessages != 0 {
		t.Fatalf("zero-option gate should disable both cap and review, got reviewAt=%d max=%d", g.reviewAt, g.maxMessages)
	}
}