package channelbridge

// 本文件承载车道的**入队与取队列**（M22 切片 LANEQUEUE；M23 加了 batch 车道的容量与
// 回插队首）。它是「读与回水解耦」的那一侧：读循环只投递，worker 只按 FIFO 取。
//
// 三条不变量落在这里：
//   - **有界且诚实**：群聊/陌生人满时在门前拒收**这一条**（队列里的条目一条都不动），
//     主人私聊永不拒收（溢到 spill 只排队，按倍率告警）；
//   - **FIFO 不可超车**：spill 一旦非空就一直追加 spill；front（回插区）里的一定早于
//     ch 与 spill 里的，因此读取顺序恒为 front → ch → spill；
//   - **整片取出**：`takeAll` 是静默窗口攒批的唯一取队列入口（只搬不筛、不改序）。

import (
	"context"

	"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter"
)

// enqueueInbound 把一条**已过流控闸**的入站消息投进它所属会话的车道，并立刻返回 ——
// 这是「读与回水解耦」的全部：适配器的读循环调用它之后马上可以继续读 socket。
//
// 🔴 这就是 M23 静默窗口合并的**入队 seam**：入参是一条**已过闸门的 prepared item**
// （Validate/ResolveActor/Admit 都已在本函数的调用方 onMessage 里做完，连同解析结果
// actor 一起带进来）。车道头部只做「等（窗口）+ 发车」，准入判断不在这里。
//
// 三件事在这里一次性做完：
//   - 认领车道（没有就建一条并起 worker）—— **类型由 actor 元数据决定**：
//     主人私聊 ⇒ direct（无窗口），其余 ⇒ batch（有窗口）；
//   - 入队（群聊/陌生人满则**在门前拒收这一条**并留痕；主人私聊**永不拒收**）；
//   - 与 Stop 的 `wg.Wait` 对齐：`b.wg.Add` 与 `closed` 的判定在同一把 `b.mu` 下，
//     因此 Stop 要么看见这个工作项、要么我们看见 closed 而不入队，二者必居其一。
//
// 锁顺序：全树只有这一处同时取两把锁，顺序恒为 `b.mu` → `lanes.mu`。
func (b *Bridge) enqueueInbound(ctx context.Context, msg adapter.UnifiedMessage, a adapter.PlatformAdapter, actor adapter.Actor) {
	if b.lanes == nil {
		return
	}
	item := laneItem{msg: msg, a: a, actor: actor, arrivedAt: b.now()}
	key := laneKey(msg, actor)
	// 主人私聊（私聊 + 主人在场，由 actor 元数据 + 白名单结构化判定）**永不丢弃**：
	// 队列满时落到溢出区只排队（主人 2026-10-24 裁定：「宁可晚回，不许不回」）。
	neverDrop := ownerPrivateLane(msg, actor)
	spec := b.laneSpecFor(key, item, neverDrop)

	b.mu.Lock()
	if b.closed {
		b.mu.Unlock()
		b.lanes.log.Debug("bridge closed, inbound not queued",
			"platform", msg.Platform, "channel", msg.ChannelID)
		return
	}
	l, start, rejected := b.lanes.offer(key, item, spec)
	if start {
		b.wg.Add(1)
	}
	b.mu.Unlock()

	if start {
		go b.laneWorker(ctx, l)
	}
	if rejected {
		b.dropInbound(item, laneDropReasonOverflow, "lane queue full")
	}
}

// laneSpecFor 决定一条车道的形状：**类型只由 actor 元数据决定**
// （`IsOwner() && !IsGroupChat()` ⇒ direct），窗口参数只影响容量。
//
// 容量：batch 车道至少要能装下一整批，否则「一批 N 条」在结构上不可达
// （队列深度 8 时第 9 条就会在门前被拒收，合并就成了摆设）。因此
// capacity = max(默认深度, 生效的单批上限)；direct 车道用默认深度。
// 内存口径因此是「活跃会话数 × 批上限」，而批上限本身被 clamp 在 [1,50]。
func (b *Bridge) laneSpecFor(key string, item laneItem, neverDrop bool) laneSpec {
	spec := laneSpec{neverDrop: neverDrop, batch: !neverDrop, capacity: b.lanes.depth}
	if neverDrop {
		return spec
	}
	if w, ok := b.coalesceWindowFor(key, laneStrangerPrivate(item)); ok && w.maxBatch > spec.capacity {
		spec.capacity = w.maxBatch
	}
	return spec
}

// offer 在**一个临界区**内完成「查/建车道 + 入队」，使 `retireLaneIfIdle` 不可能夹在
// 中间把刚查到、还没写入的车道摘掉（那样条目会落进一条没有 worker 的死车道）。
//
// 队列满时的策略分两档（都不是静默的）：
//   - **主人私聊**（neverDrop）：绝不拒收，落到溢出区 `spill` 里**只排队**，并按倍率
//     （ownerLaneBacklogWarn 及其翻倍）留 WARN —— 一个白名单账号的积压只反映「她落后了」，
//     不构成第三方洪泛面；
//   - **群聊 / 陌生人**：**在门前拒收这一条**（而不是挤掉队列里已有的某条）。队列里
//     的条目因此一条都不会凭空消失 —— 窗口的 takeAll 依赖这一点；被拒收的那条
//     由调用方留痕（WARN + 计数 + 审计）。
//
// 🔴 **FIFO 的关键**：`spill` 一旦非空，后面的条目**一律继续追加到 spill**，不再回写
// `ch`。否则「ch 腾出一个位置 + 此时正好有新条目到达」会让后到者超车（复审实测顺序
// `[m1 m2 m3 m5 m4]`）。`nextLaneItem` 因此可以安全地「先取完 ch、再按序取 spill」：
// spill 里的一定晚于 ch 里的，而 spill 一旦开始就用到底（直到它被取空）。
//
// 同理，`front`（回插队首区）里的一定**早于** ch 与 spill 里的（只有 worker 自己
// 在发车前把超量的尾部放进去），所以读取顺序恒为 front → ch → spill。
func (s *laneSet) offer(key string, item laneItem, spec laneSpec) (l *lane, start bool, rejected bool) {
	s.mu.Lock()
	defer s.mu.Unlock()
	l = s.lanes[key]
	if l == nil {
		capacity := spec.capacity
		if capacity <= 0 {
			capacity = s.depth
		}
		ch := make(chan laneItem, capacity)
		if spec.batch {
			l = newBatchLane(key, ch, spec.neverDrop, &batchWindow{after: s.after})
		} else {
			l = newDirectLane(key, ch, spec.neverDrop)
		}
		s.lanes[key] = l
		start = true
	}
	if l.neverDrop && len(l.spill) > 0 {
		s.spillLocked(l, item)
		return l, start, false
	}
	select {
	case l.ch <- item:
		return l, start, false
	default:
	}
	if l.neverDrop {
		s.spillLocked(l, item)
		return l, start, false
	}
	return l, start, true
}

// spillLocked 把条目追加进主人私聊车道的溢出区，并按**倍率**告警
// （16/32/64/128… 每翻一倍一条 WARN）：固定水位只会在恰好等于时响一次，之后继续涨就
// 再无信号，而「她落后了多少」正是主人侧唯一可观测的量。调用方必须持有 s.mu。
func (s *laneSet) spillLocked(l *lane, item laneItem) {
	l.spill = append(l.spill, item)
	if n := len(l.spill); n >= ownerLaneBacklogWarn && isPowerOfTwo(n) {
		s.log.Warn("owner private lane backlog is growing, still queueing",
			"backlog", n, "lane", l.key,
			"platform", item.msg.Platform, "channel", item.msg.ChannelID)
	}
}

// isPowerOfTwo 报告 n 是否为 2 的幂（n ≥ 1）。
func isPowerOfTwo(n int) bool { return n > 0 && n&(n-1) == 0 }

// nextLaneItem 按 FIFO 取下一条：先取回插队首区（front），再把缓冲区取完，最后取
// 主人私聊的溢出区（三者的先后由 offer / requeueFront 的写入口径保证）。
// ok=false 表示此刻真的没有待办。
func (b *Bridge) nextLaneItem(l *lane) (laneItem, bool) {
	b.lanes.mu.Lock()
	if len(l.front) > 0 {
		item := l.front[0]
		l.front = l.front[1:]
		b.lanes.mu.Unlock()
		return item, true
	}
	b.lanes.mu.Unlock()
	select {
	case item := <-l.ch:
		return item, true
	default:
	}
	b.lanes.mu.Lock()
	defer b.lanes.mu.Unlock()
	if len(l.spill) == 0 {
		return laneItem{}, false
	}
	item := l.spill[0]
	copy(l.spill, l.spill[1:])
	l.spill = l.spill[:len(l.spill)-1]
	return item, true
}

// takeAll 把车道上**此刻**排着的条目整片取出（顺序 = 到达顺序），供静默窗口的
// 批组装使用。它是合并阶段**唯一**的取队列入口（M23 蓝图 §4-1）：
//
//   - 只有车道的单消费者（该车道唯一的 worker）调用它 ⇒ 不存在两个消费者各抓一半；
//   - 取出即「这批就是这些」，等待期间新到的继续落进队列，下一轮 takeAll 再取；
//   - 它**不**做任何筛选：不丢、不重排、不改内容（窗口内零丢失的结构保证）。
func (b *Bridge) takeAll(l *lane) []laneItem {
	b.lanes.mu.Lock()
	out := l.front
	l.front = nil
	b.lanes.mu.Unlock()
	for {
		select {
		case item := <-l.ch:
			out = append(out, item)
			continue
		default:
		}
		b.lanes.mu.Lock()
		if len(l.spill) > 0 {
			out = append(out, l.spill...)
			l.spill = nil
		}
		b.lanes.mu.Unlock()
		return out
	}
}

// requeueFront 把一批**尚未发车**的条目按原顺序插回队首，成为下一批。
//
// 它是「单批上限不是丢弃」的落点（M23 蓝图 §4-3）：超量的**尾部**原样回插、保序、
// 在锁内、在发车之前。回插到 front 而不是写 ch：ch 有界且写满即拒收，
// 而 front 无界但只由 worker 在发车前写入，写入量与「刚拆出来的尾部」相等。
func (b *Bridge) requeueFront(l *lane, items []laneItem) {
	if len(items) == 0 {
		return
	}
	b.lanes.mu.Lock()
	merged := make([]laneItem, 0, len(items)+len(l.front))
	merged = append(merged, items...)
	merged = append(merged, l.front...)
	l.front = merged
	b.lanes.mu.Unlock()
}

// retireLaneIfIdle 在车道已空时把它从注册表摘掉并报告 true（worker 随即退出）。
//
// 它必须与 `offer` / `requeueFront` 用同一把锁：与 offer 交错的话会出现「offer 拿到
// 车道 → worker 退休并删除 → offer 往一条没有 worker 的车道入队」的丢失。入队与
// 「查空」在同一把锁下，就不存在这个窗口。
func (b *Bridge) retireLaneIfIdle(l *lane) bool {
	b.lanes.mu.Lock()
	defer b.lanes.mu.Unlock()
	if len(l.front) > 0 || len(l.ch) > 0 || len(l.spill) > 0 || b.lanes.lanes[l.key] != l {
		return false
	}
	delete(b.lanes.lanes, l.key)
	return true
}

// retireLane 无条件把车道从注册表摘掉（桥停止时的清理；未处理的条目随取消作废）。
func (b *Bridge) retireLane(l *lane) {
	b.lanes.mu.Lock()
	defer b.lanes.mu.Unlock()
	if b.lanes.lanes[l.key] == l {
		delete(b.lanes.lanes, l.key)
	}
}