package gateway

import (
	"context"
	"encoding/json"
	"fmt"
	"sort"
	"testing"
	"time"

	"github.com/gorilla/websocket"

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

// 这些测试锁住 design/11 §7.1 的帧门：viewer 连接可以跟随会话状态机，
// 但绝不可收到内容，而 owner 与 admin 两类都保留。frameScopeCases 是
// design/11 §7.1 表在代码中的体现 —— 表中点名的每种帧类型，都发布在
// 真正产生它的主题上 —— 因此分类错误（内容帧被公开，或状态信号被
// 私有化）会在这里失败，而不是静默泄露。

// frameScopeCase 是 design/11 §7.1 表的一行。
type frameScopeCase struct {
	typ     string // 下行帧类型
	subject string // 产生它的 bus 主题
	payload any    // 发布在该主题上的载荷
	public  bool   // 「状态机信号」：不含内容 → 每个已认证连接
}

// scopeSentinel 最后发布：bus 每个订阅跑一个 worker 且按序处理，因此
// 已经看到哨兵的连接必定已经被投递过它之前发布的每一个帧。这把
// 「viewer 没有收到别的」从猜测的空闲窗口变成精确计数。
const scopeSentinel = EventSessionIdle

// publishFrameScopeCases 使用的批次节流，远低于 hub 的 wsSendBuffer
// 深度，因此一次突发永不触发慢客户端丢弃。
const (
	scopePaceFrames = 4
	scopePacePause  = 50 * time.Millisecond
)

// scopeRead 是一个连接读取 goroutine 的结果。
type scopeRead struct {
	role  auth.Role
	types []string
	err   error
}

// frameScopeCases 返回 design/11 §7.1 表。转发帧使用它们真实的 ws.<type>
// 主题；流帧发布 loop 在 ws.stream 上发出的真实 model.StreamEvent，因此
// 测试驱动的是生产路径 bus → relay → broadcast，而不是手工构造的帧。
func frameScopeCases() []frameScopeCase {
	relayed := func(typ string, public bool) frameScopeCase {
		return frameScopeCase{typ: typ, subject: wsPrefix + typ, payload: map[string]any{"sessionId": "web"}, public: public}
	}
	streamed := func(kind model.StreamKind, typ string) frameScopeCase {
		return frameScopeCase{
			typ:     typ,
			subject: wsStreamSubject,
			payload: model.StreamEvent{Kind: kind, Delta: "x", SessionID: "web", TurnID: "turn-scope"},
		}
	}
	return []frameScopeCase{
		// 状态机信号（白名单，不含任何内容）→ 所有已认证连接。
		relayed(EventSessionBusy, true),
		relayed(EventPendingAck, true),
		relayed(EventPendingDrain, true),
		relayed(EventSessionIdle, true),
		relayed(EventSessionInterrupt, true),
		relayed(EventPendingFull, true),
		relayed(wsSessionRead, true),

		// 流式内容（含角色回复正文与推理内容）→ 仅 owner/admin。
		streamed(model.StreamKindTextStart, "message.start"),
		streamed(model.StreamKindTextDelta, "message.delta"),
		streamed(model.StreamKindTextEnd, "message.end"),
		streamed(model.StreamKindThinkingStart, "thinking.start"),
		streamed(model.StreamKindThinkingDelta, "thinking.delta"),
		streamed(model.StreamKindThinkingEnd, "thinking.end"),
		streamed(model.StreamKindToolCallStart, "tool_call.start"),
		streamed(model.StreamKindToolCallDelta, "tool_call.delta"),
		streamed(model.StreamKindToolCallEnd, "tool_call.end"),
		streamed(model.StreamKindToolExecutionStart, "tool_execution.start"),
		streamed(model.StreamKindToolExecutionUpdate, "tool_execution.update"),
		streamed(model.StreamKindToolExecutionEnd, "tool_execution.end"),
		streamed(model.StreamKindTurnStart, "turn.start"),
		streamed(model.StreamKindTurnEnd, "turn.end"),
		relayed("stream_chunk", false),
		relayed("stream_end", false),
		relayed("thinking", false),

		// 持久化内容 → 仅 owner/admin。
		relayed("message_persisted", false),
		relayed("message_edited", false),
		relayed("message_deleted", false),
		relayed("channel_persisted", false),
		// 新帧 turn_begin（design/15 §2.5.9）：载荷带 messageIds ⇒ 携带具体消息
		// 身份，落 fail-closed 默认档（只发 owner + admin，与 steer_ack /
		// chat_duplicate 同档），**不**进那 7 个状态机信号白名单。
		relayed(EventTurnBegin, false),

		// 系统内部 → 仅 owner/admin。
		relayed("graph_update", false),
		relayed("subagent_event", false),
		// 后台子代理批次结果：这里断言的是**帧名**（常量，不是字面量，避免下次
		// 再漂移）；它真实的产生主题是 orchestration.SubagentResultSubject
		// （"subagent.result"，带点，那是 bus topic 的拼法），该端点对端路径
		// 由 ws_bridge_subagent_result_test.go 单独驱动。
		relayed(wsSubagentResult, false),
		relayed("task_output", false),
		relayed("task_done", false),
		relayed("automation_result", false),
		relayed("tool_used", false),
		relayed("capacity_awaken", false),
		relayed("mode_switch", false),

		// 私有状态 → 仅 owner/admin。
		relayed("mood", false),
		relayed("device_update", false),
		relayed("reminder", false),
		// 渠道事件实时帧（design/26 §2.6 / §4.5；切片 E7）：载荷是 `channel_events`
		// 的一行（平台 / 账号 / 类别 / 摘要…），属**管理面元数据** —— 不进白名单，
		// 按 fail-closed 默认档只发 owner + admin。它真实的生产主题是
		// `events.SubjectChannelEvent`（"channel.event"，带点），端到端路径由
		// ws_channel_events_test.go 单独驱动；这一行钉的是**帧名**的分类。
		relayed(wsChannelEvent, false),
	}
}

// TestWSFrameScopePerRole 用真实角色 token 经真实 WebSocket 连接驱动
// 整张 design/11 §7.1 表：owner 与 admin 收到表中的每一帧，而 viewer
// 恰好收到白名单状态信号，且按发布顺序。设计点名的 viewer 断言
// （message.delta / thinking.delta / message_persisted / graph_update /
// mood 被拒，session_idle 被投递）在此显式复述，使契约在测试中可读。
func TestWSFrameScopePerRole(t *testing.T) {
	t.Parallel()
	cases := frameScopeCases()
	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))

	all := append(frameTypes(cases, false), scopeSentinel)
	public := append(frameTypes(cases, true), scopeSentinel)
	wants := map[auth.Role]int{
		auth.RoleOwner:  len(all),
		auth.RoleAdmin:  len(all),
		auth.RoleViewer: len(public),
	}

	// 各客户端在批次发布期间于自己的 goroutine 上排空：hub 会丢弃
	// 发送缓冲已满的客户端的帧（其慢客户端策略），因此之后才开始
	// 读取的客户端会丢掉尾部。
	results := make(chan scopeRead, len(conns))
	for role, want := range wants {
		role, want := role, want
		go func() {
			types, err := readFrameTypes(conns[role], want)
			results <- scopeRead{role: role, types: types, err: err}
		}()
	}
	publishFrameScopeCases(t, env, cases)

	got := make(map[auth.Role][]string, len(conns))
	for range conns {
		res := <-results
		if res.err != nil {
			t.Fatalf("%s reader: %v", res.role, res.err)
		}
		got[res.role] = res.types
	}

	for _, role := range []auth.Role{auth.RoleOwner, auth.RoleAdmin} {
		assertFrameTypes(t, string(role), got[role], all)
	}
	gotViewer := got[auth.RoleViewer]
	assertFrameTypes(t, string(auth.RoleViewer), gotViewer, public)
	for _, denied := range []string{"message.delta", "thinking.delta", "message_persisted", "graph_update", "mood"} {
		if containsFrameType(gotViewer, denied) {
			t.Fatalf("viewer received %q; it must not (design/11 §7.1)", denied)
		}
	}
	if !containsFrameType(gotViewer, EventSessionIdle) {
		t.Fatalf("viewer missing %q, a whitelisted state signal (got %v)", EventSessionIdle, gotViewer)
	}
}

// TestHubBroadcastScopeMatrix 是「每种帧类型 × 每种角色」矩阵，直接经
// 投递入口驱动，使表中每一行都被孤立断言（否则泄露的帧难以归因）。
// claims 按角色显式设置，因此默认拒绝的回归会表现为 viewer 丢掉它
// 本应收到的帧。
func TestHubBroadcastScopeMatrix(t *testing.T) {
	t.Parallel()
	hb := newTestHub(nil, nil)
	owner := newFakeConn(hb, 4)
	owner.claims = &auth.Claims{Role: auth.RoleOwner}
	admin := newFakeConn(hb, 4)
	admin.claims = &auth.Claims{Role: auth.RoleAdmin}
	viewer := newFakeConn(hb, 4)
	viewer.claims = &auth.Claims{Role: auth.RoleViewer}
	for _, c := range []*wsConn{owner, admin, viewer} {
		hb.register(c)
	}

	for _, tc := range frameScopeCases() {
		tc := tc
		t.Run(tc.typ, func(t *testing.T) {
			hb.broadcast(wsEvent{Type: tc.typ, SessionID: "web"})
			for _, c := range []*wsConn{owner, admin} {
				if ev := readFakeFrame(t, c); ev.Type != tc.typ {
					t.Fatalf("owner/admin frame = %q, want %q", ev.Type, tc.typ)
				}
			}
			if tc.public {
				if ev := readFakeFrame(t, viewer); ev.Type != tc.typ {
					t.Fatalf("viewer frame = %q, want %q (whitelisted state signal)", ev.Type, tc.typ)
				}
				return
			}
			if err := connHasNoFrame(viewer); err != nil {
				t.Fatalf("viewer received the restricted frame %q: %v", tc.typ, err)
			}
		})
	}
}

// TestFrameScopeAllowsExactlyTheStateSignals 钉住两条使帧门不漂移的
// 性质：未分类的帧类型永不是公开的（因此后续里程碑新增的帧不会因
// 漏分类而泄露），且公开集合恰好是 design/11 §7.1 登记的状态信号
// —— 扩大它意味着要同时改设计、帧门和这张表。
func TestFrameScopeAllowsExactlyTheStateSignals(t *testing.T) {
	t.Parallel()
	if frameReachesAll("frame_type_nobody_classified") {
		t.Fatal("an unclassified frame type reached every connection; the default must be owner/admin only")
	}
	public := make([]string, 0, len(frameScopeCases()))
	for _, tc := range frameScopeCases() {
		if got := frameReachesAll(tc.typ); got != tc.public {
			t.Fatalf("frame %q: gate says public=%v, design/11 §7.1 says public=%v", tc.typ, got, tc.public)
		}
		if tc.public {
			public = append(public, tc.typ)
		}
	}
	want := []string{
		EventSessionBusy,
		EventPendingAck,
		EventPendingDrain,
		EventSessionIdle,
		EventSessionInterrupt,
		EventPendingFull,
		wsSessionRead,
	}
	sort.Strings(public)
	sort.Strings(want)
	assertFrameTypes(t, "design/11 §7.1 whitelist", public, want)
}

// frameTypes 按表顺序返回表中的帧类型（全部，或仅公开的那些）。
func frameTypes(cases []frameScopeCase, publicOnly bool) []string {
	out := make([]string, 0, len(cases))
	for _, tc := range cases {
		if publicOnly && !tc.public {
			continue
		}
		out = append(out, tc.typ)
	}
	return out
}

// publishFrameScopeCases 在每行各自的生产主题上发布它，随后发布关闭
// 批次的哨兵。批次是节流的：发送缓冲已满的客户端其帧会被丢弃
// （hub 有文档的慢客户端策略），因此 37 帧的突发放送会去考验丢弃
// 策略而不是帧门。
func publishFrameScopeCases(t *testing.T, env *testEnv, cases []frameScopeCase) {
	t.Helper()
	for i, tc := range cases {
		if err := env.bus.Publish(context.Background(), tc.subject, tc.payload); err != nil {
			t.Fatalf("publish %s: %v", tc.subject, err)
		}
		if (i+1)%scopePaceFrames == 0 {
			time.Sleep(scopePacePause)
		}
	}
	if err := env.bus.Publish(context.Background(), wsPrefix+scopeSentinel, map[string]any{"sessionId": "web"}); err != nil {
		t.Fatalf("publish sentinel: %v", err)
	}
}

// readFrameTypes 恰好读取 want 个帧并按到达顺序返回其类型。它返回错误
// 而不是让测试失败，以便能跑在批次发布期间排空各客户端的读取
// goroutine 上。
func readFrameTypes(conn *websocket.Conn, want int) ([]string, error) {
	got := make([]string, 0, want)
	for len(got) < want {
		_ = conn.SetReadDeadline(time.Now().Add(3 * time.Second))
		_, msg, err := conn.ReadMessage()
		if err != nil {
			return got, fmt.Errorf("read frame %d/%d: %w (got %v)", len(got)+1, want, err, got)
		}
		var ev wsEvent
		if err := json.Unmarshal(msg, &ev); err != nil {
			return got, fmt.Errorf("decode frame %s: %w", msg, err)
		}
		got = append(got, ev.Type)
	}
	return got, nil
}

// assertFrameTypes 比较收到的帧类型与期望序列。
func assertFrameTypes(t *testing.T, what string, got, want []string) {
	t.Helper()
	if len(got) != len(want) {
		t.Fatalf("%s frame count = %d, want %d (got %v)", what, len(got), len(want), got)
	}
	for i := range want {
		if got[i] != want[i] {
			t.Fatalf("%s frame[%d] = %q, want %q (got %v)", what, i, got[i], want[i], got)
		}
	}
}

// containsFrameType 报告 typ 是否在收到的帧类型之中。
func containsFrameType(types []string, typ string) bool {
	for _, got := range types {
		if got == typ {
			return true
		}
	}
	return false
}