package gateway

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

	"github.com/gorilla/websocket"

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

// 本文件锁定 Android 协同 §G 的两条：G2（REST `stream:true` 不得「静默成功」）
// 与 G4（chat 全链路幂等 —— clientMsgId）。
//
// G2 的核心断言：**总线饱和（或不投递）时，经 REST `stream:true` 发起的对话
// 仍然落库并产生帧**。它与 ws_route_chat_test.go 的
// TestWSChatFrameSurvivesSaturatedBus 是同一条原则在另一条入口上的落地：会话
// 消息不得依赖可丢弃的通道（design/15 §2.1）。
//
// G4 的核心断言：同一 clientMsgId 发两次 → 只有一条用户消息、只有一个回合，且
// 第二次的响应明确（既有 messageId）；省略该字段时行为与今天逐字一致。

// postChat 以 owner token POST 一次 /api/v1/chat，返回响应与原始响应体。
func postChat(t *testing.T, url, token string, body map[string]any) (*http.Response, []byte) {
	t.Helper()
	return doReq(t, http.MethodPost, url+"/api/v1/chat", body, token)
}

// dataOf 解出 {data:...} 信封里的 data 对象。
func dataOf(t *testing.T, body []byte) map[string]any {
	t.Helper()
	raw, ok := decodeBody(t, body)["data"]
	if !ok {
		t.Fatalf("响应没有 data 信封：%s", body)
	}
	var out map[string]any
	if err := json.Unmarshal(raw, &out); err != nil {
		t.Fatalf("解码 data 失败：%v（body=%s）", err, body)
	}
	return out
}

// errorCodeOf 解出 {error:{code}} 里的 code。
func errorCodeOf(t *testing.T, body []byte) string {
	t.Helper()
	var eobj struct {
		Code string `json:"code"`
	}
	if err := json.Unmarshal(decodeBody(t, body)["error"], &eobj); err != nil {
		t.Fatalf("解码 error 信封失败：%v（body=%s）", err, body)
	}
	return eobj.Code
}

// frameTypes 汇总见过的帧类型，供失败信息使用（读帧超时时能一眼看出漏了什么）。
func chatFrameTypes(frames []wsEvent) []string {
	out := make([]string, 0, len(frames))
	for _, f := range frames {
		out = append(out, f.Type)
	}
	return out
}

// readFramesUntil 在读到一个 want 类型的帧之前持续读帧（最多 5 秒），返回它
// 见过的**全部**帧。它带读超时，因此断言失败时是明确的报错，而不是挂死。
func readChatFramesUntil(t *testing.T, conn *websocket.Conn, want string) []wsEvent {
	t.Helper()
	deadline := time.Now().Add(5 * time.Second)
	seen := make([]wsEvent, 0, 8)
	for time.Now().Before(deadline) {
		_ = conn.SetReadDeadline(deadline)
		_, msg, err := conn.ReadMessage()
		if err != nil {
			t.Fatalf("等待 %q 时读 WS 帧失败：%v（已见 %v）", want, err, chatFrameTypes(seen))
		}
		var ev wsEvent
		if err := json.Unmarshal(msg, &ev); err != nil {
			t.Fatalf("解码帧 %s：%v", msg, err)
		}
		seen = append(seen, ev)
		if ev.Type == want {
			return seen
		}
	}
	t.Fatalf("等待 %q 超时（已见 %v）", want, chatFrameTypes(seen))
	return nil
}

// hasFrame 报告帧列表里是否出现过某类型。
func hasFrame(frames []wsEvent, typ string) bool {
	for _, f := range frames {
		if f.Type == typ {
			return true
		}
	}
	return false
}

// TestChatStreamSurvivesSaturatedBus 是 G2 的主断言：总线**完全不投递**（饱和
// 丢弃）时，REST `stream:true` 依然驱动回合、落库用户消息与助手回复，并产出下行
// 帧。修复前这一轮是 publishChatStream → bus.Publish → hub 订阅：饱和时事件被
// 丢，客户端却早就拿到了 200 {"stream":true} —— 不落库、没有任何帧。
func TestChatStreamSurvivesSaturatedBus(t *testing.T) {
	t.Parallel()
	run := &recordingRunner{}
	hist := newReplayHistoryStore()
	sb := &saturatedBus{}
	srv, ts, jwt := buildEnv(t, cfgDefault(), Deps{Loop: run, History: hist, Bus: sb})
	tok := signAny(t, jwt, "acct", auth.RoleOwner)

	conn := dialWS(t, ts.URL, tok)
	defer conn.Close()
	waitConnCount(t, srv, 1)

	resp, body := postChat(t, ts.URL, tok, map[string]any{"sessionId": "web", "text": "总线饱和也要到", "stream": true})
	if resp.StatusCode != http.StatusOK {
		t.Fatalf("status = %d, want 200（body=%s）", resp.StatusCode, body)
	}
	if got := dataOf(t, body); got["stream"] != true {
		t.Fatalf("data = %v, want stream:true", got)
	}

	// 200 回给客户端时用户消息已经落库（handler 同步穿过提交与落库），助手回复
	// 随回合结算到达。
	waitUntil(t, "回合被驱动", func() bool { return run.count() == 1 })
	waitUntil(t, "用户+助手消息落库", func() bool { return len(hist.records()) == 2 })
	recs := hist.records()
	if recs[0].Role != "user" || recs[0].Content != "总线饱和也要到" {
		t.Fatalf("用户消息 = %+v，want 刚发的那条", recs[0])
	}
	if recs[1].Role != "assistant" || recs[1].Content == "" {
		t.Fatalf("助手回复 = %+v，want 非空", recs[1])
	}

	// 帧：用户消息的 message_persisted 必然在 stream_end 之前出现过。
	frames := readChatFramesUntil(t, conn, "stream_end")
	if !hasFrame(frames, "message_persisted") {
		t.Fatalf("帧 = %v，want 至少一条 message_persisted（客户端要靠它追加消息）", chatFrameTypes(frames))
	}

	// 会话消息绝不经可丢弃的通道：ws.chat 主题上不应有任何发布。
	for _, subject := range sb.subjects() {
		if subject == wsChatSubject {
			t.Fatalf("REST stream:true 仍把会话消息发布到可丢弃的 bus 主题 %q（design/15 §2.1 禁止）", subject)
		}
	}
}

// TestChatStreamWithoutRunnerIsNotSilentlyAccepted 钉住诚实的退化：没接 loop 的
// 网关（runner 为 nil）**不能**对 stream:true 回 200 —— 修复前它照样回 200，
// 而客户端永远等不到任何帧、也永远不会落库。这里与同步路径同口径回 500。
func TestChatStreamWithoutRunnerIsNotSilentlyAccepted(t *testing.T) {
	t.Parallel()
	env := newEnv(t, nil, false)
	tok := env.token(t, auth.RoleOwner)
	resp, body := postChat(t, env.ts.URL, tok, map[string]any{"sessionId": "web", "text": "hi", "stream": true})
	if resp.StatusCode != http.StatusInternalServerError {
		t.Fatalf("status = %d, want 500（body=%s）", resp.StatusCode, body)
	}
	if code := errorCodeOf(t, body); code != "chat_unavailable" {
		t.Fatalf("code = %q, want chat_unavailable", code)
	}
}

// TestChatStreamAnswersBusyAndFull 锁定 G2 的顺带项：`stream:true` 也能给出
// 「排队 / 满」这两个明确的答案，而不是一律 200。（修复前 handler 提前 return，
// 202 queued 与 409 chat_busy 在这条路径上永远不可能出现。）
func TestChatStreamAnswersBusyAndFull(t *testing.T) {
	t.Parallel()
	release := make(chan struct{})
	run := &blockingRunner{release: release}
	hist := newReplayHistoryStore()
	srv, ts, jwt := buildEnv(t, cfgDefault(), Deps{Loop: run, History: hist, Bus: &saturatedBus{}})
	tok := signAny(t, jwt, "acct", auth.RoleOwner)

	// 第一条：会话空闲 → 200 stream:true，回合在后台跑着。
	resp, body := postChat(t, ts.URL, tok, map[string]any{"sessionId": "web", "text": "A", "stream": true})
	if resp.StatusCode != http.StatusOK {
		t.Fatalf("第一条 status = %d, want 200（body=%s）", resp.StatusCode, body)
	}
	waitUntil(t, "A 回合在飞", func() bool { return run.started() == 1 })

	// 第二条：会话正忙 → 202 {stream:true, queued:true, queueDepth:1}，不阻塞。
	resp, body = postChat(t, ts.URL, tok, map[string]any{"sessionId": "web", "text": "B", "stream": true})
	if resp.StatusCode != http.StatusAccepted {
		t.Fatalf("第二条 status = %d, want 202（body=%s）", resp.StatusCode, body)
	}
	got := dataOf(t, body)
	if got["queued"] != true || got["stream"] != true || got["queueDepth"] != float64(1) {
		t.Fatalf("第二条 data = %v, want {stream:true,queued:true,queueDepth:1}", got)
	}

	// 填满待处理队列（默认上限 5；已入队 1 条），下一条必须是 409 chat_busy。
	for i := 0; i < 4; i++ {
		resp, body = postChat(t, ts.URL, tok, map[string]any{"sessionId": "web", "text": "fill", "stream": true})
		if resp.StatusCode != http.StatusAccepted {
			t.Fatalf("第 %d 条填充 status = %d, want 202（body=%s）", i+3, resp.StatusCode, body)
		}
	}
	resp, body = postChat(t, ts.URL, tok, map[string]any{"sessionId": "web", "text": "溢出", "stream": true})
	if resp.StatusCode != http.StatusConflict {
		t.Fatalf("溢出 status = %d, want 409（body=%s）", resp.StatusCode, body)
	}
	if code := errorCodeOf(t, body); code != "chat_busy" {
		t.Fatalf("溢出 code = %q, want chat_busy", code)
	}

	close(release)
	waitUntil(t, "会话排空", func() bool {
		return !srv.hub.orchestrator.Busy("web") && srv.hub.orchestrator.QueueDepth("web") == 0
	})
}

// TestChatClientMsgIDDedupesTurn 是 G4 的主断言（REST 同步路径）：同一个
// clientMsgId 发两次 → 只有一条用户消息、只有一个回合；第二次的响应明确给出既有
// messageId，客户端据此对齐本地乐观气泡，而不是靠「文本相等」猜。
func TestChatClientMsgIDDedupesTurn(t *testing.T) {
	t.Parallel()
	run := &countingRunner{}
	hist := &fakeHistory{}
	_, ts, jwt := buildEnv(t, cfgDefault(), Deps{Loop: run, History: hist, Bus: &saturatedBus{}})
	tok := signAny(t, jwt, "acct", auth.RoleOwner)
	body := map[string]any{"sessionId": "web", "text": "弱网重发的那条", "clientMsgId": "local-bubble-1"}

	resp, first := postChat(t, ts.URL, tok, body)
	if resp.StatusCode != http.StatusOK {
		t.Fatalf("第一次 status = %d, want 200（body=%s）", resp.StatusCode, first)
	}

	// 第二次：逐字相同的 clientMsgId → 明确判重，不落库、不起回合。
	resp, second := postChat(t, ts.URL, tok, body)
	if resp.StatusCode != http.StatusOK {
		t.Fatalf("第二次 status = %d, want 200（body=%s）", resp.StatusCode, second)
	}
	dup := dataOf(t, second)
	if dup["duplicate"] != true {
		t.Fatalf("第二次 data = %v, want duplicate:true", dup)
	}
	msgID, _ := dup["messageId"].(string)
	if msgID == "" {
		t.Fatalf("第二次 data = %v, want 既有 messageId（客户端要用它对齐气泡）", dup)
	}

	// 一个回合、一条用户消息：模型只被驱动一次，库里只有 user+assistant 两条。
	if got := run.count(); got != 1 {
		t.Fatalf("回合数 = %d, want 1（重复的 clientMsgId 绝不起第二个回合）", got)
	}
	recs := hist.records()
	if len(recs) != 2 {
		t.Fatalf("落库条数 = %d, want 2（一条用户 + 一条助手）", len(recs))
	}
	if recs[0].Role != "user" || recs[0].ClientMsgID != "local-bubble-1" {
		t.Fatalf("用户消息 = %+v，want clientMsgId=local-bubble-1 被回填", recs[0])
	}
	if recs[0].ID != msgID {
		t.Fatalf("响应 messageId = %q，want 等于落库那条用户消息的 id %q", msgID, recs[0].ID)
	}
}

// TestChatClientMsgIDDuplicateWhilePending 锁定幂等在**排队**分支上的行为：会话
// 忙时，同一个 clientMsgId 的第二条必须立刻判重，而不是又排一次队、又落一条用户
// 消息（重复到达的判定与队列分支无关，两者都在状态锁下的同一道门里）。
func TestChatClientMsgIDDuplicateWhilePending(t *testing.T) {
	t.Parallel()
	release := make(chan struct{})
	run := &blockingRunner{release: release}
	hist := &fakeHistory{}
	srv, ts, jwt := buildEnv(t, cfgDefault(), Deps{Loop: run, History: hist, Bus: &saturatedBus{}})
	tok := signAny(t, jwt, "acct", auth.RoleOwner)

	// 第一条（不带幂等键）占住会话：stream 路径立刻返回，回合在后台跑着。
	resp, body := postChat(t, ts.URL, tok, map[string]any{"sessionId": "web", "text": "占住会话", "stream": true})
	if resp.StatusCode != http.StatusOK {
		t.Fatalf("第一条 status = %d, want 200（body=%s）", resp.StatusCode, body)
	}
	waitUntil(t, "会话变忙", func() bool { return run.started() == 1 })

	dup := map[string]any{"sessionId": "web", "text": "排队的那条", "clientMsgId": "cm-pending"}
	resp, body = postChat(t, ts.URL, tok, dup)
	if resp.StatusCode != http.StatusAccepted {
		t.Fatalf("排队那条 status = %d, want 202（body=%s）", resp.StatusCode, body)
	}
	// 同键再来一次：判重，队列深度不再增长。
	resp, body = postChat(t, ts.URL, tok, dup)
	if resp.StatusCode != http.StatusOK {
		t.Fatalf("重复那条 status = %d, want 200（body=%s）", resp.StatusCode, body)
	}
	if got := dataOf(t, body); got["duplicate"] != true {
		t.Fatalf("重复那条 data = %v, want duplicate:true", got)
	}
	if depth := srv.hub.orchestrator.QueueDepth("web"); depth != 1 {
		t.Fatalf("待处理深度 = %d, want 1（重复的那条不得再排队）", depth)
	}
	withKey := 0
	for _, m := range hist.records() {
		if m.ClientMsgID == "cm-pending" {
			withKey++
		}
	}
	if withKey != 1 {
		t.Fatalf("带该 clientMsgId 的用户消息 = %d 条, want 1", withKey)
	}

	close(release)
	waitUntil(t, "会话排空", func() bool {
		return !srv.hub.orchestrator.Busy("web") && srv.hub.orchestrator.QueueDepth("web") == 0
	})
}

// TestChatWithoutClientMsgIDIsUnchanged 钉住 G4 的兼容口径：省略 clientMsgId 时
// 行为与今天逐字一致 —— 同一段文本发两次就是两条消息、两个回合，响应里也没有
// 任何 duplicate/messageId 字段（老客户端零影响）。
func TestChatWithoutClientMsgIDIsUnchanged(t *testing.T) {
	t.Parallel()
	run := &countingRunner{}
	hist := &fakeHistory{}
	_, ts, jwt := buildEnv(t, cfgDefault(), Deps{Loop: run, History: hist, Bus: &saturatedBus{}})
	tok := signAny(t, jwt, "acct", auth.RoleOwner)
	body := map[string]any{"sessionId": "web", "text": "同一段文本"}

	for i := 0; i < 2; i++ {
		resp, raw := postChat(t, ts.URL, tok, body)
		if resp.StatusCode != http.StatusOK {
			t.Fatalf("第 %d 次 status = %d, want 200（body=%s）", i+1, resp.StatusCode, raw)
		}
		got := dataOf(t, raw)
		if _, ok := got["duplicate"]; ok {
			t.Fatalf("第 %d 次 data = %v，未带幂等键时不该有 duplicate", i+1, got)
		}
		if got["reply"] == nil {
			t.Fatalf("第 %d 次 data = %v, want 正常回复契约", i+1, got)
		}
	}
	if got := run.count(); got != 2 {
		t.Fatalf("回合数 = %d, want 2（无幂等键时逐字保持旧行为）", got)
	}
	recs := hist.records()
	if len(recs) != 4 {
		t.Fatalf("落库条数 = %d, want 4（两次各一条用户 + 一条助手）", len(recs))
	}
	for _, m := range recs {
		if m.ClientMsgID != "" {
			t.Fatalf("落库消息 = %+v，未带幂等键时 clientMsgId 必须为空", m)
		}
	}
}

// TestWSChatClientMsgIDDuplicateIsExplicit 锁定 WS 侧的重复语义：同一
// clientMsgId 的两条 chat 帧**并发到达**（两条帧各自进 goroutine）时，只有一个
// 赢家 —— 一个回合、一条用户消息；输掉的那条收到明确的 `chat_duplicate` 帧，
// 其 messageId 指向真正落库的那条用户消息。
func TestWSChatClientMsgIDDuplicateIsExplicit(t *testing.T) {
	t.Parallel()
	run := &recordingRunner{}
	hist := newReplayHistoryStore()
	srv, ts, jwt := buildEnv(t, cfgDefault(), Deps{Loop: run, History: hist, Bus: &saturatedBus{}})
	tok := signAny(t, jwt, "acct", auth.RoleOwner)

	conn := dialWS(t, ts.URL, tok)
	defer conn.Close()
	waitConnCount(t, srv, 1)

	payload := json.RawMessage(`{"sessionId":"web","text":"同一条","clientMsgId":"ws-bubble-1"}`)
	for i := 0; i < 2; i++ {
		if err := conn.WriteJSON(wsInbound{Type: "chat", Data: payload}); err != nil {
			t.Fatalf("第 %d 条 chat 帧写入失败：%v", i+1, err)
		}
	}

	waitUntil(t, "只驱动一个回合", func() bool { return run.count() == 1 })
	waitUntil(t, "用户+助手消息落库", func() bool { return len(hist.records()) == 2 })

	var userID string
	for _, m := range hist.records() {
		if m.Role == "user" {
			if m.ClientMsgID != "ws-bubble-1" {
				t.Fatalf("用户消息 = %+v，want clientMsgId 被回填", m)
			}
			userID = m.ID
		}
	}
	if userID == "" {
		t.Fatal("没有找到落库的用户消息")
	}

	frames := readChatFramesUntil(t, conn, EventChatDuplicate)
	var dupFrame *wsEvent
	for i := range frames {
		if frames[i].Type == EventChatDuplicate {
			dupFrame = &frames[i]
		}
	}
	if dupFrame == nil {
		t.Fatalf("帧 = %v，want 一条 %s", chatFrameTypes(frames), EventChatDuplicate)
	}
	data, ok := dupFrame.Data.(map[string]any)
	if !ok {
		t.Fatalf("chat_duplicate 载荷 = %#v，want 对象", dupFrame.Data)
	}
	if data["clientMsgId"] != "ws-bubble-1" {
		t.Fatalf("chat_duplicate 载荷 = %v，want clientMsgId=ws-bubble-1", data)
	}
	if data["messageId"] != userID {
		t.Fatalf("chat_duplicate messageId = %v，want 落库那条的 id %q", data["messageId"], userID)
	}
	// 重复的那条绝不产生第二个回合。
	if got := run.count(); got != 1 {
		t.Fatalf("回合数 = %d, want 1", got)
	}
}