package gateway

import (
	"bytes"
	"context"
	"encoding/json"
	"net/http"
	"sync"
	"time"

	"git.yeij.top/AskaEth/Cyrene_v2/internal/auth"
)

// 本文件是裁决 05.4.3 + 16.4.3-WS 的落点：日志 / 指标 / trace 从「前端轮询
// 管理 REST」改为「网关主动推 WS 帧」（design/10 §5 口径）。
//
// 两条边界分开，谁都不越界：
//
//   - **谁能收到帧**：仍由 ws_frame_scope.go 的角色门决定。这三个帧类型没有
//     登记进 frameReachesAll，因此按 fail-closed 默认档只发 owner + admin；
//     viewer 继续走既有 REST 只读档取数（design/11 §7.1）。
//   - **什么时候发帧**：由本文件的订阅登记决定。没有订阅者时推送循环不启动，
//     网关不为可观测数据做任何周期性工作 —— 这正是「推送替代轮询」相对
//     「每个面板各自 1s 轮询」的收益所在。
//
// 订阅从哪来：面板挂载时的首次读取（GET /api/v1/admin/observability/{logs,
// metrics,trace}）兼任订阅握手。上游订阅帧需要 ws_route.go 的上行分发开口，
// 那不在本次可动的文件面内；用「读过实时数据」作显式订阅动作语义等价，且天然
// 只由真正打开过面板的凭据触发。订阅以凭据身份（sub + role + iat，等价于那次
// 登录）为键，其连接全部断开后自动退订。

const (
	// wsObservabilityLog / Trace / Metrics 是三个下行帧类型。帧壳与既有下行
	// 帧一致（wsEvent{type,data}），不带 sessionId：它们不属任何会话。
	wsObservabilityLog     = "observability_log"
	wsObservabilityTrace   = "observability_trace"
	wsObservabilityMetrics = "observability_metrics"

	// 产生这三个帧的 bus 主题。ws_bridge.go 的 eventBridge 把主题映射成上面的
	// 帧类型，因此推送走的是与其它实时事件完全相同的
	// 「bus → relay → 角色门 → 连接」路径（不新开通道、不新开端口）。
	obsLogSubject     = "observability.log"
	obsTraceSubject   = "observability.trace"
	obsMetricsSubject = "observability.metrics"

	// observabilityPushInterval 是推送循环的节拍：每拍最多各发一帧
	// （日志增量 / 新 span / 变化过的指标快照），只发真的有新数据的那类。
	observabilityPushInterval = time.Second
	// maxLogPushBatch / maxSpanPushBatch 界定单帧载荷大小。一拍的突发超过上限
	// 时只发最新的那些 —— 更旧的条目本就落在面板保留窗口之外，与既有 REST 尾部
	// （limit 上限）语义一致，因此不丢「看得见」的数据。
	maxLogPushBatch  = 200
	maxSpanPushBatch = 100
)

// AdminLogPush 是 observability_log 帧的 data：一批日志条目，**最新在前**
// （与 GET /api/v1/admin/observability/logs 同序），面板据此前缀拼接。
type AdminLogPush struct {
	Entries []AdminLogEntry `json:"entries"`
}

// AdminTracePush 是 observability_trace 帧的 data：一批新完成的 span，
// **最新在前**（与 GET /api/v1/admin/observability/trace 同序）。
type AdminTracePush struct {
	Spans []AdminTraceSpan `json:"spans"`
}

// obsWatchKey 是「谁在订阅」的键：凭据身份，等价于那次登录。用 iat 而不用
// token 串，避免把凭据本身留在内存里。
type obsWatchKey struct {
	sub  string
	role auth.Role
	iat  int64
}

// watchKeyOf 把连接/请求的凭据折叠成订阅键。
func watchKeyOf(claims *auth.Claims) obsWatchKey {
	return obsWatchKey{sub: claims.Sub, role: claims.Role, iat: claims.IssuedAt}
}

// obsWatches 是订阅登记表。它与推送循环同生命周期：第一个订阅建立时启动
// 循环，最后一个订阅被剪掉时循环自行退出，因此订阅表空着的时候没有任何
// 常驻 goroutine，也没有任何周期性开销。
type obsWatches struct {
	mu      sync.Mutex
	set     map[obsWatchKey]struct{}
	running bool
}

// newObsWatches 构建一个空的订阅登记表。
func newObsWatches() obsWatches {
	return obsWatches{set: make(map[obsWatchKey]struct{})}
}

// add 登记一个订阅，必要时启动推送循环。没有 bus 就没有下行通道，
// 登记也不会产生任何推送，因此直接跳过（不启动循环）。
func (w *obsWatches) add(s *Server, key obsWatchKey) {
	if s == nil || s.bus == nil {
		return
	}
	w.mu.Lock()
	w.set[key] = struct{}{}
	start := !w.running
	if start {
		w.running = true
	}
	w.mu.Unlock()
	if start {
		go w.loop(s)
	}
}

// loop 是推送循环：每拍先剪掉「凭据已不在线」的订阅，再按需发布三类数据。
// 返回即代表循环已退出（下一个订阅会重新启动它）。
//
// 锚点在循环启动时取定（见 newObsPushState）：只发订阅之后新产生的数据，
// 历史由面板自己那次引导读取提供。
func (w *obsWatches) loop(s *Server) {
	ticker := time.NewTicker(observabilityPushInterval)
	defer ticker.Stop()
	state := newObsPushState(s)
	for range ticker.C {
		if !w.prune(s) {
			return
		}
		state.publish(s)
	}
}

// prune 剪掉没有在线 owner/admin 连接的订阅，并报告循环是否还要继续。
// 「订阅表空了」与「退出」必须在同一把锁里判定：否则一个新订阅可能刚看到
// running=true（以为循环还活着），而循环已经走向退出，推送就此静默停摆。
func (w *obsWatches) prune(s *Server) bool {
	w.mu.Lock()
	defer w.mu.Unlock()
	live := liveWatchKeys(s)
	for k := range w.set {
		if !live[k] {
			delete(w.set, k)
		}
	}
	if len(w.set) == 0 {
		w.running = false
		return false
	}
	return true
}

// liveWatchKeys 返回当前在线 owner/admin 连接的凭据身份集合。
func liveWatchKeys(s *Server) map[obsWatchKey]bool {
	out := make(map[obsWatchKey]bool)
	if s == nil || s.hub == nil {
		return out
	}
	s.hub.mu.RLock()
	defer s.hub.mu.RUnlock()
	for c := range s.hub.conns {
		if c.claims == nil || !ownerAdminClaims(c.claims) {
			continue
		}
		out[watchKeyOf(c.claims)] = true
	}
	return out
}

// watchObservability 把一次可观测读取登记为订阅。只有 owner/admin 会登记：
// viewer 仍走纯 REST 只读档，不被推送（design/10 §5）。读取本身照常返回引导
// 数据，因此面板不需要轮询也能拿到首屏。
func (s *Server) watchObservability(r *http.Request) {
	if s == nil || s.obs == nil || r == nil {
		return
	}
	claims, ok := auth.ClaimsFromContext(r.Context())
	if !ok || claims == nil || !ownerAdminClaims(claims) {
		return
	}
	s.obs.watches.add(s, watchKeyOf(claims))
}

// obsPushState 是推送循环的游标：日志/span 已经发到哪里、上一次发出的指标
// 指纹是什么。它只被循环自己的 goroutine 触碰，因此不需要锁。
type obsPushState struct {
	lastLogTime time.Time
	lastSpanSeq uint64
	lastMetrics []byte
}

// newObsPushState 取定本轮的推送锚点：日志取当前最新条目的时间、trace 取当前
// 写入序号、指标取当前快照的指纹。三者都表示「截至此刻的数据面板已经通过
// REST 引导读取看到过了」，因此循环不会把环形缓冲里的历史整段灌成巨型帧。
func newObsPushState(s *Server) *obsPushState {
	st := &obsPushState{}
	if buf := s.logBuffer(); buf != nil {
		if entries := buf.Snapshot(1); len(entries) == 1 {
			st.lastLogTime = entries[0].Time
		}
	}
	if s.obs != nil {
		st.lastSpanSeq = s.obs.traceBuf.Seq()
		if raw, err := json.Marshal(s.obs.metrics.Snapshot(time.Now())); err == nil {
			st.lastMetrics = raw
		}
	}
	return st
}

// publish 发布本拍新增的数据：日志增量、新 span、变化过的指标快照。
// 三类互相独立 —— 没有新数据的那类不发帧（design/10 §5「推送只在有数据时
// 发一帧」），因此空转的服务端一帧都不发。
func (st *obsPushState) publish(s *Server) {
	st.publishLogs(s)
	st.publishSpans(s)
	st.publishMetrics(s)
}

// publishLogs 把锚点之后的新日志条目发成一帧（最新在前）。日志环形缓冲未注入
// （网关没接组合根的日志尾部）时没有数据源，直接跳过。
func (st *obsPushState) publishLogs(s *Server) {
	buf := s.logBuffer()
	if buf == nil {
		return
	}
	entries := buf.Snapshot(maxLogPushBatch)
	fresh := make([]AdminLogEntry, 0, len(entries))
	for _, e := range entries {
		if !e.Time.After(st.lastLogTime) {
			break // 最新在前：一旦遇到不新的条目，后面的只会更旧
		}
		fresh = append(fresh, adminLogOf(e))
	}
	if len(fresh) == 0 {
		return
	}
	st.lastLogTime = entries[0].Time
	s.publishObservability(obsLogSubject, AdminLogPush{Entries: fresh})
}

// publishSpans 把锚点之后新完成的 span 发成一帧（最新在前）。
func (st *obsPushState) publishSpans(s *Server) {
	if s.obs == nil {
		return
	}
	entries := s.obs.traceBuf.SnapshotSince(st.lastSpanSeq, maxSpanPushBatch)
	if len(entries) == 0 {
		return
	}
	spans := make([]AdminTraceSpan, 0, len(entries))
	for _, e := range entries {
		spans = append(spans, adminSpanOf(e.rec))
	}
	st.lastSpanSeq = entries[0].seq
	s.publishObservability(obsTraceSubject, AdminTracePush{Spans: spans})
}

// publishMetrics 在当前指标快照与上次发出的不同时发一帧。用序列化结果做指纹：
// 快照的 map 副本会被编码，编码是确定性的（键排序），因此「没变」可以精确判定，
// 而 QPS 这类会随时间衰减的量自然在变化时照常推。
func (st *obsPushState) publishMetrics(s *Server) {
	if s.obs == nil {
		return
	}
	snap := s.obs.metrics.Snapshot(time.Now())
	raw, err := json.Marshal(snap)
	if err != nil {
		return
	}
	if bytes.Equal(raw, st.lastMetrics) {
		return
	}
	st.lastMetrics = raw
	s.publishObservability(obsMetricsSubject, snap)
}

// publishObservability 把一个可观测事件发上 bus：它会经 ws_bridge.go 的主题
// 映射变成下行帧，并照常过一遍角色门（本包唯一的可见性决策点）。bus 缺席时
// 是 no-op。
func (s *Server) publishObservability(subject string, payload any) {
	if s == nil || s.bus == nil {
		return
	}
	if err := s.bus.Publish(context.Background(), subject, payload); err != nil {
		s.log.Warn("publish observability frame", "subject", subject, "err", err)
	}
}