Cyrene_v2/internal/channel/adapter/flowgate.go

package adapter

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

// Reviewer decides which buffered messages are worth keeping. It receives the
// current pending set (in arrival order) and returns the indices it wants to keep.
// Messages at a returned index survive; all others are pruned and audited.
//
// It is a pure noise/importance filter (design/12 §6 策略四). Implementations must
// never decide identity: who the sender is was already resolved structurally by the
// channel layer. A nil Reviewer makes the gate fall back to a plain FIFO cap.
type Reviewer func(ctx context.Context, msgs []UnifiedMessage) (keep []int, err error)

// GateVerdict reports how the gate classified one inbound message.
type GateVerdict struct {
	// Proceed reports whether the message should be delivered downstream.
	Proceed bool
	// Reason is a short token describing a drop: "rate_limited", "review_pruned"
	// or "capped". It is empty when Proceed is true.
	Reason string
	// DroppedMsg is the message that failed admission (nil when Proceed is true).
	DroppedMsg *UnifiedMessage
	// Pruned lists every message the gate dropped during this admission. On the
	// review path it is the noise the reviewer rejected; on the cap path it is the
	// oldest messages evicted to honor the total cap. Each is recorded to the
	// OnPruned audit hook before being dropped, so nothing is lost silently.
	Pruned []UnifiedMessage
}

// FlowGateOptions configures a per-account ingress gate. Every limit is
// independent and a zero value disables that limit, so an account that does not
// declare the new fields keeps its previous unlocked behavior (backward
// compatible, design/12 §6 实现要点).
//
// **计数键 = 会话级**（裁决 06.4.3）：GroupMsgPerMin 与 MaxMessages 都按
// ConversationKey（`platform+channelID+groupID/peerID`）每个群聊/私聊独立计数，不再在
// 适配器顶层汇总——两个群并发互不挤占。MsgPerMin 是防刷屏的**账号级总量护栏**，按设计
// 口径仍留在适配器顶层，不随会话拆分。
type FlowGateOptions struct {
	// GroupMsgPerMin caps group-chat inbound messages per minute **per
	// conversation** (design/12 §6 策略三 / 裁决 06.4.3). Zero disables the limit.
	GroupMsgPerMin int
	// MsgPerMin caps total inbound non-owner messages per minute across the whole
	// account — the anti-flood total guard. Zero disables it.
	MsgPerMin int
	// MaxMessages is the **per-conversation** pending-message cap (design/12 §6
	// 策略四 / 裁决 06.4.3). When a conversation's pending set nears it the Reviewer
	// is invoked to prune noise; with no Reviewer the oldest messages are evicted
	// FIFO. Zero disables the cap.
	MaxMessages int
	// Window is the sliding window over which the per-minute limits are counted.
	// Zero uses defaultRateWindow.
	Window time.Duration
	// Review is the optional review model. When nil the gate enforces the cap with
	// FIFO eviction instead of model judgment.
	Review Reviewer
	// OnPruned is called once per pruned/dropped message with a short reason token
	// ("review_pruned" or "capped") so the caller can record an audit trail.
	// It may be nil (then the caller only sees Pruned on the verdict).
	OnPruned func(UnifiedMessage, string)
	// Now overrides the clock for tests. Nil uses time.Now.
	Now func() time.Time
}

const (
	// defaultRateWindow is the sliding window over which per-minute limits count.
	defaultRateWindow = time.Minute
	// defaultReviewPct is how full the pending set must be before the review model
	// runs. Design/12 §6 策略四 says "maxMessages 的 80%": this is the cost guard
	// that keeps review from firing on every message.
	defaultReviewPct = 80
	// FlowLimitDefault is the protective default of 6 group messages per minute
	// (design/12 §6 策略三). It is applied when an account pins no group rate limit.
	FlowLimitDefault = 6
	// FlowCapDefault is the protective default total-cap of 50 pending messages
	// (design/12 §6 策略四).
	FlowCapDefault = 50
	// sessionGCInterval is how many admissions pass between idle-conversation sweeps,
	// so a long-lived gate that has seen many conversations does not grow unbounded.
	sessionGCInterval = 256
)

// ConversationKey 返回一条消息所属会话的节流键（裁决 06.4.3）：
// `platform+channelID+groupID/peerID`——群聊按 (平台, 群) 独立计数，私聊按 (平台, 对端)
// 独立计数。它同时被入站滑窗（本文件的 FlowGate）与出处参与引擎
// (channelbridge.engageKey) 使用，使「每会话独立计数」只有一处口径。
// 多账号（platform:confId）之间的隔离由每个账号各自的 FlowGate/适配器实例承担。
func ConversationKey(msg UnifiedMessage) string {
	conv := msg.GroupID
	if msg.IsGroup() {
		if conv == "" {
			// 群聊缺 group id 时退回 channel id（qq-obv11 里 channelID 就是群号），
			// 避免同一个群被拆成多个桶而让限流失效。
			conv = msg.ChannelID
		}
	} else {
		conv = msg.SenderID
		if conv == "" {
			conv = msg.Sender.UserID
		}
		if conv == "" {
			conv = msg.ChannelID
		}
	}
	return msg.Platform + "|" + msg.ChannelID + "|" + conv
}

// flowSession 是一个会话（群聊/私聊）独立的节流状态。FlowGate 按 ConversationKey 为
// 每个会话保存一份，因此两个群并发互不挤占（裁决 06.4.3）。
type flowSession struct {
	groupHits  []time.Time
	buffer     []queuedMsg
	lastReview time.Time
}

// FlowGate is a concurrency-safe, per-account ingress gate implementing the two
// measures from design/12 §6 策略三/四:
//
//   - 策略三: a sliding-window rate limit on group chat (and, optionally, all
//     non-owner messages). Over-limit messages are dropped (downgraded) rather
//     than blocking a flood of messages from saturating the loop.
//   - 策略四: a bounded pending-message set. When it nears its cap the Reviewer is
//     invoked to prune low-value messages; with no Reviewer the oldest messages are
//     evicted FIFO. Either way every dropped message is audited.
//
// 两个策略的计数都按会话独立（裁决 06.4.3），只有 MsgPerMin 总量护栏按账号汇总。
//
// Owner-private messages are not routed through the gate (the caller bypasses it),
// so the owner is never throttled or pruned. Identity is never examined here.
type FlowGate struct {
	groupPerMin int
	msgPerMin   int
	maxMessages int
	reviewAt    int
	window      time.Duration
	review      Reviewer
	onPruned    func(UnifiedMessage, string)
	now         func() time.Time

	mu        sync.Mutex
	totalHits []time.Time
	sessions  map[string]*flowSession
	admitted  uint64
	seq       uint64
}

// queuedMsg wraps a buffer entry with a stable sequence so pruning decisions can
// be audited and the fate of the just-admitted message be tracked across index
// shifts (value equality on UnifiedMessage is unsafe because it holds slices).
type queuedMsg struct {
	seq uint64
	msg UnifiedMessage
}

// NewFlowGate builds a gate from opts. Options that are zero disable their limit,
// except Window which defaults to one minute when left zero.
func NewFlowGate(opts FlowGateOptions) *FlowGate {
	g := &FlowGate{
		groupPerMin: opts.GroupMsgPerMin,
		msgPerMin:   opts.MsgPerMin,
		maxMessages: opts.MaxMessages,
		window:      opts.Window,
		review:      opts.Review,
		onPruned:    opts.OnPruned,
		now:         opts.Now,
		sessions:    make(map[string]*flowSession),
	}
	if g.window <= 0 {
		g.window = defaultRateWindow
	}
	if g.maxMessages > 0 {
		g.reviewAt = g.maxMessages * defaultReviewPct / 100
		if g.reviewAt < 1 {
			g.reviewAt = 1
		}
	}
	if g.now == nil {
		g.now = time.Now
	}
	return g
}

// Admit applies the rate limits then the per-conversation cap/review gate to one
// inbound message. It returns a verdict telling the caller whether to proceed;
// messages that the gate drops are returned (and audited) so nothing is lost
// silently.
//
// 计数按会话独立（裁决 06.4.3）：群聊滑窗与待处理上限都落在 ConversationKey 对应的
// flowSession 上（两个群并发互不挤占），只有 MsgPerMin 这一账号级总量护栏仍按适配器汇总。
func (g *FlowGate) Admit(ctx context.Context, msg UnifiedMessage) GateVerdict {
	now := g.now()
	key := ConversationKey(msg)
	g.mu.Lock()
	defer g.mu.Unlock()
	g.maybeGCLocked(now)

	// 策略三（会话级）: sliding-window rate limits. 群聊上限按每个群独立计数；MsgPerMin
	// 是账号级的非主人消息总量护栏。Over-limit messages are dropped at the door
	// (downgraded) instead of blocking.
	if g.groupPerMin > 0 && msg.ChannelType == ChannelGroup {
		st := g.sessionLocked(key)
		st.groupHits = g.trimHits(st.groupHits, now)
		if len(st.groupHits) >= g.groupPerMin {
			return g.drop("rate_limited", msg)
		}
		st.groupHits = append(st.groupHits, now)
	}
	if g.msgPerMin > 0 {
		g.totalHits = g.trimHits(g.totalHits, now)
		if len(g.totalHits) >= g.msgPerMin {
			return g.drop("rate_limited", msg)
		}
		g.totalHits = append(g.totalHits, now)
	}

	if g.maxMessages <= 0 {
		return GateVerdict{Proceed: true}
	}

	// 策略四（会话级）: add to this conversation's pending set, then prune when it
	// nears the cap.
	st := g.sessionLocked(key)
	g.seq++
	seq := g.seq
	st.buffer = append(st.buffer, queuedMsg{seq: seq, msg: msg})

	var dropped []queuedMsg
	if g.review != nil && len(st.buffer) >= g.reviewAt && !g.reviewRecently(st, now) {
		keep, err := g.review(ctx, queueMessages(st.buffer))
		if err == nil {
			st.buffer, dropped = g.applyReview(st.buffer, keep)
			st.lastReview = now
			g.auditDropped(dropped, "review_pruned")
		}
	}
	if len(st.buffer) > g.maxMessages {
		n := len(st.buffer) - g.maxMessages
		var evicted []queuedMsg
		st.buffer, evicted = g.dropFront(st.buffer, n)
		dropped = append(dropped, evicted...)
		g.auditDropped(evicted, "capped")
	}

	if !bufferHasSeq(st.buffer, seq) {
		reason := "capped"
		if containsSeq(dropped, seq) {
			reason = "review_pruned"
		}
		v := GateVerdict{Proceed: false, Reason: reason, Pruned: queueMessages(dropped)}
		c := msg
		v.DroppedMsg = &c
		return v
	}
	return GateVerdict{Proceed: true, Pruned: queueMessages(dropped)}
}

// drop builds a refused verdict from a single message that never entered the
// buffer (rate-limited at the door), so the audit hook still sees it.
func (g *FlowGate) drop(reason string, msg UnifiedMessage) GateVerdict {
	c := msg
	v := GateVerdict{Proceed: false, Reason: reason, DroppedMsg: &c, Pruned: []UnifiedMessage{msg}}
	g.auditMsg(msg, reason)
	return v
}

// sessionLocked 返回 key 对应的会话节流状态，首次使用时惰性创建。调用方须持有 g.mu。
func (g *FlowGate) sessionLocked(key string) *flowSession {
	if g.sessions == nil {
		g.sessions = make(map[string]*flowSession)
	}
	st := g.sessions[key]
	if st == nil {
		st = &flowSession{}
		g.sessions[key] = st
	}
	return st
}

// maybeGCLocked 每 sessionGCInterval 次放行清理一次空闲会话，避免会话状态表随历史会话
// 无界增长：滑窗时间戳全部过期且待处理队列为空的会话直接删除。调用方须持有 g.mu。
func (g *FlowGate) maybeGCLocked(now time.Time) {
	g.admitted++
	if g.admitted%sessionGCInterval != 0 {
		return
	}
	for key, st := range g.sessions {
		st.groupHits = g.trimHits(st.groupHits, now)
		if len(st.groupHits) == 0 && len(st.buffer) == 0 {
			delete(g.sessions, key)
		}
	}
}

// reviewRecently throttles review so it does not run on every message: it fires at
// most once per window per conversation, which is the cost guard for the review model.
func (g *FlowGate) reviewRecently(st *flowSession, now time.Time) bool {
	if st == nil || st.lastReview.IsZero() {
		return false
	}
	return now.Sub(st.lastReview) < g.window
}

// trimHits drops event timestamps older than the sliding window.
func (g *FlowGate) trimHits(hits []time.Time, now time.Time) []time.Time {
	cutoff := now.Add(-g.window)
	keep := 0
	for _, t := range hits {
		if t.After(cutoff) {
			hits[keep] = t
			keep++
		}
	}
	return hits[:keep]
}

// applyReview keeps only the entries at the returned indices (validated, de-duped,
// ascending) and returns the kept slice plus the dropped entries in original order.
func (g *FlowGate) applyReview(buffer []queuedMsg, keep []int) (kept, dropped []queuedMsg) {
	seen := make(map[int]bool, len(keep))
	for _, i := range keep {
		if i >= 0 && i < len(buffer) {
			seen[i] = true
		}
	}
	out := make([]queuedMsg, 0, len(buffer))
	drop := make([]queuedMsg, 0)
	for i, entry := range buffer {
		if seen[i] {
			out = append(out, entry)
		} else {
			drop = append(drop, entry)
		}
	}
	return out, drop
}

// dropFront removes n entries from the front (oldest) and returns the rest plus the
// dropped ones.
func (g *FlowGate) dropFront(buffer []queuedMsg, n int) (kept, dropped []queuedMsg) {
	if n >= len(buffer) {
		return nil, append([]queuedMsg(nil), buffer...)
	}
	return buffer[n:], append([]queuedMsg(nil), buffer[:n]...)
}

// bufferHasSeq reports whether an entry with the given sequence is still buffered
// in the conversation's pending set.
func bufferHasSeq(buffer []queuedMsg, seq uint64) bool {
	for _, e := range buffer {
		if e.seq == seq {
			return true
		}
	}
	return false
}

// containsSeq reports whether seq appears among the dropped entries.
func containsSeq(dropped []queuedMsg, seq uint64) bool {
	for _, e := range dropped {
		if e.seq == seq {
			return true
		}
	}
	return false
}

// auditDropped records every dropped entry to the OnPruned audit hook.
func (g *FlowGate) auditDropped(dropped []queuedMsg, reason string) {
	for _, e := range dropped {
		g.auditMsg(e.msg, reason)
	}
}

// auditMsg records one dropped message to the OnPruned audit hook.
func (g *FlowGate) auditMsg(msg UnifiedMessage, reason string) {
	if g.onPruned != nil {
		g.onPruned(msg, reason)
	}
}

// queueMessages extracts the payload messages of queued entries in order.
func queueMessages(queue []queuedMsg) []UnifiedMessage {
	out := make([]UnifiedMessage, 0, len(queue))
	for _, e := range queue {
		out = append(out, e.msg)
	}
	return out
}