Scoreboard/server/internal/realtime/hub_test.go

package realtime

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log/slog"
	"net/http"
	"net/http/httptest"
	"path/filepath"
	"reflect"
	"strings"
	"sync/atomic"
	"testing"
	"time"

	"github.com/coder/websocket"
	_ "modernc.org/sqlite"

	"git.yeij.top/AskaEth/Scoreboard/server/internal/auth"
	"git.yeij.top/AskaEth/Scoreboard/server/internal/dto"
	"git.yeij.top/AskaEth/Scoreboard/server/internal/middleware"
	"git.yeij.top/AskaEth/Scoreboard/server/internal/store"
)

// ---------- 测试脚手架 ----------

type testEnv struct {
	hub *Hub
	db  *sql.DB
	srv *httptest.Server
}

// newTestEnv 建一个未登录（guest）的服务端：默认身份与大屏端一致。
func newTestEnv(t *testing.T) *testEnv {
	t.Helper()
	return newTestEnvAs(t, &auth.User{Role: auth.RoleGuest})
}

// newTestEnvAs 注入会话身份，等价于真实链路里 Authenticate 中间件写进 context 的 User。
func newTestEnvAs(t *testing.T, u *auth.User) *testEnv {
	t.Helper()
	return newTestEnvAsSession(t, u, "")
}

// newTestEnvAsSession 在身份之外再注入会话 id，等价于 Authenticate 中间件写进 context 的
// *auth.Session：会话复查（撤销即断开）要按会话 id 判定，测试必须能给出这一条。
func newTestEnvAsSession(t *testing.T, u *auth.User, sessionID string) *testEnv {
	t.Helper()

	db := newTestDB(t)
	h := NewHub(db, slog.New(slog.NewTextHandler(io.Discard, &slog.HandlerOptions{Level: slog.LevelError})))
	handler := h.Handler()
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		ctx := middleware.WithUser(r.Context(), u)
		if sessionID != "" {
			ctx = middleware.WithSession(ctx, &auth.Session{ID: sessionID, UserID: u.ID})
		}
		handler.ServeHTTP(w, r.WithContext(ctx))
	}))

	t.Cleanup(func() {
		srv.Close()
		_ = h.Close()
		_ = db.Close()
	})
	return &testEnv{hub: h, db: db, srv: srv}
}

func newTestDB(t *testing.T) *sql.DB {
	t.Helper()

	path := filepath.Join(t.TempDir(), "event.db")
	db, err := sql.Open("sqlite", "file:"+path+
		"?_pragma=journal_mode(WAL)&_pragma=foreign_keys(1)&_pragma=busy_timeout(5000)")
	if err != nil {
		t.Fatalf("打开测试库: %v", err)
	}
	t.Cleanup(func() { _ = db.Close() })

	if err := store.MigrateEvent(db); err != nil {
		t.Fatalf("迁移测试库: %v", err)
	}
	return db
}

func (e *testEnv) dial(t *testing.T, hello *helloPayload) *wsClient {
	t.Helper()

	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	url := "ws" + strings.TrimPrefix(e.srv.URL, "http") + "/ws"
	conn, _, err := websocket.Dial(ctx, url, nil)
	if err != nil {
		cancel()
		t.Fatalf("连接 /ws: %v", err)
	}

	c := &wsClient{t: t, ctx: ctx, cancel: cancel, conn: conn, msgs: make(chan wsMessage, 256)}
	go c.readLoop()
	t.Cleanup(c.close)
	if hello != nil {
		c.send(EventHello, *hello)
	}
	return c
}

type wsMessage struct {
	event     string
	data      json.RawMessage
	err       error
	closeCode websocket.StatusCode
}

type wsClient struct {
	t      *testing.T
	ctx    context.Context
	cancel context.CancelFunc
	conn   *websocket.Conn
	msgs   chan wsMessage
}

// readLoop 单独读取：测试里等待「没有消息」时不能靠取消 Read 的 context，
// 那会把整条连接关掉，后续用例就没法继续读了。
func (c *wsClient) readLoop() {
	defer close(c.msgs)
	for {
		_, data, err := c.conn.Read(c.ctx)
		if err != nil {
			c.msgs <- wsMessage{err: err, closeCode: websocket.CloseStatus(err)}
			return
		}
		var env envelope
		if err := json.Unmarshal(data, &env); err != nil {
			c.msgs <- wsMessage{err: err}
			return
		}
		c.msgs <- wsMessage{event: env.Event, data: env.Data}
	}
}

func (c *wsClient) close() {
	_ = c.conn.CloseNow()
	c.cancel()
}

func (c *wsClient) send(event string, data any) {
	c.t.Helper()
	msg, err := encode(event, data)
	if err != nil {
		c.t.Fatalf("编码 %s: %v", event, err)
	}
	ctx, cancel := context.WithTimeout(c.ctx, 5*time.Second)
	defer cancel()
	if err := c.conn.Write(ctx, websocket.MessageText, msg); err != nil {
		c.t.Fatalf("发送 %s: %v", event, err)
	}
}

// recvWithin 在 d 内取一条消息；超时返回 ok=false（连接保持可用）。
func (c *wsClient) recvWithin(d time.Duration) (string, json.RawMessage, bool) {
	c.t.Helper()
	timer := time.NewTimer(d)
	defer timer.Stop()

	select {
	case msg, ok := <-c.msgs:
		if !ok || msg.err != nil {
			c.t.Fatalf("连接已断开: %v", msg.err)
		}
		return msg.event, msg.data, true
	case <-timer.C:
		return "", nil, false
	}
}

// recvEvent 等待指定事件，跳过心跳等无关消息。
func (c *wsClient) recvEvent(want string) json.RawMessage {
	c.t.Helper()
	deadline := time.Now().Add(5 * time.Second)
	for time.Now().Before(deadline) {
		ev, data, ok := c.recvWithin(time.Until(deadline))
		if !ok {
			break
		}
		if ev == want {
			return data
		}
	}
	c.t.Fatalf("等待事件 %s 超时", want)
	return nil
}

// awaitSnapshot 在 d 内等待一份整库快照（心跳等无关消息跳过）；等到返回 true。
func (c *wsClient) awaitSnapshot(d time.Duration) bool {
	c.t.Helper()
	deadline := time.Now().Add(d)
	for time.Now().Before(deadline) {
		ev, _, ok := c.recvWithin(time.Until(deadline))
		if !ok {
			return false
		}
		if ev == EventSnapshot {
			return true
		}
	}
	return false
}

// noSnapshotWithin 断言 d 内没有整库快照；心跳等无关消息不算（连接断开则直接失败）。
func (c *wsClient) noSnapshotWithin(d time.Duration) {
	c.t.Helper()
	deadline := time.Now().Add(d)
	for time.Now().Before(deadline) {
		ev, _, ok := c.recvWithin(time.Until(deadline))
		if !ok {
			return
		}
		if ev == EventSnapshot {
			c.t.Fatalf("这段时间内不该有整库快照")
		}
	}
}

// expectClosed 断言连接在 d 内被服务端关闭，且关闭前没有下发任何事件。
func (c *wsClient) expectClosed(d time.Duration) {
	c.t.Helper()
	timer := time.NewTimer(d)
	defer timer.Stop()
	for {
		select {
		case msg, ok := <-c.msgs:
			if !ok || msg.err != nil {
				return
			}
			c.t.Fatalf("连接关闭前仍收到事件 %s", msg.event)
		case <-timer.C:
			c.t.Fatalf("等待服务端关闭连接超时")
		}
	}
}

// expectClosedWithCode 断言连接被服务端以指定关闭码关闭，且关闭前没有下发任何事件。
func (c *wsClient) expectClosedWithCode(d time.Duration, want websocket.StatusCode) {
	c.t.Helper()
	timer := time.NewTimer(d)
	defer timer.Stop()
	for {
		select {
		case msg, ok := <-c.msgs:
			if !ok {
				c.t.Fatalf("连接已关闭但没有收到关闭帧")
			}
			if msg.err != nil {
				if msg.closeCode != want {
					c.t.Fatalf("关闭码应为 %d，得到 %d (%v)", want, msg.closeCode, msg.err)
				}
				return
			}
			c.t.Fatalf("连接关闭前仍收到事件 %s", msg.event)
		case <-timer.C:
			c.t.Fatalf("等待服务端关闭连接超时")
		}
	}
}

// expectClosedWithCodeDraining 断言连接最终被服务端以指定关闭码关闭。
// 关闭前已下发的正常事件会被丢弃：慢客户端被踢属于「先推后关」，不能要求关闭前零事件。
func (c *wsClient) expectClosedWithCodeDraining(d time.Duration, want websocket.StatusCode) {
	c.t.Helper()
	timer := time.NewTimer(d)
	defer timer.Stop()
	for {
		select {
		case msg, ok := <-c.msgs:
			if !ok {
				c.t.Fatalf("连接已关闭但没有收到关闭帧")
			}
			if msg.err != nil {
				if msg.closeCode != want {
					c.t.Fatalf("关闭码应为 %d，得到 %d (%v)", want, msg.closeCode, msg.err)
				}
				return
			}
		case <-timer.C:
			c.t.Fatalf("等待服务端关闭连接超时")
		}
	}
}

func decode[T any](t *testing.T, raw json.RawMessage) T {
	t.Helper()
	var v T
	if err := json.Unmarshal(raw, &v); err != nil {
		t.Fatalf("解析载荷 %s: %v", raw, err)
	}
	return v
}

func insertChange(t *testing.T, db *sql.DB, entity string, entityID int64, action, payload string) int64 {
	t.Helper()
	res, err := db.Exec(
		`INSERT INTO change_log(entity_type, entity_id, action, payload_json) VALUES (?, ?, ?, ?)`,
		entity, entityID, action, payload)
	if err != nil {
		t.Fatalf("写入 change_log: %v", err)
	}
	id, err := res.LastInsertId()
	if err != nil {
		t.Fatalf("读取 change_log id: %v", err)
	}
	return id
}

func loadChanges(t *testing.T, h *Hub, since int64, limit int) []Change {
	t.Helper()
	changes, err := h.LoadChanges(context.Background(), since, limit)
	if err != nil {
		t.Fatalf("读取增量变更: %v", err)
	}
	return changes
}

func waitFor(t *testing.T, what string, cond func() bool) {
	t.Helper()
	deadline := time.Now().Add(3 * time.Second)
	for time.Now().Before(deadline) {
		if cond() {
			return
		}
		time.Sleep(5 * time.Millisecond)
	}
	t.Fatalf("等待条件超时: %s", what)
}

func ptrInt64(v int64) *int64 { return &v }

// ---------- 客户端状态归约（模拟前端按 version 应用变更） ----------

type clientState map[string]json.RawMessage

func stateKey(ch Change) string { return fmt.Sprintf("%s/%d", ch.Entity, ch.ID) }

func applyChanges(s clientState, changes []Change) clientState {
	for _, ch := range changes {
		switch ch.Action {
		case ActionVoid:
			delete(s, stateKey(ch))
		case ActionReplace:
			s[ch.Entity] = ch.Data
		default:
			s[stateKey(ch)] = ch.Data
		}
	}
	return s
}

// ---------- 用例 ----------

func TestHelloSnapshotThenChange(t *testing.T) {
	env := newTestEnv(t)
	c := env.dial(t, &helloPayload{Role: RoleDisplay, InstanceID: "d1"})

	snap := decode[Snapshot](t, c.recvEvent(EventSnapshot))
	if snap.Version != 0 {
		t.Fatalf("空库快照版本应为 0，得到 %d", snap.Version)
	}

	v := insertChange(t, env.db, "result", 42, ActionUpsert, `{"rank":1}`)
	env.hub.Broadcast(loadChanges(t, env.hub, 0, 0))

	got := decode[changePayload](t, c.recvEvent(EventChange))
	if got.Version != v {
		t.Fatalf("change 版本应为 %d，得到 %d", v, got.Version)
	}
	if len(got.Changes) != 1 {
		t.Fatalf("change 条数应为 1，得到 %d", len(got.Changes))
	}
	ch := got.Changes[0]
	if ch.Entity != "result" || ch.ID != 42 || ch.Action != ActionUpsert {
		t.Fatalf("change 内容不符: %+v", ch)
	}
	if string(ch.Data) != `{"rank":1}` {
		t.Fatalf("change data 不符: %s", ch.Data)
	}
	if env.hub.Version() != v {
		t.Fatalf("Hub 版本应推进到 %d，得到 %d", v, env.hub.Version())
	}
}

// 断线期间的变更通过 catchup 补齐，且终态与全量重放的终态一致。
func TestReconnectCatchupTerminalState(t *testing.T) {
	env := newTestEnv(t)
	v1 := insertChange(t, env.db, "result", 1, ActionUpsert, `{"rank":1}`)
	v2 := insertChange(t, env.db, "standings", 0, ActionReplace, `[{"teamId":1,"points":10}]`)

	c1 := env.dial(t, &helloPayload{Role: RoleDisplay, InstanceID: "d1"})
	snap := decode[Snapshot](t, c1.recvEvent(EventSnapshot))
	if snap.Version != v2 || snap.Version != v1+1 {
		t.Fatalf("首次快照版本应为 %d，得到 %d", v2, snap.Version)
	}
	c1.close()

	v3 := insertChange(t, env.db, "result", 2, ActionUpsert, `{"rank":2}`)
	v4 := insertChange(t, env.db, "result", 1, ActionVoid, `{}`)
	v5 := insertChange(t, env.db, "standings", 0, ActionReplace, `[{"teamId":1,"points":12}]`)
	if v4 != v3+1 || v5 != v4+1 {
		t.Fatalf("change_log 版本应连续递增: %d %d %d", v3, v4, v5)
	}

	c2 := env.dial(t, &helloPayload{Role: RoleDisplay, InstanceID: "d1", LastVersion: ptrInt64(snap.Version)})
	cp := decode[catchupPayload](t, c2.recvEvent(EventCatchup))

	if cp.FromVersion != snap.Version {
		t.Fatalf("fromVersion 应为 %d，得到 %d", snap.Version, cp.FromVersion)
	}
	if cp.Version != v5 {
		t.Fatalf("catchup 版本应为 %d，得到 %d", v5, cp.Version)
	}
	// 版本号不上线路（协议只在消息层带 version），因此按内容核对补发的三条变更。
	wantChanges := []struct {
		entity string
		id     int64
		action string
	}{
		{"result", 2, ActionUpsert},
		{"result", 1, ActionVoid},
		{"standings", 0, ActionReplace},
	}
	if len(cp.Changes) != len(wantChanges) {
		t.Fatalf("catchup 条数应为 %d，得到 %d", len(wantChanges), len(cp.Changes))
	}
	for i, want := range wantChanges {
		got := cp.Changes[i]
		if got.Entity != want.entity || got.ID != want.id || got.Action != want.action {
			t.Fatalf("第 %d 条变更应为 %+v，得到 %+v", i, want, got)
		}
	}

	// 断线前已应用的变更 + catchup 的变更 == 从头全量重放，终态必须一致。
	base := applyChanges(clientState{}, loadChanges(t, env.hub, 0, int(snap.Version)))
	got := applyChanges(base, cp.Changes)
	want := applyChanges(clientState{}, loadChanges(t, env.hub, 0, 0))
	if !reflect.DeepEqual(got, want) {
		t.Fatalf("catchup 终态与全量重放不一致:\n got=%v\nwant=%v", got, want)
	}

	// 新连接的整包快照版本必须与 catchup 的终点一致。
	fresh, err := env.hub.Snapshot(context.Background())
	if err != nil {
		t.Fatalf("构建快照: %v", err)
	}
	if fresh.Version != cp.Version {
		t.Fatalf("快照版本 %d 与 catchup 版本 %d 不一致", fresh.Version, cp.Version)
	}
}

// 增量超过阈值时回落全量快照。
func TestCatchupFallsBackToSnapshotWhenTooOld(t *testing.T) {
	env := newTestEnv(t)
	env.hub.SetMaxCatchup(2)
	first := insertChange(t, env.db, "result", 1, ActionUpsert, `{}`)
	for i := int64(2); i <= 5; i++ {
		insertChange(t, env.db, "result", i, ActionUpsert, `{}`)
	}

	c := env.dial(t, &helloPayload{Role: RoleDisplay, InstanceID: "d1", LastVersion: ptrInt64(first - 1)})
	snap := decode[Snapshot](t, c.recvEvent(EventSnapshot))
	if snap.Version != 5 {
		t.Fatalf("应回落全量快照且版本为 5，得到 %d", snap.Version)
	}
}

func TestDuplicateDeliveryIgnored(t *testing.T) {
	env := newTestEnv(t)
	c := env.dial(t, &helloPayload{Role: RoleDisplay, InstanceID: "d1"})
	c.recvEvent(EventSnapshot)

	v1 := insertChange(t, env.db, "result", 1, ActionUpsert, `{"rank":1}`)
	changes := loadChanges(t, env.hub, 0, 0)
	env.hub.Broadcast(changes)

	got := decode[changePayload](t, c.recvEvent(EventChange))
	if got.Version != v1 {
		t.Fatalf("首投版本应为 %d，得到 %d", v1, got.Version)
	}
	c.send(EventAck, ackPayload{Version: v1})

	// 同一批变更再投一次，客户端不应再收到。
	env.hub.Broadcast(changes)
	if ev, _, ok := c.recvWithin(300 * time.Millisecond); ok {
		t.Fatalf("重复投递未被忽略，收到事件 %s", ev)
	}

	// 新变更仍能正常下发。
	v2 := insertChange(t, env.db, "result", 2, ActionUpsert, `{"rank":2}`)
	env.hub.Broadcast(loadChanges(t, env.hub, v1, 0))
	got = decode[changePayload](t, c.recvEvent(EventChange))
	if got.Version != v2 {
		t.Fatalf("新变更版本应为 %d，得到 %d", v2, got.Version)
	}
}

// 心跳携带版本号与服务器时间；版本落后时客户端 resync 能拿到最新快照。
// 本用例测的是协议本身（请求 → 响应），整库快照窗口另有专门用例，这里关闭限流。
func TestHeartbeatAndResync(t *testing.T) {
	env := newTestEnv(t)
	env.hub.SetHeartbeatInterval(40 * time.Millisecond)
	env.hub.SetSnapshotMinInterval(0)

	c := env.dial(t, &helloPayload{Role: RoleDisplay, InstanceID: "d1"})
	c.recvEvent(EventSnapshot)

	// 只写库不广播：心跳版本会领先于客户端，客户端据此发起 resync。
	insertChange(t, env.db, "result", 1, ActionUpsert, `{}`)
	insertChange(t, env.db, "result", 2, ActionUpsert, `{}`)

	hb := decode[heartbeatPayload](t, c.recvEvent(EventHeartbeat))
	if hb.Version != 2 {
		t.Fatalf("心跳版本应为 2，得到 %d", hb.Version)
	}
	if _, err := time.Parse(time.RFC3339Nano, hb.ServerTime); err != nil {
		t.Fatalf("serverTime 不是 RFC3339: %q (%v)", hb.ServerTime, err)
	}

	c.send(EventResync, struct{}{})
	snap := decode[Snapshot](t, c.recvEvent(EventSnapshot))
	if snap.Version != 2 {
		t.Fatalf("resync 快照版本应为 2，得到 %d", snap.Version)
	}
}

// 大屏实例注册表：instanceId、在线状态、当前页、最后心跳可查。
// 重复 hello 会回一份快照（窗口内属于限流范围，这里关掉限流以保持用例只关心注册表）。
func TestDisplayInstancesRegistry(t *testing.T) {
	env := newTestEnv(t)
	env.hub.SetSnapshotMinInterval(0)
	c := env.dial(t, &helloPayload{Role: RoleDisplay, InstanceID: "screen-1", Page: "standings"})
	c.recvEvent(EventSnapshot)

	insts := env.hub.Instances()
	if len(insts) != 1 {
		t.Fatalf("实例数应为 1，得到 %d", len(insts))
	}
	if insts[0].ID != "screen-1" || !insts[0].Online || insts[0].Page != "standings" {
		t.Fatalf("实例注册内容不符: %+v", insts[0])
	}
	if insts[0].LastHeartbeat.IsZero() {
		t.Fatalf("最后心跳不应为零值: %+v", insts[0])
	}

	// 重复 hello 不应把实例引用计数加两次，否则断开后仍会显示在线。
	c.send(EventHello, helloPayload{Role: RoleDisplay, InstanceID: "screen-1", Page: "standings"})
	c.recvEvent(EventSnapshot)
	if insts := env.hub.Instances(); len(insts) != 1 || !insts[0].Online {
		t.Fatalf("重复 hello 后实例应仍只有一个且在线: %+v", insts)
	}

	gotoCmd, err := NewDisplayCommand(CmdGoto, map[string]string{"page": "medals"})
	if err != nil {
		t.Fatalf("构造 goto: %v", err)
	}
	env.hub.SendCommand(gotoCmd)
	cmd := decode[DisplayCommand](t, c.recvEvent(EventDisplayCommand))
	if cmd.Action != CmdGoto {
		t.Fatalf("指令 action 应为 goto，得到 %s", cmd.Action)
	}
	if insts := env.hub.Instances(); insts[0].Page != "medals" {
		t.Fatalf("goto 后当前页应为 medals，得到 %s", insts[0].Page)
	}

	detail, err := NewDisplayCommand(CmdShowDetail, map[string]any{"eventId": 7})
	if err != nil {
		t.Fatalf("构造 show-detail: %v", err)
	}
	env.hub.SendCommand(detail)
	if got := decode[DisplayCommand](t, c.recvEvent(EventDisplayCommand)); got.Action != CmdShowDetail {
		t.Fatalf("指令 action 应为 show-detail，得到 %s", got.Action)
	}

	c.close()
	waitFor(t, "实例离线", func() bool {
		insts := env.hub.Instances()
		return len(insts) == 1 && !insts[0].Online
	})
}

// 房间隔离：admin:<userId> 只发给对应管理端，且 userId 来自会话而非客户端自报。
func TestAdminRoomTargeting(t *testing.T) {
	env := newTestEnvAs(t, &auth.User{ID: 7, Username: "boss", Role: auth.RoleAdmin, Active: true})
	display := env.dial(t, &helloPayload{Role: RoleDisplay, InstanceID: "d1"})
	display.recvEvent(EventSnapshot)
	// hello 不再携带 userId：定向房间必须用会话里的 ID。
	admin := env.dial(t, &helloPayload{Role: RoleAdmin})
	admin.recvEvent(EventSnapshot)

	insertChange(t, env.db, "result", 1, ActionUpsert, `{"rank":1}`)
	env.hub.BroadcastTo(AdminRoom(7), loadChanges(t, env.hub, 0, 0))

	got := decode[changePayload](t, admin.recvEvent(EventChange))
	if got.Version != 1 {
		t.Fatalf("定向 change 版本应为 1，得到 %d", got.Version)
	}
	if ev, _, ok := display.recvWithin(300 * time.Millisecond); ok {
		t.Fatalf("大屏不应收到 admin:<uid> 房间的消息，收到 %s", ev)
	}
}

// guest 自报 role=admin 必须被拒绝：管理端房间只能由会话授予。
// 断开必须带上策略性关闭码，否则客户端会当成网络异常无限重连。
func TestGuestCannotSubscribeAdminRoom(t *testing.T) {
	env := newTestEnv(t)
	c := env.dial(t, &helloPayload{Role: RoleAdmin})

	c.expectClosedWithCode(3*time.Second, websocket.StatusPolicyViolation)
}

// guest 只能订阅 display 房间：即使 hello 里塞了 userId 也不能进入定向房间。
func TestGuestStaysInDisplayRoom(t *testing.T) {
	env := newTestEnv(t)
	c := env.dial(t, nil)
	c.send(EventHello, map[string]any{"role": RoleDisplay, "userId": 99})
	c.recvEvent(EventSnapshot)

	insertChange(t, env.db, "result", 1, ActionUpsert, `{"rank":1}`)
	changes := loadChanges(t, env.hub, 0, 0)
	env.hub.BroadcastTo(RoomAdmin, changes)
	env.hub.BroadcastTo(AdminRoom(99), changes)
	if ev, _, ok := c.recvWithin(300 * time.Millisecond); ok {
		t.Fatalf("guest 不应收到管理端房间的消息，收到 %s", ev)
	}

	// display 房间的消息仍然正常到达。
	env.hub.BroadcastTo(RoomDisplay, changes)
	if got := decode[changePayload](t, c.recvEvent(EventChange)); got.Version != 1 {
		t.Fatalf("大屏应收到 display 房间消息，版本得到 %d", got.Version)
	}
}

// recorder 会话可以进管理端房间（大屏控制/数据同步都由管理端承担）。
func TestRecorderSessionJoinsAdminRoom(t *testing.T) {
	env := newTestEnvAs(t, &auth.User{ID: 3, Username: "rec", Role: auth.RoleRecorder, Active: true})
	c := env.dial(t, &helloPayload{Role: RoleAdmin})
	c.recvEvent(EventSnapshot)

	insertChange(t, env.db, "result", 1, ActionUpsert, `{"rank":1}`)
	env.hub.BroadcastTo(RoomAdmin, loadChanges(t, env.hub, 0, 0))

	if got := decode[changePayload](t, c.recvEvent(EventChange)); got.Version != 1 {
		t.Fatalf("recorder 应进入管理端房间，版本得到 %d", got.Version)
	}
}

// 慢客户端：ack 落后超过阈值即被识别并断开。
func TestSlowClientDetected(t *testing.T) {
	env := newTestEnv(t)
	env.hub.SetSlowLag(1)

	c := env.dial(t, &helloPayload{Role: RoleDisplay, InstanceID: "slow-1"})
	c.recvEvent(EventSnapshot)

	for i := int64(1); i <= 3; i++ {
		insertChange(t, env.db, "result", i, ActionUpsert, `{}`)
	}
	env.hub.Broadcast(loadChanges(t, env.hub, 0, 0))

	waitFor(t, "慢客户端被标记", func() bool {
		insts := env.hub.Instances()
		return len(insts) == 1 && insts[0].Slow
	})
	waitFor(t, "慢客户端连接被断开", func() bool {
		insts := env.hub.Instances()
		return len(insts) == 1 && !insts[0].Online
	})
	// 必须用自定义关闭码：前端把所有 1008 当登录失效并永久停重连，大屏会就此冻结。
	c.expectClosedWithCodeDraining(3*time.Second, slowCloseCode)
}

// 快照对所有角色下发，但系统配置是 admin-only：guest/recorder 只能拿到大屏渲染
// 所需的 event/display/榜单开关，默认积分模板、审核开关与折算系数必须置空/清零。
func TestSnapshotSettingsRedactedForNonAdmin(t *testing.T) {
	for _, tc := range []struct {
		name string
		user *auth.User
	}{
		{"guest", &auth.User{Role: auth.RoleGuest}},
		{"recorder", &auth.User{ID: 3, Username: "rec", Role: auth.RoleRecorder, Active: true}},
	} {
		t.Run(tc.name, func(t *testing.T) {
			env := newTestEnvAs(t, tc.user)
			mustExec(t, env.db, `UPDATE settings SET value = '0.5' WHERE key = 'standings.individual_to_team_weight'`)
			mustExec(t, env.db, `UPDATE settings SET value = 'true' WHERE key = 'results.review_required'`)

			c := env.dial(t, &helloPayload{Role: RoleDisplay, InstanceID: tc.name + "-1"})
			snap := decode[Snapshot](t, c.recvEvent(EventSnapshot))

			if snap.Settings.DefaultProfileID != nil {
				t.Fatalf("非 admin 不应看到 defaultProfileId: %+v", snap.Settings.DefaultProfileID)
			}
			if snap.Settings.RequireApproval {
				t.Fatalf("非 admin 不应看到 requireApproval")
			}
			if snap.Settings.Standings.IndividualToTeamWeight != 0 {
				t.Fatalf("非 admin 不应看到 individualToTeamWeight: %v", snap.Settings.Standings.IndividualToTeamWeight)
			}
			// 大屏要用的字段必须保留，裁剪不能误伤标题/主题/榜单开关。
			if snap.Settings.Event.Name == "" || snap.Settings.Display.RotateSecond == 0 {
				t.Fatalf("event/display 不应被裁剪: %+v", snap.Settings)
			}
			if snap.Settings.Standings.RankingMode != "points" ||
				!snap.Settings.Standings.ShowIndividualBoard || !snap.Settings.Standings.ShowIndividualMedals {
				t.Fatalf("standings 展示开关不应被裁剪: %+v", snap.Settings.Standings)
			}
		})
	}
}

// admin 仍拿到完整配置，管理端才能在 WS 首包里读到默认积分模板与审核开关。
func TestSnapshotSettingsFullForAdmin(t *testing.T) {
	env := newTestEnvAs(t, &auth.User{ID: 1, Username: "boss", Role: auth.RoleAdmin, Active: true})
	mustExec(t, env.db, `UPDATE settings SET value = '0.5' WHERE key = 'standings.individual_to_team_weight'`)
	mustExec(t, env.db, `UPDATE settings SET value = 'true' WHERE key = 'results.review_required'`)

	c := env.dial(t, &helloPayload{Role: RoleAdmin})
	snap := decode[Snapshot](t, c.recvEvent(EventSnapshot))

	if snap.Settings.DefaultProfileID == nil {
		t.Fatalf("admin 应看到 defaultProfileId")
	}
	if !snap.Settings.RequireApproval {
		t.Fatalf("admin 应看到 requireApproval")
	}
	if snap.Settings.Standings.IndividualToTeamWeight != 0.5 {
		t.Fatalf("admin 应看到 individualToTeamWeight=0.5，得到 %v", snap.Settings.Standings.IndividualToTeamWeight)
	}
}

// resync 在 hello 之前不可用：否则会绕过房间分配并提前清掉 hello 超时。
func TestResyncBeforeHelloIgnored(t *testing.T) {
	env := newTestEnv(t)
	env.hub.SetSnapshotMinInterval(200 * time.Millisecond)
	insertChange(t, env.db, "result", 1, ActionUpsert, `{"rank":1}`)

	c := env.dial(t, nil)
	c.send(EventResync, struct{}{})
	if ev, _, ok := c.recvWithin(300 * time.Millisecond); ok {
		t.Fatalf("hello 之前的 resync 不应下发快照，收到事件 %s", ev)
	}

	c.send(EventHello, helloPayload{Role: RoleDisplay, InstanceID: "d1"})
	if snap := decode[Snapshot](t, c.recvEvent(EventSnapshot)); snap.Version != 1 {
		t.Fatalf("hello 后应收到版本 1 的快照，得到 %d", snap.Version)
	}
	// hello 之后 resync 仍然可用，只是要与刚发出去的整库快照共用同一个窗口
	// （生产默认窗口与客户端 3s 冷却一致；这里把它调短，免得测试睡 3 秒）。
	time.Sleep(250 * time.Millisecond)
	c.send(EventResync, struct{}{})
	if snap := decode[Snapshot](t, c.recvEvent(EventSnapshot)); snap.Version != 1 {
		t.Fatalf("hello 后 resync 应拿到快照，得到 %d", snap.Version)
	}
}

// 离线实例不能无限堆积：超过 TTL 后在下一次读取注册表时被清理。
func TestInstanceRegistryPrunesStaleOffline(t *testing.T) {
	env := newTestEnv(t)
	c := env.dial(t, &helloPayload{Role: RoleDisplay, InstanceID: "stale-1"})
	c.recvEvent(EventSnapshot)

	c.close()
	waitFor(t, "实例离线", func() bool {
		insts := env.hub.Instances()
		return len(insts) == 1 && !insts[0].Online
	})

	env.hub.mu.Lock()
	env.hub.instances["stale-1"].lastHeartbeat = time.Now().Add(-instanceTTL - time.Minute)
	env.hub.mu.Unlock()

	if insts := env.hub.Instances(); len(insts) != 0 {
		t.Fatalf("过期离线实例应被清理，得到 %+v", insts)
	}
}

// 读写 goroutine 的 panic 必须被兜住并断开该连接，而不是带走进程。
func TestGuardLoopRecoversPanic(t *testing.T) {
	env := newTestEnv(t)
	c := newClient(env.hub, nil, "test", wsIdentity{Role: RoleDisplay})

	func() {
		defer c.guardLoop("读")
		panic("boom")
	}()

	select {
	case <-c.done:
	default:
		t.Fatal("panic 后应关闭该连接")
	}
	if c.closeCode != websocket.StatusInternalError {
		t.Fatalf("panic 后关闭码应为 %d，得到 %d", websocket.StatusInternalError, c.closeCode)
	}
}

// ---------- 按实例定向下发 ----------

// instancePage 取注册表里某个实例的当前页；实例不存在返回空串。
func instancePage(insts []Instance, id string) string {
	for _, st := range insts {
		if st.ID == id {
			return st.Page
		}
	}
	return ""
}

// 定向下发只发给被点名的在线实例；注册表里的「当前页」也只更新那一块屏。
func TestSendCommandTargetsSingleInstance(t *testing.T) {
	env := newTestEnv(t)
	a := env.dial(t, &helloPayload{Role: RoleDisplay, InstanceID: "screen-a", Page: "standings", Screen: "1"})
	b := env.dial(t, &helloPayload{Role: RoleDisplay, InstanceID: "screen-b", Page: "standings", Screen: "2"})
	a.recvEvent(EventSnapshot)
	b.recvEvent(EventSnapshot)

	cmd, err := NewDisplayCommand(CmdGoto, map[string]string{"page": "medals"})
	if err != nil {
		t.Fatalf("构造 goto: %v", err)
	}
	matched, err := env.hub.SendCommandToInstances([]string{"screen-a"}, cmd)
	if err != nil {
		t.Fatalf("定向下发: %v", err)
	}
	if matched != 1 {
		t.Fatalf("命中实例数应为 1，得到 %d", matched)
	}
	if got := decode[DisplayCommand](t, a.recvEvent(EventDisplayCommand)); got.Action != CmdGoto {
		t.Fatalf("被点名的实例应收到 goto，得到 %s", got.Action)
	}
	// 没被点名的实例一条都不能收到（心跳之外的静默是这条断言的全部意义）。
	if ev, _, ok := b.recvWithin(200 * time.Millisecond); ok {
		t.Fatalf("未被点名的实例不该收到消息，得到 %s", ev)
	}

	insts := env.hub.Instances()
	if got := instancePage(insts, "screen-a"); got != "medals" {
		t.Fatalf("被点名实例的当前页应更新为 medals，得到 %q", got)
	}
	if got := instancePage(insts, "screen-b"); got != "standings" {
		t.Fatalf("未被点名实例的当前页不该变，得到 %q", got)
	}
}

// targets 为空保持广播语义：所有在线实例都收到。
func TestSendCommandEmptyTargetsBroadcasts(t *testing.T) {
	env := newTestEnv(t)
	a := env.dial(t, &helloPayload{Role: RoleDisplay, InstanceID: "screen-a"})
	b := env.dial(t, &helloPayload{Role: RoleDisplay, InstanceID: "screen-b"})
	a.recvEvent(EventSnapshot)
	b.recvEvent(EventSnapshot)

	cmd, err := NewDisplayCommand(CmdLock, map[string]any{})
	if err != nil {
		t.Fatalf("构造 lock: %v", err)
	}
	matched, err := env.hub.SendCommandToInstances(nil, cmd)
	if err != nil {
		t.Fatalf("广播: %v", err)
	}
	if matched != 2 {
		t.Fatalf("广播应命中 2 个在线实例，得到 %d", matched)
	}
	for _, c := range []*wsClient{a, b} {
		if got := decode[DisplayCommand](t, c.recvEvent(EventDisplayCommand)); got.Action != CmdLock {
			t.Fatalf("广播时每个实例都应收到 lock，得到 %s", got.Action)
		}
	}
}

// 目标全都不在线（已断开或 id 不存在）：命中 0，且没有任何连接收到指令。
func TestSendCommandTargetsOfflineInstances(t *testing.T) {
	env := newTestEnv(t)
	a := env.dial(t, &helloPayload{Role: RoleDisplay, InstanceID: "screen-a"})
	a.recvEvent(EventSnapshot)
	a.close()
	waitFor(t, "实例离线", func() bool {
		insts := env.hub.Instances()
		return len(insts) == 1 && !insts[0].Online
	})

	cmd, err := NewDisplayCommand(CmdReload, map[string]any{})
	if err != nil {
		t.Fatalf("构造 reload: %v", err)
	}
	matched, err := env.hub.SendCommandToInstances([]string{"screen-a", "never-seen"}, cmd)
	if err != nil {
		t.Fatalf("定向下发: %v", err)
	}
	if matched != 0 {
		t.Fatalf("离线/不存在的实例不该计入命中，得到 %d", matched)
	}
}

// 目标里混了在线与离线：在线的收到，命中数只算在线的。
func TestSendCommandTargetsMixedOnlineOffline(t *testing.T) {
	env := newTestEnv(t)
	a := env.dial(t, &helloPayload{Role: RoleDisplay, InstanceID: "screen-a"})
	b := env.dial(t, &helloPayload{Role: RoleDisplay, InstanceID: "screen-b"})
	a.recvEvent(EventSnapshot)
	b.recvEvent(EventSnapshot)
	b.close()
	waitFor(t, "screen-b 离线", func() bool {
		for _, st := range env.hub.Instances() {
			if st.ID == "screen-b" {
				return !st.Online
			}
		}
		return false
	})

	cmd, err := NewDisplayCommand(CmdGoto, map[string]string{"page": "results"})
	if err != nil {
		t.Fatalf("构造 goto: %v", err)
	}
	matched, err := env.hub.SendCommandToInstances([]string{"screen-a", "screen-b"}, cmd)
	if err != nil {
		t.Fatalf("定向下发: %v", err)
	}
	if matched != 1 {
		t.Fatalf("只应在线的 screen-a 计入命中，得到 %d", matched)
	}
	if got := decode[DisplayCommand](t, a.recvEvent(EventDisplayCommand)); got.Action != CmdGoto {
		t.Fatalf("在线的实例应收到 goto，得到 %s", got.Action)
	}
}

// 非法 action 走定向通道要报错，不能静默丢掉。
func TestSendCommandTargetsRejectsUnknownAction(t *testing.T) {
	env := newTestEnv(t)
	if _, err := env.hub.SendCommandToInstances([]string{"screen-a"}, DisplayCommand{Action: "nope"}); err == nil {
		t.Fatal("非法 action 应报错")
	}
}

// 屏号上报：Console 窗口带 screen，局域网浏览器不带（空串），两者都要能查出来。
func TestDisplayInstancesReportScreen(t *testing.T) {
	env := newTestEnv(t)
	console := env.dial(t, &helloPayload{Role: RoleDisplay, InstanceID: "console-1", Page: "standings", Screen: "2"})
	lan := env.dial(t, &helloPayload{Role: RoleDisplay, InstanceID: "lan-1", Page: "medals"})
	console.recvEvent(EventSnapshot)
	lan.recvEvent(EventSnapshot)

	byID := make(map[string]Instance)
	for _, st := range env.hub.Instances() {
		byID[st.ID] = st
	}
	if got := byID["console-1"].Screen; got != "2" {
		t.Fatalf("Console 窗口的屏号应为 2，得到 %q", got)
	}
	if got := byID["lan-1"].Screen; got != "" {
		t.Fatalf("局域网浏览器不带屏号，应为空串，得到 %q", got)
	}
}

// sessionCheckerFunc 把桩会话复查器写成普通函数：用例里只需回答「这条会话还在不在」。
type sessionCheckerFunc func(ctx context.Context, sessionID string) (bool, error)

func (f sessionCheckerFunc) SessionActive(ctx context.Context, sessionID string) (bool, error) {
	return f(ctx, sessionID)
}

// ---------- 会话撤销：长连接必须跟着断开 ----------

// 会话被撤销（登出 / 改密 / 管理员重置密码 / 停用用户）后，已经建立的 /ws 连接必须被断开，
// 不能再以原角色继续收推送；管理端连接按 1008 关闭，前端据此停止重连并引导重新登录。
func TestSessionRevokedClosesAdminConnection(t *testing.T) {
	env := newTestEnvAsSession(t, &auth.User{ID: 9, Username: "boss", Role: auth.RoleAdmin, Active: true}, "sid-1")
	env.hub.SetHeartbeatInterval(50 * time.Millisecond)

	var revoked atomic.Bool
	env.hub.SetSessionChecker(sessionCheckerFunc(func(context.Context, string) (bool, error) {
		return !revoked.Load(), nil
	}))

	c := env.dial(t, &helloPayload{Role: RoleAdmin})
	c.recvEvent(EventSnapshot)

	// 会话仍有效时照常收推送：复查不能误伤在线用户。
	insertChange(t, env.db, "teams", 1, ActionUpsert, `{"id":1}`)
	env.hub.Broadcast(loadChanges(t, env.hub, 0, 0))
	c.recvEvent(EventChange)

	// 撤销之后：最迟一个心跳周期内断开，关闭码为「登录失效」。
	revoked.Store(true)
	c.expectClosedWithCodeDraining(3*time.Second, websocket.StatusPolicyViolation)

	// 断开后这条连接不再是任何广播的投递对象：后续变更推不到它。
	waitFor(t, "失效连接退出", func() bool { return len(env.hub.roomClients("")) == 0 })
	insertChange(t, env.db, "teams", 2, ActionUpsert, `{"id":2}`)
	env.hub.Broadcast(loadChanges(t, env.hub, 0, 0))
	if n := len(env.hub.roomClients("")); n != 0 {
		t.Fatalf("会话失效后不应再有投递对象，得到 %d 条连接", n)
	}
}

// 大屏连接（role=display 但带会话）同样要断开，但用 1001：客户端会自动重连，
// 重连时握手按会话失效降级成 guest，现场大屏不该因为操作员登出而永久冻结。
func TestSessionRevokedClosesDisplayConnectionAsGoingAway(t *testing.T) {
	env := newTestEnvAsSession(t, &auth.User{ID: 9, Username: "boss", Role: auth.RoleAdmin, Active: true}, "sid-2")
	env.hub.SetHeartbeatInterval(50 * time.Millisecond)

	var revoked atomic.Bool
	env.hub.SetSessionChecker(sessionCheckerFunc(func(context.Context, string) (bool, error) {
		return !revoked.Load(), nil
	}))

	c := env.dial(t, &helloPayload{Role: RoleDisplay, InstanceID: "screen-1"})
	c.recvEvent(EventSnapshot)

	revoked.Store(true)
	c.expectClosedWithCodeDraining(3*time.Second, websocket.StatusGoingAway)
}

// 没有会话的连接（局域网大屏）不做复查：撤销语义不适用于 guest，也不该白查一次库。
func TestSessionCheckSkippedWithoutSession(t *testing.T) {
	env := newTestEnv(t)
	env.hub.SetHeartbeatInterval(50 * time.Millisecond)
	env.hub.SetSessionChecker(sessionCheckerFunc(func(context.Context, string) (bool, error) {
		return false, nil // 一旦真被问到就会判死，但 guest 连接根本不该被问
	}))

	c := env.dial(t, &helloPayload{Role: RoleDisplay, InstanceID: "lan-1"})
	c.recvEvent(EventSnapshot)

	if ev, _, ok := c.recvWithin(300 * time.Millisecond); !ok || ev != EventHeartbeat {
		t.Fatalf("guest 连接应继续收心跳，得到 %q ok=%v", ev, ok)
	}
}

// 复查失败（系统库忙 / 被长操作占住）按「仍有效」处理：一次查询抖动不该把大屏整片踢下线。
func TestSessionCheckFailureKeepsConnection(t *testing.T) {
	env := newTestEnvAsSession(t, &auth.User{ID: 9, Username: "boss", Role: auth.RoleAdmin, Active: true}, "sid-3")
	env.hub.SetHeartbeatInterval(50 * time.Millisecond)
	env.hub.SetSessionChecker(sessionCheckerFunc(func(context.Context, string) (bool, error) {
		return false, errors.New("库忙")
	}))

	c := env.dial(t, &helloPayload{Role: RoleAdmin})
	c.recvEvent(EventSnapshot)
	if ev, _, ok := c.recvWithin(300 * time.Millisecond); !ok || ev != EventHeartbeat {
		t.Fatalf("复查失败时应保留连接并继续心跳，得到 %q ok=%v", ev, ok)
	}
}

// ---------- 增量载荷按角色裁剪 ----------

// fullSettingsPayload 是 settings 实体的变更载荷：既有 admin 专属的系统配置，
// 也有大屏要用的展示字段（公告、标题、榜单开关）。
const fullSettingsPayload = `{"event":{"name":"测试运动会","subtitle":null,"host":null,"logoPath":null,"primaryColor":null},` +
	`"display":{"background":null,"logo":null,"title":"测试运动会","subtitle":null,"primaryColor":null,` +
	`"rotateSeconds":15,"announce":"请注意检录","fontTitle":null,"fontBody":null,"liveEvents":[],"liveSource":"auto"},` +
	`"standings":{"individualToTeamWeight":0.5,"rankingMode":"points","showIndividualBoard":true,` +
	`"showIndividualMedals":true,"liveOnPublish":true},"defaultProfileId":1,"requireApproval":true}`

// assertSettingsPayload 校验一条 settings 载荷是否按身份裁过：
// 非 admin 不得出现默认积分模板 / 审核开关 / 个人折算系数，但公告与标题这类大屏字段必须保留。
func assertSettingsPayload(t *testing.T, raw json.RawMessage, wantFull bool) {
	t.Helper()
	var got dto.EventSettings
	if err := json.Unmarshal(raw, &got); err != nil {
		t.Fatalf("settings 载荷无法解析: %s", raw)
	}
	if wantFull {
		if got.DefaultProfileID == nil || !got.RequireApproval || got.Standings.IndividualToTeamWeight != 0.5 {
			t.Fatalf("admin 应拿到完整系统配置: %s", raw)
		}
	} else {
		if got.DefaultProfileID != nil {
			t.Fatalf("非 admin 不应看到 defaultProfileId: %s", raw)
		}
		if got.RequireApproval {
			t.Fatalf("非 admin 不应看到 requireApproval: %s", raw)
		}
		if got.Standings.IndividualToTeamWeight != 0 {
			t.Fatalf("非 admin 不应看到 individualToTeamWeight: %s", raw)
		}
	}
	if got.Event.Name != "测试运动会" || got.Display.Announce == nil || *got.Display.Announce != "请注意检录" {
		t.Fatalf("大屏展示字段不该被裁剪: %s", raw)
	}
	if got.Standings.RankingMode != "points" || !got.Standings.ShowIndividualBoard || !got.Standings.LiveOnPublish {
		t.Fatalf("榜单展示开关不该被裁剪: %s", raw)
	}
}

// 快照裁剪之外，change 与 catchup 两条增量路径也必须按角色裁剪：
// 否则任何打开 /ws 的浏览器都能从增量里读到系统配置，快照上的裁剪等于被旁路。
func TestSettingsChangeRedactedByRole(t *testing.T) {
	for _, tc := range []struct {
		name string
		user *auth.User
		role string
		full bool
	}{
		{"guest", &auth.User{Role: auth.RoleGuest}, RoleDisplay, false},
		{"recorder", &auth.User{ID: 3, Username: "rec", Role: auth.RoleRecorder, Active: true}, RoleAdmin, false},
		{"admin", &auth.User{ID: 1, Username: "boss", Role: auth.RoleAdmin, Active: true}, RoleAdmin, true},
	} {
		t.Run(tc.name, func(t *testing.T) {
			env := newTestEnvAs(t, tc.user)

			// 一条历史变更：用来构造 catchup（重连补齐）。先垫一条无关变更，
			// 让下面的 lastVersion 落在增量区间内（0 会走全量快照）。
			insertChange(t, env.db, "result", 1, ActionUpsert, `{"rank":1}`)
			old := insertChange(t, env.db, entitySettings, 0, ActionUpsert, fullSettingsPayload)

			// change：在线连接收到一条新的 settings 变更。
			live := env.dial(t, &helloPayload{Role: tc.role, InstanceID: tc.name + "-live"})
			live.recvEvent(EventSnapshot)
			insertChange(t, env.db, entitySettings, 0, ActionUpsert, fullSettingsPayload)
			sent := loadChanges(t, env.hub, old, 0)
			env.hub.Broadcast(sent)

			got := decode[changePayload](t, live.recvEvent(EventChange))
			if len(got.Changes) != 1 || got.Changes[0].Entity != entitySettings {
				t.Fatalf("应收到一条 settings 变更，得到 %+v", got.Changes)
			}
			assertSettingsPayload(t, got.Changes[0].Data, tc.full)
			// 裁剪必须产出新对象：就地改写共享的 Change 会让 admin 连接也拿到裁剪版。
			if string(sent[0].Data) != fullSettingsPayload {
				t.Fatalf("裁剪不应改写共享的 Change: %s", sent[0].Data)
			}

			// catchup：重连连接按 lastVersion 补齐，同样不能漏出系统配置。
			catchupClient := env.dial(t, &helloPayload{
				Role: tc.role, InstanceID: tc.name + "-catchup", LastVersion: ptrInt64(old - 1),
			})
			cp := decode[catchupPayload](t, catchupClient.recvEvent(EventCatchup))
			var found bool
			for _, ch := range cp.Changes {
				if ch.Entity != entitySettings {
					continue
				}
				found = true
				assertSettingsPayload(t, ch.Data, tc.full)
			}
			if !found {
				t.Fatalf("catchup 应包含 settings 变更，得到 %+v", cp.Changes)
			}
		})
	}
}

// 载荷不是 EventSettings 形状时宁可只下发空对象，也不能把可能含系统配置的原文透给非 admin。
func TestSettingsChangeUnknownShapeRedacted(t *testing.T) {
	env := newTestEnv(t)
	c := env.dial(t, &helloPayload{Role: RoleDisplay, InstanceID: "d1"})
	c.recvEvent(EventSnapshot)

	insertChange(t, env.db, entitySettings, 0, ActionUpsert, `"旧版扁平配置"`)
	env.hub.Broadcast(loadChanges(t, env.hub, 0, 0))

	got := decode[changePayload](t, c.recvEvent(EventChange))
	if string(got.Changes[0].Data) != `{}` {
		t.Fatalf("无法识别的 settings 载荷应降级成空对象，得到 %s", got.Changes[0].Data)
	}
}

// ---------- 整库快照限流 ----------
//
// 整库快照有两个入口：resync，和不带 lastVersion 的 hello（首包 / 翻页补的那条）。
// 两者在 sendInitial 里共用同一个 per-connection 窗口，所以「交替发」绕不过去。

// 窗口内连发多次 resync 只出一份整库快照；越过窗口后恢复正常，
// 客户端「落后了要一次全量」的补齐行为不受影响。
func TestSnapshotThrottleCoversResync(t *testing.T) {
	env := newTestEnv(t)
	env.hub.SetSnapshotMinInterval(300 * time.Millisecond)
	insertChange(t, env.db, "result", 1, ActionUpsert, `{"rank":1}`)

	// 新连接的第一份快照照发，并且它就是本窗口里的那一份。
	c := env.dial(t, &helloPayload{Role: RoleDisplay, InstanceID: "d1"})
	if snap := decode[Snapshot](t, c.recvEvent(EventSnapshot)); snap.Version != 1 {
		t.Fatalf("新连接应收到版本 1 的快照，得到 %d", snap.Version)
	}
	for i := 0; i < 5; i++ {
		c.send(EventResync, struct{}{})
	}
	if ev, _, ok := c.recvWithin(150 * time.Millisecond); ok {
		t.Fatalf("同一窗口内不该有第二份整库快照，收到 %s", ev)
	}

	// 越过窗口后再连发 5 次：只放行一份。
	time.Sleep(350 * time.Millisecond)
	for i := 0; i < 5; i++ {
		c.send(EventResync, struct{}{})
	}
	decode[Snapshot](t, c.recvEvent(EventSnapshot))
	if ev, _, ok := c.recvWithin(150 * time.Millisecond); ok {
		t.Fatalf("一个窗口只应放行一份整库快照，收到 %s", ev)
	}
}

// 不带 lastVersion 的重复 hello 与 resync 等价：也是要一份整库快照，同样按窗口限流。
func TestSnapshotThrottleCoversRepeatHello(t *testing.T) {
	env := newTestEnv(t)
	env.hub.SetSnapshotMinInterval(300 * time.Millisecond)
	insertChange(t, env.db, "result", 1, ActionUpsert, `{"rank":1}`)

	c := env.dial(t, &helloPayload{Role: RoleDisplay, InstanceID: "d1"})
	if snap := decode[Snapshot](t, c.recvEvent(EventSnapshot)); snap.Version != 1 {
		t.Fatalf("新连接应收到版本 1 的快照，得到 %d", snap.Version)
	}

	for i := 0; i < 5; i++ {
		c.send(EventHello, helloPayload{Role: RoleDisplay, InstanceID: "d1"})
	}
	if ev, _, ok := c.recvWithin(300 * time.Millisecond); ok {
		t.Fatalf("窗口内重复 hello 只应回一份快照（首包那份），收到 %s", ev)
	}

	// 越过窗口后恢复：真实节奏下翻页补的那条 hello 远晚于窗口，照旧生效。
	time.Sleep(350 * time.Millisecond)
	c.send(EventHello, helloPayload{Role: RoleDisplay, InstanceID: "d1"})
	decode[Snapshot](t, c.recvEvent(EventSnapshot))
}

// hello 与 resync 交替发也绕不过限流：两者共用一个窗口，而不是各记各的。
func TestSnapshotThrottleSharedByHelloAndResync(t *testing.T) {
	env := newTestEnv(t)
	env.hub.SetSnapshotMinInterval(time.Minute)
	insertChange(t, env.db, "result", 1, ActionUpsert, `{"rank":1}`)

	c := env.dial(t, &helloPayload{Role: RoleDisplay, InstanceID: "d1"})
	decode[Snapshot](t, c.recvEvent(EventSnapshot))

	resync := func() { c.send(EventResync, struct{}{}) }
	hello := func() { c.send(EventHello, helloPayload{Role: RoleDisplay, InstanceID: "d1"}) }
	resync()
	hello()
	resync()
	hello()

	if ev, _, ok := c.recvWithin(300 * time.Millisecond); ok {
		t.Fatalf("交替 hello / resync 不应绕过整库快照窗口，收到 %s", ev)
	}
}

// 不变式：带 lastVersion 的 hello 走 catchup（增量），与整库快照无关——
// 刚发过快照也不能把它挡住，而且它不会留下「待发快照」的挂起标记。
func TestSnapshotThrottleDoesNotAffectCatchup(t *testing.T) {
	env := newTestEnv(t)
	// 窗口与心跳都调短：万一 catchup 请求被误当成「要整库快照」，下面的
	// noSnapshotWithin 立刻就能看见那次多余的补发。
	env.hub.SetHeartbeatInterval(40 * time.Millisecond)
	env.hub.SetSnapshotMinInterval(150 * time.Millisecond)
	insertChange(t, env.db, "result", 1, ActionUpsert, `{"rank":1}`)

	c := env.dial(t, &helloPayload{Role: RoleDisplay, InstanceID: "d1"})
	snap := decode[Snapshot](t, c.recvEvent(EventSnapshot))

	// 窗口内的第二条 hello 带 lastVersion：必须立刻拿到增量。
	insertChange(t, env.db, "result", 2, ActionUpsert, `{"rank":2}`)
	c.send(EventHello, helloPayload{
		Role: RoleDisplay, InstanceID: "d1", LastVersion: ptrInt64(snap.Version),
	})
	cp := decode[catchupPayload](t, c.recvEvent(EventCatchup))
	if cp.FromVersion != snap.Version || cp.Version != snap.Version+1 {
		t.Fatalf("catchup 应为 %d → %d，得到 %d → %d",
			snap.Version, snap.Version+1, cp.FromVersion, cp.Version)
	}
	if len(cp.Changes) != 1 || cp.Changes[0].ID != 2 {
		t.Fatalf("catchup 应补上那一条新变更，得到 %+v", cp.Changes)
	}
	// catchup 不是整库快照：不该挂起、也不该在窗口过后补发。
	c.noSnapshotWithin(400 * time.Millisecond)
}

// 不变式：窗口按连接记账。一条连接把自己的窗口占满，不影响别的连接的第一份快照；
// 断线重连也是新连接（新的 client，窗口为空），第一份快照同样照发。
func TestSnapshotThrottleIsScopedToConnection(t *testing.T) {
	env := newTestEnv(t)
	env.hub.SetSnapshotMinInterval(time.Minute)
	insertChange(t, env.db, "result", 1, ActionUpsert, `{"rank":1}`)

	// A 连发 resync 把自己的窗口占满。
	a := env.dial(t, &helloPayload{Role: RoleDisplay, InstanceID: "a"})
	a.recvEvent(EventSnapshot)
	for i := 0; i < 3; i++ {
		a.send(EventResync, struct{}{})
	}
	if ev, _, ok := a.recvWithin(200 * time.Millisecond); ok {
		t.Fatalf("A 的窗口内不该再有整库快照，收到 %s", ev)
	}

	// B 是新连接：第一份快照照发，不受 A 影响。
	b := env.dial(t, &helloPayload{Role: RoleDisplay, InstanceID: "b"})
	if snap := decode[Snapshot](t, b.recvEvent(EventSnapshot)); snap.Version != 1 {
		t.Fatalf("B 作为新连接应立刻拿到快照，得到版本 %d", snap.Version)
	}

	// 断线重连同样是新连接：现场刷新页面 / 网络抖动都不该被限流影响。
	a.close()
	reconnected := env.dial(t, &helloPayload{Role: RoleDisplay, InstanceID: "a"})
	if snap := decode[Snapshot](t, reconnected.recvEvent(EventSnapshot)); snap.Version != 1 {
		t.Fatalf("重连应立刻拿到快照，得到版本 %d", snap.Version)
	}
}

// ---------- 被窗口挡下的整库快照：挂起 + 心跳补发 ----------

// 窗口内连发多次 resync：当下不回（否则等于没限流），但请求被挂起，
// 窗口过后由心跳补一份——大屏不会因为限流静默停在旧数据上。
func TestPendingSnapshotFlushedByHeartbeat(t *testing.T) {
	env := newTestEnv(t)
	env.hub.SetHeartbeatInterval(40 * time.Millisecond)
	env.hub.SetSnapshotMinInterval(150 * time.Millisecond)
	insertChange(t, env.db, "result", 1, ActionUpsert, `{"rank":1}`)

	c := env.dial(t, &helloPayload{Role: RoleDisplay, InstanceID: "d1"})
	c.recvEvent(EventSnapshot) // 首包即时发，窗口从这一份开始记账

	for i := 0; i < 3; i++ {
		c.send(EventResync, struct{}{})
	}
	c.noSnapshotWithin(100 * time.Millisecond)

	if !c.awaitSnapshot(2 * time.Second) {
		t.Fatal("窗口过后应由心跳补发一份整库快照")
	}
	// 补发即清标记：之后的心跳不再重复发（否则限流就退化成心跳频率了）。
	c.noSnapshotWithin(300 * time.Millisecond)
}

// 不带 lastVersion 的重复 hello 与 resync 走同一条挂起路径：也只补一份。
func TestPendingSnapshotFromRepeatHelloFlushedOnce(t *testing.T) {
	env := newTestEnv(t)
	env.hub.SetHeartbeatInterval(40 * time.Millisecond)
	env.hub.SetSnapshotMinInterval(150 * time.Millisecond)
	insertChange(t, env.db, "result", 1, ActionUpsert, `{"rank":1}`)

	c := env.dial(t, &helloPayload{Role: RoleDisplay, InstanceID: "d1"})
	c.recvEvent(EventSnapshot)

	for i := 0; i < 5; i++ {
		c.send(EventHello, helloPayload{Role: RoleDisplay, InstanceID: "d1"})
	}
	c.noSnapshotWithin(100 * time.Millisecond)

	if !c.awaitSnapshot(2 * time.Second) {
		t.Fatal("窗口过后应由心跳补发一份整库快照")
	}
	c.noSnapshotWithin(300 * time.Millisecond)
}

// hello 与 resync 交替触发时共用一个挂起标记：补发仍然只有一份。
func TestPendingSnapshotSharedByHelloAndResync(t *testing.T) {
	env := newTestEnv(t)
	env.hub.SetHeartbeatInterval(40 * time.Millisecond)
	env.hub.SetSnapshotMinInterval(150 * time.Millisecond)
	insertChange(t, env.db, "result", 1, ActionUpsert, `{"rank":1}`)

	c := env.dial(t, &helloPayload{Role: RoleDisplay, InstanceID: "d1"})
	c.recvEvent(EventSnapshot)

	for i := 0; i < 3; i++ {
		c.send(EventHello, helloPayload{Role: RoleDisplay, InstanceID: "d1"})
		c.send(EventResync, struct{}{})
	}
	c.noSnapshotWithin(100 * time.Millisecond)

	if !c.awaitSnapshot(2 * time.Second) {
		t.Fatal("窗口过后应由心跳补发一份整库快照")
	}
	c.noSnapshotWithin(300 * time.Millisecond)
}

// 补发的量级仍然是「每窗口一份」：两条连接各自挂起，各自只补一份。
func TestPendingSnapshotFlushIsPerConnection(t *testing.T) {
	env := newTestEnv(t)
	env.hub.SetHeartbeatInterval(40 * time.Millisecond)
	env.hub.SetSnapshotMinInterval(150 * time.Millisecond)
	insertChange(t, env.db, "result", 1, ActionUpsert, `{"rank":1}`)

	a := env.dial(t, &helloPayload{Role: RoleDisplay, InstanceID: "a"})
	b := env.dial(t, &helloPayload{Role: RoleDisplay, InstanceID: "b"})
	a.recvEvent(EventSnapshot)
	b.recvEvent(EventSnapshot)

	a.send(EventResync, struct{}{})
	b.send(EventResync, struct{}{})
	// 只对 A 做一次「当下不回」的断言：两条连接的静默窗口不能串行等待，
	// 后一条会把自己的补发也等进去（B 的那份本来就该在窗口过后到）。
	a.noSnapshotWithin(100 * time.Millisecond)

	if !a.awaitSnapshot(2*time.Second) || !b.awaitSnapshot(2*time.Second) {
		t.Fatal("两条挂起的连接都应在窗口过后各补一份快照")
	}
	a.noSnapshotWithin(300 * time.Millisecond)
	b.noSnapshotWithin(300 * time.Millisecond)
}