package gateway

import (
	"context"
	"encoding/json"
	"net/http"
	"testing"
	"time"

	"github.com/gorilla/websocket"

	"git.yeij.top/AskaEth/Cyrene_v2/internal/auth"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/logger"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace"
)

// 本文件锁住 05.4.3 + 16.4.3-WS（可观测面 WS 推送，替换轮询）：
//
//   - 帧壳与载荷：observability_log / observability_trace / observability_metrics
//     三个帧类型经 ws_bridge 的主题映射成形，data 与既有 REST 端点同形；
//   - 角色门：三者都不是 design/11 §7.1 的状态机信号，因此 fail-closed，
//     只发 owner + admin，viewer 收不到（新增一帧忘记分类也不会泄露）；
//   - 推送时机：只有建立了订阅（面板读取过一次实时数据）才启动推送循环 ——
//     这是「推送替代轮询」不把成本转嫁给并发面板的那一半，也保证未订阅的
//     连接不会凭空收到周期性帧。
//
// 断言都走真路径：真 bus Publish（或真订阅 + 真推送循环）→ hub 的真订阅与中继
// → 真 /ws/chat 连接（真 JWT）。

// obsPushReadBudget 是等待推送帧的总预算。推送循环的节拍是
// observabilityPushInterval，一个节拍内应到齐全部三类帧。
const obsPushReadBudget = 6 * time.Second

// TestObservabilityFramesReachOwnerAndAdminOnly 在真实主题上发布三类可观测
// 载荷：owner/admin 收到全部三帧且载荷键集与登记的一致，viewer 只收到白名单
// 的哨兵帧。
func TestObservabilityFramesReachOwnerAndAdminOnly(t *testing.T) {
	t.Parallel()
	for _, typ := range []string{wsObservabilityLog, wsObservabilityTrace, wsObservabilityMetrics} {
		if frameReachesAll(typ) {
			t.Fatalf("%s reached every connection; an observability frame carries content and must stay owner/admin only", typ)
		}
	}

	env := newEnv(t, nil, false)
	conns := map[auth.Role]*websocket.Conn{}
	for _, role := range []auth.Role{auth.RoleOwner, auth.RoleAdmin, auth.RoleViewer} {
		conn := dialWS(t, env.ts.URL, env.token(t, role))
		defer conn.Close()
		conns[role] = conn
	}
	waitConnCount(t, env.srv, len(conns))

	logs := AdminLogPush{Entries: []AdminLogEntry{
		{Time: "2026-10-04T10:00:01Z", Level: "warn", Source: "gateway/observability", Msg: "推送测试"},
	}}
	spans := AdminTracePush{Spans: []AdminTraceSpan{
		{ID: "span-push-1", Name: "llm.call", Kind: "llm_call", Status: "ok",
			StartedAt: "2026-10-04T10:00:02Z", EndedAt: "2026-10-04T10:00:03Z", DurationNs: 1234},
	}}
	metrics := AdminMetrics{
		TotalRequests: 7, QPS: 1, AvgLatencyMs: 5, MaxLatencyMs: 9,
		PerMethod: map[string]int64{"GET": 7}, PerPath: map[string]int64{"/api/v1/x": 7},
		StatusCounts: map[string]int64{"200": 7}, LLMCalls: 2, SubagentEvents: 1,
	}
	for _, pub := range []struct {
		subject string
		payload any
	}{
		{obsLogSubject, logs},
		{obsTraceSubject, spans},
		{obsMetricsSubject, metrics},
	} {
		if err := env.bus.Publish(context.Background(), pub.subject, pub.payload); err != nil {
			t.Fatalf("publish %s: %v", pub.subject, err)
		}
	}
	// 哨兵：一个白名单状态信号收尾，使「viewer 什么都没收到」不会与
	// 「连接根本没工作」混淆（与既有帧作用域用例同一手法）。
	if err := env.bus.Publish(context.Background(), wsPrefix+scopeSentinel, map[string]any{"sessionId": "web"}); err != nil {
		t.Fatalf("publish sentinel: %v", err)
	}

	for _, role := range []auth.Role{auth.RoleOwner, auth.RoleAdmin} {
		frames := drainFrames(t, conns[role])
		for _, typ := range []string{wsObservabilityLog, wsObservabilityTrace, wsObservabilityMetrics} {
			assertFrameCount(t, string(role), frames, typ, 1)
			if frame := findFrame(t, string(role), frames, typ); frame.SessionID != "" {
				t.Fatalf("%s %s frame sessionId = %q, want none: observability frames are not session-scoped",
					role, typ, frame.SessionID)
			}
		}
		if len(frames) != 4 {
			t.Fatalf("%s received %d frames, want the three observability frames + the sentinel: %v",
				role, len(frames), frameTypesOf(frames))
		}
		assertLogPushFrame(t, string(role), findFrame(t, string(role), frames, wsObservabilityLog), logs)
		assertTracePushFrame(t, string(role), findFrame(t, string(role), frames, wsObservabilityTrace), spans)
		assertMetricsPushFrame(t, string(role), findFrame(t, string(role), frames, wsObservabilityMetrics), metrics)
	}

	viewerFrames := drainFrames(t, conns[auth.RoleViewer])
	if len(viewerFrames) != 1 || viewerFrames[0].Type != scopeSentinel {
		t.Fatalf("viewer received %v, want only the whitelisted %q: observability frames are owner/admin only (design/11 §7.1)",
			frameTypesOf(viewerFrames), scopeSentinel)
	}
}

// TestObservabilityPushStreamsToSubscribers 驱动真实的生产者路径：面板的引导
// 读取建立订阅 → 推送循环取锚点 → 之后新产生的日志 / 指标 / span 各发一帧，
// owner 收得到；同一次推送里 viewer 一帧都收不到（角色门）。
func TestObservabilityPushStreamsToSubscribers(t *testing.T) {
	t.Parallel()
	env := newEnv(t, nil, false)
	ownerToken := env.token(t, auth.RoleOwner)
	viewerToken := env.token(t, auth.RoleViewer)
	owner := dialWS(t, env.ts.URL, ownerToken)
	defer owner.Close()
	viewer := dialWS(t, env.ts.URL, viewerToken)
	defer viewer.Close()
	waitConnCount(t, env.srv, 2)

	ring := logger.NewRingBuffer(64)
	env.srv.SetLogBuffer(ring)

	// 面板挂载时的引导读取：既拿首屏，也建立订阅（watchObservability）。
	resp, _ := doReq(t, http.MethodGet, env.ts.URL+"/api/v1/admin/observability/logs?limit=5", nil, ownerToken)
	if resp.StatusCode != http.StatusOK {
		t.Fatalf("bootstrap log read = %d, want 200: the read must keep serving the guide data", resp.StatusCode)
	}

	// 等循环起步（它在启动时取锚点：只发订阅之后的数据）。
	time.Sleep(300 * time.Millisecond)

	ring.Append(logger.Entry{Time: time.Now(), Level: "warn", Source: "gateway/push-test", Msg: "订阅后的新日志"})
	env.srv.recordMetric(http.MethodGet, "/api/v1/admin/push-test", http.StatusOK, 4)
	env.srv.ObserveSpan(trace.SpanRecord{
		ID: "span-sub", Name: "llm.call", Kind: trace.SpanKindLLM, Status: trace.SpanStatusOK,
		StartedAt: time.Now(), EndedAt: time.Now(), Duration: 3 * time.Millisecond,
	})

	wanted := []string{wsObservabilityLog, wsObservabilityTrace, wsObservabilityMetrics}
	frames := readFramesUntil(t, owner, wanted, obsPushReadBudget)
	for _, typ := range wanted {
		assertFrameCount(t, "owner", frames, typ, 1)
	}
	logFrame := findFrame(t, "owner", frames, wsObservabilityLog)
	entries := decodeFrameEntries(t, "owner", logFrame)
	if len(entries) != 1 || entries[0].Msg != "订阅后的新日志" || entries[0].Source != "gateway/push-test" || entries[0].Level != "warn" {
		t.Fatalf("pushed log entry = %+v, want the appended entry", entries)
	}
	var traceData struct {
		Spans []AdminTraceSpan `json:"spans"`
	}
	unmarshalFrameData(t, "owner", findFrame(t, "owner", frames, wsObservabilityTrace), &traceData)
	if len(traceData.Spans) != 1 || traceData.Spans[0].ID != "span-sub" || traceData.Spans[0].Kind != "llm_call" {
		t.Fatalf("pushed spans = %+v, want the observed span", traceData.Spans)
	}
	var pushedMetrics AdminMetrics
	unmarshalFrameData(t, "owner", findFrame(t, "owner", frames, wsObservabilityMetrics), &pushedMetrics)
	if pushedMetrics.TotalRequests < 1 || pushedMetrics.LLMCalls < 1 {
		t.Fatalf("pushed metrics = %+v, want the recorded request and the observed LLM span", pushedMetrics)
	}

	// 同一次推送里 viewer 依旧一帧都拿不到：门在 hub，与订阅无关。
	viewerFrames := readFramesUntil(t, viewer, wanted, 400*time.Millisecond)
	for _, typ := range wanted {
		assertFrameCount(t, "viewer", viewerFrames, typ, 0)
	}
}

// TestObservabilityPushRequiresSubscription 是推送时机的反证：连接上没有任何
// 人读取过实时数据 → 没有订阅 → 推送循环根本不启动，即使数据在连接建立后
// 立刻变化，也不会凭空出现周期性帧（这正是既有「精确帧数」用例得以保持的
// 前提，也是推送替代轮询不把成本转嫁给所有连接的那一半）。
func TestObservabilityPushRequiresSubscription(t *testing.T) {
	t.Parallel()
	env := newEnv(t, nil, false)
	token := env.token(t, auth.RoleOwner)
	conn := dialWS(t, env.ts.URL, token)
	defer conn.Close()
	waitConnCount(t, env.srv, 1)

	ring := logger.NewRingBuffer(16)
	env.srv.SetLogBuffer(ring)
	ring.Append(logger.Entry{Time: time.Now(), Level: "info", Source: "gateway/no-sub", Msg: "无人订阅"})
	env.srv.recordMetric(http.MethodGet, "/api/v1/admin/no-sub", http.StatusOK, 1)
	env.srv.ObserveSpan(trace.SpanRecord{
		ID: "span-no-sub", Name: "llm.call", Kind: trace.SpanKindLLM, Status: trace.SpanStatusOK,
		StartedAt: time.Now(), EndedAt: time.Now(), Duration: time.Millisecond,
	})

	// 覆盖一个完整节拍：未订阅时这里必须一帧都没有。
	time.Sleep(observabilityPushInterval + 200*time.Millisecond)
	frames := drainFrames(t, conn)
	if len(frames) != 0 {
		t.Fatalf("an unsubscribed connection received %v; the push loop must not run without a subscription", frameTypesOf(frames))
	}
}

// assertLogPushFrame 断言一帧 observability_log 的 data 形状（键集恰好
// {"entries"}）与条目内容。
func assertLogPushFrame(t *testing.T, role string, frame subagentFrame, want AdminLogPush) {
	t.Helper()
	assertFrameDataKeys(t, role, frame, []string{"entries"})
	entries := decodeFrameEntries(t, role, frame)
	if len(entries) != len(want.Entries) {
		t.Fatalf("%s log entries = %d, want %d", role, len(entries), len(want.Entries))
	}
	for i, e := range entries {
		if e != want.Entries[i] {
			t.Fatalf("%s log entry[%d] = %+v, want %+v", role, i, e, want.Entries[i])
		}
	}
}

// decodeFrameEntries 解出 observability_log 帧的条目，并锁住每条目的键集。
func decodeFrameEntries(t *testing.T, role string, frame subagentFrame) []AdminLogEntry {
	t.Helper()
	var data struct {
		Entries []json.RawMessage `json:"entries"`
	}
	unmarshalFrameData(t, role, frame, &data)
	out := make([]AdminLogEntry, 0, len(data.Entries))
	for _, raw := range data.Entries {
		assertRawObjectKeys(t, role+" log entry", raw, []string{"time", "level", "source", "msg"})
		var e AdminLogEntry
		if err := json.Unmarshal(raw, &e); err != nil {
			t.Fatalf("%s log entry %s: %v", role, raw, err)
		}
		out = append(out, e)
	}
	return out
}

// assertTracePushFrame 断言一帧 observability_trace 的 data 形状（键集恰好
// {"spans"}）与 span 内容。
func assertTracePushFrame(t *testing.T, role string, frame subagentFrame, want AdminTracePush) {
	t.Helper()
	var data struct {
		Spans []json.RawMessage `json:"spans"`
	}
	assertFrameDataKeys(t, role, frame, []string{"spans"})
	unmarshalFrameData(t, role, frame, &data)
	if len(data.Spans) != len(want.Spans) {
		t.Fatalf("%s spans = %d, want %d", role, len(data.Spans), len(want.Spans))
	}
	assertRawObjectKeys(t, role+" span", data.Spans[0],
		[]string{"id", "name", "kind", "status", "startedAt", "endedAt", "durationNs"})
	var sp AdminTraceSpan
	if err := json.Unmarshal(data.Spans[0], &sp); err != nil {
		t.Fatalf("%s span %s: %v", role, data.Spans[0], err)
	}
	if sp.ID != want.Spans[0].ID || sp.Name != want.Spans[0].Name || sp.Kind != want.Spans[0].Kind ||
		sp.Status != want.Spans[0].Status || sp.DurationNs != want.Spans[0].DurationNs {
		t.Fatalf("%s span = %+v, want %+v", role, sp, want.Spans[0])
	}
}

// assertMetricsPushFrame 断言一帧 observability_metrics 的 data 就是 REST 端点
// 那份 AdminMetrics（键集逐个锁定），避免推送与轮询两条路给出不同形状。
func assertMetricsPushFrame(t *testing.T, role string, frame subagentFrame, want AdminMetrics) {
	t.Helper()
	wantKeys := []string{
		"totalRequests", "qps", "avgLatencyMs", "maxLatencyMs", "perMethod",
		"perPath", "statusCounts", "llmCalls", "subagentEvents",
	}
	assertFrameDataKeys(t, role, frame, wantKeys)
	var got AdminMetrics
	unmarshalFrameData(t, role, frame, &got)
	if got.TotalRequests != want.TotalRequests || got.QPS != want.QPS ||
		got.AvgLatencyMs != want.AvgLatencyMs || got.MaxLatencyMs != want.MaxLatencyMs ||
		got.LLMCalls != want.LLMCalls || got.SubagentEvents != want.SubagentEvents ||
		got.PerPath["/api/v1/x"] != 7 || got.StatusCounts["200"] != 7 {
		t.Fatalf("%s metrics = %+v, want %+v (REST shape)", role, got, want)
	}
}

// assertFrameDataKeys 断言帧的 data 是键集恰好为 want 的 JSON 对象。
func assertFrameDataKeys(t *testing.T, role string, frame subagentFrame, want []string) {
	t.Helper()
	assertRawObjectKeys(t, role+" frame data", frame.Data, want)
}

// assertRawObjectKeys 断言 raw 是键集恰好为 want 的 JSON 对象。
func assertRawObjectKeys(t *testing.T, what string, raw json.RawMessage, want []string) {
	t.Helper()
	var keys map[string]json.RawMessage
	if err := json.Unmarshal(raw, &keys); err != nil {
		t.Fatalf("%s is not a JSON object: %v (%s)", what, err, raw)
	}
	if len(keys) != len(want) {
		t.Fatalf("%s keys = %v, want exactly %v (api-client 登记的形状)", what, keys, want)
	}
	for _, k := range want {
		if _, ok := keys[k]; !ok {
			t.Fatalf("%s lacks key %q: %v", what, k, keys)
		}
	}
}

// unmarshalFrameData 把帧的 data 解进 out。
func unmarshalFrameData(t *testing.T, role string, frame subagentFrame, out any) {
	t.Helper()
	if err := json.Unmarshal(frame.Data, out); err != nil {
		t.Fatalf("%s frame data %s: %v", role, frame.Data, err)
	}
}

// readFramesUntil 读连接直到 wanted 里的每个帧类型都至少出现一次，或预算耗尽。
// 它返回已收到的帧（超时也返回），使「帧没来」的失败信息里能看到到底来了什么。
func readFramesUntil(t *testing.T, conn *websocket.Conn, wanted []string, budget time.Duration) []subagentFrame {
	t.Helper()
	deadline := time.Now().Add(budget)
	out := make([]subagentFrame, 0, len(wanted))
	for !hasAllFrameTypes(out, wanted) && time.Now().Before(deadline) {
		_ = conn.SetReadDeadline(deadline)
		_, msg, err := conn.ReadMessage()
		if err != nil {
			return out
		}
		var frame subagentFrame
		if uerr := json.Unmarshal(msg, &frame); uerr != nil {
			t.Fatalf("decode frame %s: %v", msg, uerr)
		}
		out = append(out, frame)
	}
	return out
}

// hasAllFrameTypes 报告 frames 是否已经覆盖 wanted 里的每个类型。
func hasAllFrameTypes(frames []subagentFrame, wanted []string) bool {
	for _, typ := range wanted {
		found := false
		for _, f := range frames {
			if f.Type == typ {
				found = true
				break
			}
		}
		if !found {
			return false
		}
	}
	return true
}