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.
//
// The gate calls it **off the admission path** (see FlowGate.Admit), so an
// implementation may take as long as the model takes; it is bounded by
// reviewTimeout and by the per-window budget, never by the caller's patience.
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. After M24 it is always
	// "rate_limited" (over the sliding window — dropped at the door, the message
	// never enters the pending set), because nothing else on the admission path can
	// refuse a message: the pending cap evicts the **oldest** entries and the review
	// model runs in the background, so neither can reject the message that just
	// arrived. The audit hook still reports their drops with "capped" /
	// "review_pruned"; those are retained-message bookkeeping, not admissions.
	// Reason 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 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.
	// Backlog drops made by a background review reach the audit hook with
	// "review_pruned" when that review completes, not through this field.
	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
	// DirectMsgPerMin caps **direct (private)** inbound messages per minute
	// **per conversation** — and for a private message the conversation key is
	// (platform, channel, sender), so the limit throttles one sender rather than
	// the whole account. It is the mandatory flow control for stranger private
	// chats (主人 2026-10-05 裁决③): strangers chat normally, so a sender must be
	// unable to flood her. Zero disables the limit.
	DirectMsgPerMin 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 — in the background — 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. It is always called from a background
	// goroutine, at most once per conversation per Window, with at most one call in
	// flight per gate.
	Review Reviewer
	// ReviewPerMin is the hard per-account ceiling on Reviewer calls started per
	// Window: the budget that bounds review cost no matter how many conversations
	// cross their own threshold in the same window. Zero (or negative) uses
	// defaultReviewPerWindow. It bounds calls started, not calls finished, so the
	// bound holds independently of model latency.
	ReviewPerMin int
	// 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
	// defaultReviewPerWindow is the hard per-adapter ceiling on reviewer calls per
	// window (6/minute, the same order of magnitude as FlowLimitDefault). Before it
	// existed the only bound was "one per conversation per window", which an account
	// with many conversations could multiply without limit. At p50 5.3 s per call
	// this still allows ~half a minute of review work per minute per account, so it
	// bounds the tail without starving normal traffic.
	defaultReviewPerWindow = 6
	// 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
	// FlowDirectDefault is the protective default of 6 direct (private) messages
	// per minute **per sender** — the mandatory flow control for stranger private
	// chats (主人 2026-10-05 裁决③). 6/分钟与群聊保护性默认同一口径：一个人正常
	// 一来一回打字到不了 6 条/分钟，而刷屏/机器人会在第 7 条被挡下。它不随
	// groupMsgPerMin 变化（陌生人流控不允许被「没配那项」静默关掉）。
	FlowDirectDefault = 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
	directHits []time.Time
	buffer     []queuedMsg
	// lastReview 是**最近一次审查被派发**的时刻（不是完成时刻，也无论成败）：
	// 写在这里而不是「成功之后」，失败路径才同样被窗口节流（修复前它只在
	// err == nil 分支置位，而 68% 的审查因对端回带 BOM 解析失败 ⇒ 缓冲满之后
	// 每条入站帧都付一次 ~6 s 的模型调用）。
	lastReview time.Time
	// inFlight 是该会话**正在跑**的回合数（AcquireTurn/ReleaseTurn 维护）。受限车道
	// 用它把「同一发送者同时最多 1 个在跑的回合」落到实处。
	inFlight int
}

// 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), plus the mandatory **per-sender** rate limit on direct
//     (private) messages that backs the stranger private-chat flow control
//     (AllowDirect). 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 总量护栏按账号汇总。
//
// **审查不在准入路径上**：Admit 只读待处理集合里已经存在的结论（限流计数、
// 上一批已完成审查留下的队列），触发条件满足时只把一次审查**交给后台**，因此
// 适配器的读 goroutine 永远不会等模型 —— 修复前它会同步等一次 `noise_review`
// （实测 p50 5.3 s / p95 6.7 s / max 9.2 s），把消息、心跳与探活回执全排在同一条
// 队上（见 DevDocs/audit/群聊排队与回旧话-2026-10-24.md）。后台审查的三重上界：
// 同一适配器至多一次在飞（reviewInFlight）、同一会话每 Window 至多一次
// （lastReview）、整个适配器每 Window 至多 defaultReviewPerWindow 次（reviewHits）。
//
// 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.
// The gate also carries the per-conversation running-turn slot (AcquireTurn /
// ReleaseTurn) used by the stranger private-chat flow control.
type FlowGate struct {
	groupPerMin  int
	directPerMin int
	msgPerMin    int
	maxMessages  int
	reviewAt     int
	window       time.Duration
	review       Reviewer
	reviewBudget int
	onPruned     func(UnifiedMessage, string)
	now          func() time.Time

	mu        sync.Mutex
	totalHits []time.Time
	// reviewHits 是本适配器在窗口内**派发**审查的时刻，用于账号级预算
	// （defaultReviewPerWindow）。
	reviewHits []time.Time
	// reviewInFlight 表示本适配器已有一次审查在后台跑：同一适配器至多一次在飞，
	// 使后台审查有界（不会因为多个群同时越阈而并发起 N 个模型调用）。
	reviewInFlight bool
	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).
// The sequence also lets a review verdict land on the entries the reviewer actually
// saw, even when the buffer has moved on while the call was in flight.
type queuedMsg struct {
	seq uint64
	msg UnifiedMessage
}

// ReviewAtFor 返回给定待处理上限对应的审查触发水位（design/12 §6 策略四：
// maxMessages 的 80%，至少 1）。它是 NewFlowGate 内部同一处的口径，导出只为让调用方
// **如实报出生效值**（例如 cmd/cyrene 的逐账号入口流控启动日志），而不是在别处另写
// 一遍 80% 的算法。非正上限表示上限关闭，返回 0（永不触发审查）。
func ReviewAtFor(maxMessages int) int {
	if maxMessages <= 0 {
		return 0
	}
	at := maxMessages * defaultReviewPct / 100
	if at < 1 {
		return 1
	}
	return at
}

// 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,
		directPerMin: opts.DirectMsgPerMin,
		msgPerMin:    opts.MsgPerMin,
		maxMessages:  opts.MaxMessages,
		window:       opts.Window,
		review:       opts.Review,
		reviewBudget: opts.ReviewPerMin,
		onPruned:     opts.OnPruned,
		now:          opts.Now,
		sessions:     make(map[string]*flowSession),
	}
	if g.window <= 0 {
		g.window = defaultRateWindow
	}
	if g.reviewBudget <= 0 {
		g.reviewBudget = defaultReviewPerWindow
	}
	g.reviewAt = ReviewAtFor(g.maxMessages)
	if g.now == nil {
		g.now = time.Now
	}
	return g
}

// Admit applies the rate limits then the per-conversation cap 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.
//
// **它从不同步调用审查模型**：限流与上限是 O(1) 的判定，审查只是一次「后台修剪」
// 的派发。修复前这里同步等 `noise_review`（p50 5.3 s），而五路入站里有三路
// （qq-ws、kook-ws、weixin 长轮询）是「单读 goroutine + 同步派发」，于是消息、
// 心跳、探活回执共享同一条 6 s 队列 —— 主人口中「消息一直排队」的机制。
//
// 计数按会话独立（裁决 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()
	g.maybeGCLocked(now)
	verdict, job := g.admitLocked(msg, key, now)
	g.mu.Unlock()

	if job != nil {
		// 派发在解锁之后：后台审查自己会加锁把结论落回待处理集合，绝不能让它
		// 与这次准入抢同一把锁（也绝不能让准入等它）。
		go g.runReview(ctx, job)
	}
	return verdict
}

// admitLocked runs the whole admission decision under g.mu and returns the verdict
// plus, when due, the background review job the caller must launch after unlocking.
// It never calls the Reviewer itself.
func (g *FlowGate) admitLocked(msg UnifiedMessage, key string, now time.Time) (GateVerdict, *reviewJob) {
	// 策略三（会话级）: 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), nil
		}
		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), nil
		}
		g.totalHits = append(g.totalHits, now)
	}

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

	// 策略四（会话级）: add to this conversation's pending set, then enforce the cap
	// synchronously. 审查只被**派发**（下面的 scheduleReviewLocked），结论稍后由
	// 后台落回 —— 准入读的是「当前已有结论」。
	st := g.sessionLocked(key)
	g.seq++
	seq := g.seq
	st.buffer = append(st.buffer, queuedMsg{seq: seq, msg: msg})

	var dropped []queuedMsg
	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")
	}
	job := g.scheduleReviewLocked(st, key, now)

	// 本条一定还在缓冲里：上限按新旧顺序丢最旧，而它是最新的（maxMessages ≥ 1）。
	// 于是修复后准入路径上唯一能让消息 Proceed=false 的判据就是上面的限流 —— 审查
	// （已移到后台）与上限（丢最旧）都只影响**已经放行过的**历史条目，只留审计痕迹。
	return GateVerdict{Proceed: true, Pruned: queueMessages(dropped)}, job
}

// AllowDirect enforces the per-sender sliding-window rate limit for a **direct
// (private)** message (主人 2026-10-05 裁决③：陌生人私聊=正常聊天，因此必须有流控).
// For a private message the conversation key is (platform, channel, sender), so
// the window throttles one sender instead of the whole account and two strangers
// never crowd each other out. It returns false when that sender is over
// DirectMsgPerMin; the refused message is recorded to the OnPruned audit hook so
// nothing is dropped silently. A zero limit or a group message always allows.
func (g *FlowGate) AllowDirect(msg UnifiedMessage) bool {
	if g.directPerMin <= 0 || msg.ChannelType != ChannelDirect {
		return true
	}
	now := g.now()
	key := ConversationKey(msg)
	g.mu.Lock()
	defer g.mu.Unlock()
	st := g.sessionLocked(key)
	st.directHits = g.trimHits(st.directHits, now)
	if len(st.directHits) >= g.directPerMin {
		g.auditMsg(msg, "rate_limited")
		return false
	}
	st.directHits = append(st.directHits, now)
	return true
}

// DirectPerMin 返回当前生效的私聊（非主人）每分钟每发送者上限，0 = 该限制已关闭。
// 调用方据此**如实报出实际额度**（例如 channelbridge 被限流时的告警日志），而不是
// 写死保护性默认 6 —— 主人 2026-10-05 裁决 13 把该额度变成按账号可配
// （`directMsgPerMin`），硬编码会让日志报出一个与真实生效值不符的数字。
// 该值在构造后不再变化，因此这里直接读，无需加锁。
func (g *FlowGate) DirectPerMin() int {
	return g.directPerMin
}

// AcquireTurn reserves the single running-turn slot of a conversation, returning
// false when a turn is already in flight for it. Callers must pair every true
// result with ReleaseTurn. It is the second half of the mandatory stranger flow
// control (主人 2026-10-05 裁决③：「同一发送者同时最多 1 个在跑的回合」): a stranger
// can send several messages inside the per-minute allowance, yet only one of them
// may occupy an expensive agent turn at a time. It never decides identity — the
// key is the same structural ConversationKey the rest of the gate uses.
func (g *FlowGate) AcquireTurn(msg UnifiedMessage) bool {
	key := ConversationKey(msg)
	g.mu.Lock()
	defer g.mu.Unlock()
	st := g.sessionLocked(key)
	if st.inFlight >= 1 {
		return false
	}
	st.inFlight++
	return true
}

// ReleaseTurn releases a slot reserved by AcquireTurn.
func (g *FlowGate) ReleaseTurn(msg UnifiedMessage) {
	key := ConversationKey(msg)
	g.mu.Lock()
	defer g.mu.Unlock()
	st := g.sessionLocked(key)
	if st.inFlight > 0 {
		st.inFlight--
	}
}

// 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)
		st.directHits = g.trimHits(st.directHits, now)
		if len(st.groupHits) == 0 && len(st.directHits) == 0 &&
			len(st.buffer) == 0 && st.inFlight == 0 {
			delete(g.sessions, key)
		}
	}
}

// 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]
}

// 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]...)
}

// 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
}