package gateway

import (
	"context"
	"encoding/json"
	"errors"
	"net"
	"slices"
	"sync"
	"testing"
	"time"

	"github.com/gorilla/websocket"

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

// 本文件锁定协同项 A2 的 `interrupt {sessionId, scope}` 口径（上游清单：
// `Cyrene-For-Android/devdocs/02-前后端协同事项.md`）。客户端有「追加队列」与
// 「打断」两种模式，而后端过去只有一种口径（取消在途回合 + 清空待处理队列），
// 于是用户选「打断」时连发的多条消息会被静默丢弃，只回一个 dropped 数字。
//
// 落地口径（省略 = 维持现状）：
//   - scope 省略 / "all" ⇒ 现状语义：取消在途回合 + 清空待处理队列；
//   - scope "current" ⇒ 只取消在途回合，**保留**待处理队列（它们按编排器
//     的续跑规则继续执行），dropped 恒为 0；
//   - 其余取值 ⇒ 只回发送者 {type:"error", data:{code:"bad_request"}}，
//     不广播、不打断任何会话（显式报错，绝不静默当默认执行）。
//
// 连接、令牌、编排器与 WS 路由都是真的（httptest + 真 JWT + 真 orchestrator），
// 只有模型适配器是脚本化的，因此「消息到底有没有被驱动」是被观测到的事实，
// 而不是从下行帧反推的推测。
const (
	// a2ReadWait 界定一次预期会产出帧的读取。
	a2ReadWait = 3 * time.Second
	// a2Settle 让一个「本不该发生」的副作用有足够时间暴露自己，再断言它不存在。
	a2Settle = 200 * time.Millisecond
	// a2MainSession 是本组用例使用的会话 id。
	a2MainSession = "web"
)

// a2Runner 是 WS 用例使用的假 Chatter：它记录每次 Run 的末条用户文本后阻塞，
// 直到 release 被关闭或自己的 ctx 被取消（打断路径）。会话因此稳定处于 busy，
// 测试能确定地断言忙时连发的消息最终有没有被驱动。
type a2Runner struct {
	mu      sync.Mutex
	calls   []string
	release chan struct{}
}

func newA2Runner() *a2Runner { return &a2Runner{release: make(chan struct{})} }

func (r *a2Runner) Run(ctx context.Context, _ string, msgs []model.Message) (*loop.Result, error) {
	text := ""
	if n := len(msgs); n > 0 {
		text = messageText(msgs[n-1])
	}
	r.mu.Lock()
	r.calls = append(r.calls, text)
	r.mu.Unlock()
	select {
	case <-r.release:
	case <-ctx.Done():
		return nil, ctx.Err()
	}
	return &loop.Result{Text: "reply-" + text, Steps: 1, StopReason: "stop"}, nil
}

// count 返回已被驱动的回合数。
func (r *a2Runner) count() int {
	r.mu.Lock()
	defer r.mu.Unlock()
	return len(r.calls)
}

// texts 返回已驱动回合的末条用户文本快照。
func (r *a2Runner) texts() []string {
	r.mu.Lock()
	defer r.mu.Unlock()
	return append([]string(nil), r.calls...)
}

// a2Runner 与 a2Event/a2Sink 等夹具、以及单元级用例见
// interrupt_scope_unit_test.go（拆开以守住单文件 ≤500 行）。

// interruptFrameWire 是解码后用于断言的 session_interrupted 帧（§3.2 形状：
// {sessionId, mode, dropped}）。
type interruptFrameWire struct {
	Type      string `json:"type"`
	SessionID string `json:"sessionId"`
	Data      struct {
		SessionID string `json:"sessionId"`
		Mode      string `json:"mode"`
		Dropped   int    `json:"dropped"`
	} `json:"data"`
}

// a2Env 为一个 A2 用例接线网关：脚本化的 runner + 真编排器 + 真 JWT + 真 WS。
type a2Env struct {
	srv *Server
	url string
	jwt *auth.JWTService
	run *a2Runner
	o   *Orchestrator
}

func newA2Env(t *testing.T, run *a2Runner) *a2Env {
	t.Helper()
	srv, ts, jwt := buildEnv(t, cfgDefault(), Deps{Loop: run, History: &fakeHistory{}})
	return &a2Env{srv: srv, url: ts.URL, jwt: jwt, run: run, o: srv.hub.orchestrator}
}

// owner 拨一条 owner 连接的 WS。
func (e *a2Env) owner(t *testing.T) *websocket.Conn {
	t.Helper()
	return dialWS(t, e.url, signAny(t, e.jwt, "acct", auth.RoleOwner))
}

// chatAndQueue 发出一条在飞消息与一条待处理消息，并等到两条都真的落地
// （在飞 + 队列深度 1），使后续断言面对一个确定的忙态而不是靠 sleep 赌时序。
func (e *a2Env) chatAndQueue(t *testing.T, conn *websocket.Conn, inFlight, queued string) {
	t.Helper()
	if err := conn.WriteJSON(wsInbound{Type: "chat", Data: a2ChatData(inFlight)}); err != nil {
		t.Fatalf("write chat %s: %v", inFlight, err)
	}
	waitUntil(t, inFlight+" 回合在飞", func() bool { return e.run.count() == 1 })
	if err := conn.WriteJSON(wsInbound{Type: "chat", Data: a2ChatData(queued)}); err != nil {
		t.Fatalf("write chat %s: %v", queued, err)
	}
	waitUntil(t, queued+" 进入 pending", func() bool { return e.o.QueueDepth(a2MainSession) == 1 })
}

// a2ChatData 构造 chat 上行帧的 data。
func a2ChatData(text string) json.RawMessage {
	b, _ := json.Marshal(map[string]string{"sessionId": a2MainSession, "text": text})
	return b
}

// readFrameOfType 读帧直到出现 want 类型（握手回放、持久化增量等一并跳过），
// 返回那条帧的原始字节；超期未出现即失败。
func readFrameOfType(t *testing.T, conn *websocket.Conn, want, what string) []byte {
	t.Helper()
	deadline := time.Now().Add(a2ReadWait)
	for time.Now().Before(deadline) {
		_ = conn.SetReadDeadline(deadline)
		_, msg, err := conn.ReadMessage()
		if err != nil {
			t.Fatalf("%s: 等待 %q 帧时读取失败: %v", what, want, err)
		}
		var shell struct {
			Type string `json:"type"`
		}
		if err := json.Unmarshal(msg, &shell); err != nil {
			t.Fatalf("%s: 解帧 %s: %v", what, msg, err)
		}
		if shell.Type == want {
			return msg
		}
	}
	t.Fatalf("%s: 截止时间前没有收到 %q 帧", what, want)
	return nil
}

// a2DrainTypes 读取帧直到 window 内再无帧到达，返回窗口内收到的帧类型。
// 它给「本不该出现的帧」一个真实的时间窗口，并按类型断言（例如「不再有
// session_interrupted」），比单个 assertNoFrame 更能表达意图。它是该连接的
// 最后一次读取 —— 读截止时间到期后 gorilla 会留下一个已损坏的连接。
func a2DrainTypes(t *testing.T, conn *websocket.Conn, window time.Duration) []string {
	t.Helper()
	var types []string
	// 每次读取单独续期，但总窗口有上限，使迟到的帧仍能被看见而测试不会挂住。
	stop := time.Now().Add(4 * window)
	for {
		readBy := time.Now().Add(window)
		if readBy.After(stop) {
			readBy = stop
		}
		_ = conn.SetReadDeadline(readBy)
		_, msg, err := conn.ReadMessage()
		if err != nil {
			var netErr net.Error
			if errors.As(err, &netErr) && netErr.Timeout() {
				return types
			}
			t.Fatalf("读取帧失败: %v", err)
		}
		var shell struct {
			Type string `json:"type"`
		}
		if err := json.Unmarshal(msg, &shell); err != nil {
			t.Fatalf("解帧 %s: %v", msg, err)
		}
		types = append(types, shell.Type)
	}
}

// TestInterruptScopeOmittedAndAllDropPending 锁定 A2 的「维持现状」侧：scope
// 省略（老客户端只发 {sessionId}）与显式 "all" 必须逐字同解 —— 取消在途回合 +
// 清空待处理队列，session_interrupted 的 dropped 等于真的被丢弃的条数，且被
// 丢弃的那条绝不被驱动。
func TestInterruptScopeOmittedAndAllDropPending(t *testing.T) {
	t.Parallel()
	cases := []struct {
		name string
		data string
	}{
		{"scope 省略", `{"sessionId":"web"}`},
		{"scope=all", `{"sessionId":"web","scope":"all"}`},
	}
	for _, tc := range cases {
		tc := tc
		t.Run(tc.name, func(t *testing.T) {
			t.Parallel()
			run := newA2Runner()
			env := newA2Env(t, run)
			conn := env.owner(t)
			defer conn.Close()
			waitConnCount(t, env.srv, 1)
			env.chatAndQueue(t, conn, "A", "B")

			if err := conn.WriteJSON(wsInbound{Type: "interrupt", Data: json.RawMessage(tc.data)}); err != nil {
				t.Fatalf("write interrupt: %v", err)
			}
			raw := readFrameOfType(t, conn, "session_interrupted", tc.name)
			var frame interruptFrameWire
			if err := json.Unmarshal(raw, &frame); err != nil {
				t.Fatalf("%s: 解 session_interrupted 帧 %s: %v", tc.name, raw, err)
			}
			if frame.SessionID != a2MainSession || frame.Data.SessionID != a2MainSession {
				t.Fatalf("%s: 帧的会话作用域 = %q/%q, want %q", tc.name, frame.SessionID, frame.Data.SessionID, a2MainSession)
			}
			if frame.Data.Dropped != 1 {
				t.Fatalf("%s: dropped = %d, want 1（队列里那一条真的被丢弃）frame=%s", tc.name, frame.Data.Dropped, raw)
			}
			if frame.Data.Mode != string(ModeIdle) {
				t.Fatalf("%s: mode = %q, want %q（队列清空后会话真的空闲）", tc.name, frame.Data.Mode, ModeIdle)
			}
			waitUntil(t, "队列清空且会话空闲", func() bool {
				return !env.o.Busy(a2MainSession) && env.o.QueueDepth(a2MainSession) == 0
			})

			// 打断后 A 的收尾会发出一条 stream_end（被取消的回合如实收口），
			// 因此这里按类型断言窗口内容，而不是断言「再无任何帧」：既不允许
			// 第二条 session_interrupted，也不允许被丢弃的 B 被驱动。
			close(run.release)
			types := a2DrainTypes(t, conn, a2Settle)
			if slices.Contains(types, "session_interrupted") {
				t.Fatalf("%s: 一次 interrupt 只应上报一次 session_interrupted，实际帧序列 %v", tc.name, types)
			}
			if got := run.texts(); len(got) != 1 || got[0] != "A" {
				t.Fatalf("%s: driven = %v, want [A]（被丢弃的 B 绝不能执行）", tc.name, got)
			}
		})
	}
}

// TestInterruptScopeCurrentOverWSResumesPending 是 A2 在真实 WS 路由上的端到端
// 落地：scope:"current" 的下行帧必须如实报告 dropped=0 且会话未空闲，而下一条
// 已被排队的消息必须真的被驱动（这就是「不再静默丢弃」的证据）。
func TestInterruptScopeCurrentOverWSResumesPending(t *testing.T) {
	t.Parallel()
	run := newA2Runner()
	env := newA2Env(t, run)
	conn := env.owner(t)
	defer conn.Close()
	waitConnCount(t, env.srv, 1)
	env.chatAndQueue(t, conn, "A", "B")

	if err := conn.WriteJSON(wsInbound{Type: "interrupt", Data: json.RawMessage(`{"sessionId":"web","scope":"current"}`)}); err != nil {
		t.Fatalf("write interrupt: %v", err)
	}
	raw := readFrameOfType(t, conn, "session_interrupted", "scope=current")
	var frame interruptFrameWire
	if err := json.Unmarshal(raw, &frame); err != nil {
		t.Fatalf("解 session_interrupted 帧 %s: %v", raw, err)
	}
	if frame.Data.Dropped != 0 {
		t.Fatalf("dropped = %d, want 0（current 口径不丢弃任何待处理消息）frame=%s", frame.Data.Dropped, raw)
	}
	if frame.Data.Mode == string(ModeIdle) {
		t.Fatalf("mode = %q, want 非 idle（队列仍在续跑，会话没有被报成空闲）frame=%s", frame.Data.Mode, raw)
	}

	waitUntil(t, "被保留的 B 续跑", func() bool { return run.count() == 2 })
	if got := run.texts(); len(got) != 2 || got[0] != "A" || got[1] != "B" {
		t.Fatalf("driven = %v, want [A B]（pending 必须续跑）", got)
	}
	close(run.release)
	waitUntil(t, "会话安定", func() bool {
		return !env.o.Busy(a2MainSession) && env.o.QueueDepth(a2MainSession) == 0
	})
}

// TestInterruptScopeInvalidIsDirectedErrorOnly 锁定 A2 的报错侧：非法 scope
// 只回发送者一条 {code:"bad_request"}，不广播（第二条连接收不到）、不打断任何
// 会话（在飞回合继续跑、待处理队列原样保留）。
func TestInterruptScopeInvalidIsDirectedErrorOnly(t *testing.T) {
	t.Parallel()
	run := newA2Runner()
	env := newA2Env(t, run)
	sender := env.owner(t)
	defer sender.Close()
	observer := env.owner(t)
	defer observer.Close()
	waitConnCount(t, env.srv, 2)
	env.chatAndQueue(t, sender, "A", "B")
	if err := sender.WriteJSON(wsInbound{Type: "interrupt", Data: json.RawMessage(`{"sessionId":"web","scope":"pending"}`)}); err != nil {
		t.Fatalf("write interrupt: %v", err)
	}
	raw := readFrameOfType(t, sender, "error", "非法 scope")
	var ef errorFrameWire
	if err := json.Unmarshal(raw, &ef); err != nil {
		t.Fatalf("解错误帧 %s: %v", raw, err)
	}
	if ef.Data.Code != wsBadRequestCode {
		t.Fatalf("error code = %q, want %q（非法 scope 必须显式报错，不得静默当默认）frame=%s", ef.Data.Code, wsBadRequestCode, raw)
	}

	// 只到发送者：观察者（它先收到了 chat 带来的状态帧与握手回放，这些被跳过）
	// 在窗口里既收不到 error、也收不到 session_interrupted —— 拒绝不是广播。
	seen := a2DrainTypes(t, observer, a2Settle)
	if slices.Contains(seen, "error") || slices.Contains(seen, "session_interrupted") {
		t.Fatalf("非法 scope 被广播了：观察者收到 %v（必须只定向回发送者）", seen)
	}
	// 发送者自己也只多了那一条错误帧：没有 session_interrupted。
	sendSeen := a2DrainTypes(t, sender, a2Settle)
	if slices.Contains(sendSeen, "session_interrupted") {
		t.Fatalf("非法 scope 仍然打断了会话：发送者收到 %v", sendSeen)
	}

	// 会话完全没有被触动：回合 A 仍在飞（a2Runner 没被取消），队列仍是 1。
	if got := run.texts(); len(got) != 1 || got[0] != "A" {
		t.Fatalf("driven = %v, want [A]（非法 scope 不得打断或续跑任何东西）", got)
	}
	if env.o.QueueDepth(a2MainSession) != 1 || !env.o.Busy(a2MainSession) {
		t.Fatalf("会话被非法 scope 触动了：busy=%v depth=%d", env.o.Busy(a2MainSession), env.o.QueueDepth(a2MainSession))
	}
	close(run.release)
}