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
	// 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 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
	// 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 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 总量护栏按账号汇总。
//
// 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
	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,
		directPerMin: opts.DirectMsgPerMin,
		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)}
}

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

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