Scoreboard/server/internal/realtime/hub.go

package realtime

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"log/slog"
	"net/http"
	"strings"
	"sync"
	"time"

	"github.com/coder/websocket"

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

// 默认参数。
//
// 注意：Set* 目前只有测试在调用——现场调优要求先把它们接到 config.yaml / main.go 上，
// 那是本包之外的决定，所以不要照着 Set* 的存在去宣称「现场可调」。
const (
	defaultHeartbeatInterval = 10 * time.Second
	defaultMaxCatchup        = 500
	defaultSlowLag           = 500
	defaultHelloTimeout      = 15 * time.Second
	defaultWriteTimeout      = 10 * time.Second
	defaultCloseTimeout      = 2 * time.Second
	defaultSendBuffer        = 64
	// defaultLivenessInterval 是「一点客户端帧都没收到」的静默期上限：超过它就发一次
	// WebSocket ping 求证对端还在不在（见 client.livenessLoop）。
	//
	// 不能把「静默」直接当掉线：大屏在数据没变时本来就不上行。取 15s，配合
	// maxProbeFailures = 2 与 defaultPongTimeout = 5s，真正掉线的屏最迟
	// 2 × (15s + 5s) = 40s 内被判定并断开；稳态下每块屏每分钟只多发 4 个 ping 帧。
	defaultLivenessInterval = 15 * time.Second
	// defaultPongTimeout 是等一次 pong 的超时。ping 帧本身要走写锁，正常浏览器在
	// 网络层立刻应答，5s 已经宽裕。
	defaultPongTimeout = 5 * time.Second
	// maxProbeFailures 是连续多少次探测失败才断开：单次失败可能只是对端在忙
	// （正在收一份大快照，写锁被占住），连续失败才说明连接上没有人了。
	maxProbeFailures = 2
	// defaultSnapshotMinInterval 是同一连接两次「整库快照」之间的最小间隔。
	//
	// 取 2.5s，比客户端自己的 3s 冷却（web/src/shared/realtime.ts 的 resyncCooldownMs）短半秒：
	// 客户端按自己的节奏回来时（≥ 它的上次发送 + 3s），服务端的窗口必须已经过期，
	// 否则两个等长的窗口会因收发延迟抖动而被随机判定成「超频」，把正常的补齐请求挡掉。
	// 反过来，2.5s 也已经把放大面压到「每个连接每 2.5 秒最多一份整库快照」
	// （一份 = 一个事务 + 8 条查询，赛事库还是单连接）。
	defaultSnapshotMinInterval = 2500 * time.Millisecond
	// sessionCheckTimeout 是会话复查查询的超时：系统库同样只有一条连接，
	// 被长操作占住时不能把心跳循环一起卡住。
	sessionCheckTimeout = 2 * time.Second
)

// slowCloseCode 是踢掉慢客户端时使用的自定义关闭码。
// 不能用 1008：前端把所有 1008 当「登录失效」并永久停止重连，慢客户端被踢后大屏会冻结。
const slowCloseCode = websocket.StatusCode(4000)

// unresponsiveCloseCode 是踢掉「不回 pong」的连接时使用的关闭码。同样避开 1008：
// 那块屏只是掉线了，重连回来就能补上数据，不该被引导去重新登录。
const unresponsiveCloseCode = websocket.StatusCode(4001)

// Hub 是实时同步中心：维护房间、连接、大屏实例注册表与版本号。
//
// 版本号等于 change_log 的最大 id；写入方提交事务后调用 Broadcast 推送变更，
// 断线客户端用 hello.lastVersion 请求增量补齐。
type Hub struct {
	db  *sql.DB
	log *slog.Logger

	mu        sync.RWMutex
	rooms     map[string]map[*client]struct{}
	clients   map[*client]struct{}
	instances map[string]*instanceState
	version   int64
	closed    bool

	// sendMu 串行化「版本推进 + 下发」，保证同一客户端收到的 change 与版本严格递增，
	// 也避免初始快照与并发广播交错导致漏变更。
	sendMu sync.Mutex

	ctx    context.Context
	cancel context.CancelFunc
	hbWG   sync.WaitGroup // 心跳循环
	connWG sync.WaitGroup // 连接处理

	heartbeatInterval time.Duration
	heartbeatReset    chan struct{}
	maxCatchup        int
	slowLag           int64
	helloTO           time.Duration
	writeTO           time.Duration
	allowedOrigins    []string
	// snapshotMinInterval 是同一连接两次整库快照之间的最小间隔；<=0 表示不限流。
	snapshotMinInterval time.Duration
	// livenessInterval 是「一点客户端帧都没收到」的静默期上限；pongTimeout 是等一次
	// pong 的超时。两者只影响存活探测（见 client.livenessLoop），默认值见常量区注释；
	// livenessInterval <= 0 表示关闭探测（只有测试会这么设）。
	livenessInterval time.Duration
	pongTimeout      time.Duration
	// sessionChecker 复查长连接背后的会话是否还有效；未注入时只在握手时校验一次。
	sessionChecker SessionChecker
}

// SessionChecker 判定一条会话当前是否仍然有效（未被撤销 / 未过期 / 用户未停用）。
// auth.Service 满足这个接口；测试可注入桩实现。
type SessionChecker interface {
	SessionActive(ctx context.Context, sessionID string) (bool, error)
}

// NewHub 创建实时中心并启动心跳循环；db 是赛事库（change_log 所在库）。
// 调用方在服务关闭时应调用 Hub.Close。
func NewHub(db *sql.DB, log *slog.Logger) *Hub {
	if log == nil {
		log = slog.Default()
	}
	ctx, cancel := context.WithCancel(context.Background())
	h := &Hub{
		db:                  db,
		log:                 log,
		rooms:               make(map[string]map[*client]struct{}),
		clients:             make(map[*client]struct{}),
		instances:           make(map[string]*instanceState),
		ctx:                 ctx,
		cancel:              cancel,
		heartbeatInterval:   defaultHeartbeatInterval,
		heartbeatReset:      make(chan struct{}, 1),
		maxCatchup:          defaultMaxCatchup,
		slowLag:             defaultSlowLag,
		helloTO:             defaultHelloTimeout,
		writeTO:             defaultWriteTimeout,
		snapshotMinInterval: defaultSnapshotMinInterval,
		livenessInterval:    defaultLivenessInterval,
		pongTimeout:         defaultPongTimeout,
	}
	if db != nil {
		if v, err := h.currentVersion(context.Background()); err == nil {
			h.version = v
		} else {
			log.Warn("读取初始版本号失败", "err", err)
		}
	}

	h.hbWG.Add(1)
	go h.heartbeatLoop()
	return h
}

// Close 停止心跳、断开所有连接并等待处理中的连接退出。
func (h *Hub) Close() error {
	h.mu.Lock()
	if h.closed {
		h.mu.Unlock()
		return nil
	}
	h.closed = true
	h.mu.Unlock()

	h.cancel()
	for _, c := range h.roomClients("") {
		c.stop(websocket.StatusGoingAway, "服务关闭")
	}
	h.connWG.Wait()
	h.hbWG.Wait()
	return nil
}

// Version 返回内存中记录的当前版本号。
func (h *Hub) Version() int64 {
	h.mu.RLock()
	defer h.mu.RUnlock()
	return h.version
}

func (h *Hub) setVersion(v int64) {
	if v <= 0 {
		return
	}
	h.mu.Lock()
	if v > h.version {
		h.version = v
	}
	h.mu.Unlock()
}

// SetHeartbeatInterval 调整心跳间隔并立即生效。
//
// 唯一调用方是测试（把心跳调快，好在毫秒级跑完「会话撤销即断开」这类用例）；
// 生产路径用的是 defaultHeartbeatInterval。要做成现场可调，得先在 main.go 里
// 从 config.yaml 读出来再调这里——本包不读配置，也没有别的消费方。
func (h *Hub) SetHeartbeatInterval(d time.Duration) {
	if d <= 0 {
		return
	}
	h.mu.Lock()
	h.heartbeatInterval = d
	h.mu.Unlock()

	select {
	case h.heartbeatReset <- struct{}{}:
	default:
	}
}

// SetMaxCatchup 设置增量补发的条数上限；超过就回落全量快照。同样只有测试在调用。
func (h *Hub) SetMaxCatchup(n int) {
	if n <= 0 {
		return
	}
	h.mu.Lock()
	h.maxCatchup = n
	h.mu.Unlock()
}

// SetSlowLag 设置「慢客户端」的版本落后阈值；<=0 表示不判定。同样只有测试在调用。
func (h *Hub) SetSlowLag(n int64) {
	h.mu.Lock()
	h.slowLag = n
	h.mu.Unlock()
}

// SetAllowedOrigins 设置 WebSocket 跨站 Origin 白名单（path.Match 模式）；
// 同源与不带 Origin 的握手始终放行，不受此设置影响。
func (h *Hub) SetAllowedOrigins(origins []string) {
	h.mu.Lock()
	h.allowedOrigins = append([]string(nil), origins...)
	h.mu.Unlock()
}

// SetSessionChecker 注入会话复查器：连接建立后周期性复查会话是否仍然有效，失效即断开。
//
// /ws 的身份在握手时固化，而会话随时可能被撤销（登出、改密、管理员重置密码、停用用户）。
// 不复查的话，那条长连接会一直以原角色收推送——被停用的人只要不关页面，管理端就还挂着。
func (h *Hub) SetSessionChecker(c SessionChecker) {
	h.mu.Lock()
	h.sessionChecker = c
	h.mu.Unlock()
}

// SetSnapshotMinInterval 设置同一连接两次整库快照之间的最小间隔；<= 0 表示不限流（仅测试用）。
//
// 撞到窗口的请求不会拿到第二份全量：客户端本来就刚拿到过一份，心跳里的版本号会让它
// 在需要时重新发起；越过窗口后一切照旧。
func (h *Hub) SetSnapshotMinInterval(d time.Duration) {
	h.mu.Lock()
	h.snapshotMinInterval = d
	h.mu.Unlock()
}

func (h *Hub) sessionCheckerValue() SessionChecker {
	h.mu.RLock()
	defer h.mu.RUnlock()
	return h.sessionChecker
}

func (h *Hub) snapshotMinIntervalValue() time.Duration {
	h.mu.RLock()
	defer h.mu.RUnlock()
	return h.snapshotMinInterval
}

func (h *Hub) heartbeatIntervalValue() time.Duration {
	h.mu.RLock()
	defer h.mu.RUnlock()
	return h.heartbeatInterval
}

func (h *Hub) maxCatchupValue() int {
	h.mu.RLock()
	defer h.mu.RUnlock()
	return h.maxCatchup
}

func (h *Hub) slowLagValue() int64 {
	h.mu.RLock()
	defer h.mu.RUnlock()
	return h.slowLag
}

func (h *Hub) helloTimeout() time.Duration {
	h.mu.RLock()
	defer h.mu.RUnlock()
	return h.helloTO
}

func (h *Hub) writeTimeout() time.Duration {
	h.mu.RLock()
	defer h.mu.RUnlock()
	return h.writeTO
}

func (h *Hub) livenessIntervalValue() time.Duration {
	h.mu.RLock()
	defer h.mu.RUnlock()
	return h.livenessInterval
}

func (h *Hub) pongTimeoutValue() time.Duration {
	h.mu.RLock()
	defer h.mu.RUnlock()
	return h.pongTimeout
}

func (h *Hub) allowedOriginsValue() []string {
	h.mu.RLock()
	defer h.mu.RUnlock()
	return h.allowedOrigins
}

// Handler 返回 WebSocket 处理器，由调用方挂到 /ws 上。
func (h *Hub) Handler() http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		conn, err := websocket.Accept(w, r, &websocket.AcceptOptions{
			// 同源与不带 Origin 的握手由库放行；跨站需命中 server.allowedOrigins 白名单。
			OriginPatterns:  h.allowedOriginsValue(),
			CompressionMode: websocket.CompressionDisabled,
		})
		if err != nil {
			h.log.Debug("WebSocket 握手失败", "remote", r.RemoteAddr, "origin", r.Header.Get("Origin"), "err", err)
			return
		}

		h.mu.Lock()
		if h.closed {
			h.mu.Unlock()
			_ = conn.Close(websocket.StatusGoingAway, "服务关闭")
			return
		}
		h.connWG.Add(1)
		h.mu.Unlock()
		defer h.connWG.Done()

		// 身份只认会话：/ws 必须被 Authenticate 中间件包住，否则一律按 guest 处理。
		h.serve(conn, r.RemoteAddr, wsIdentityFromContext(r.Context()))
	})
}

// wsIdentity 是握手时从会话取到的身份；整条连接的生命周期都用它。
//
// sessionID 不进协议，只为会话复查保留：会话被撤销后要据此找到并断开对应的连接。
type wsIdentity struct {
	Role          string
	UserID        int64
	SessionID     string
	SettingsAdmin bool
}

// wsIdentityFromContext 从请求上下文取会话身份并折算成实时层角色。
// 未登录（guest）只能进大屏房间；recorder/admin 才有管理端房间的订阅权。
// settingsAdmin 单独返回真实的 admin 判定：recorder 能进管理端房间，但读不到系统配置。
func wsIdentityFromContext(ctx context.Context) wsIdentity {
	id := wsIdentity{}
	if sess := middleware.Session(ctx); sess != nil {
		id.SessionID = sess.ID
	}
	u := middleware.User(ctx)
	if u == nil || !u.Role.Allows(auth.RoleRecorder) {
		// 会话无效时 Authenticate 已经降级成 guest，这里拿不到会话 id，
		// 因此降级后的连接不会被会话复查误伤。
		id.Role = RoleDisplay
		return id
	}
	id.Role = RoleAdmin
	id.UserID = u.ID
	id.SettingsAdmin = u.Role.Allows(auth.RoleAdmin)
	return id
}

func (h *Hub) serve(conn *websocket.Conn, remoteAddr string, id wsIdentity) {
	conn.SetReadLimit(1 << 20)

	c := newClient(h, conn, remoteAddr, id)
	h.addClient(c)

	ctx, cancel := context.WithCancel(h.ctx)
	defer cancel()

	var loops sync.WaitGroup
	loops.Add(3)
	go func() { defer loops.Done(); c.readLoop(ctx) }()
	go func() { defer loops.Done(); c.writeLoop(ctx) }()
	// 存活探测：服务端「连接对象还在」不等于对端还在，注册表的在线判定靠这条循环求证。
	go func() { defer loops.Done(); c.livenessLoop(ctx) }()

	<-c.done
	// 先做关闭握手，再取消读上下文：coder/websocket 在读上下文被取消时会直接
	// 关闭底层连接，那样对端只能看到 EOF，收不到策略性关闭码（如 1008）。
	// 写循环已因 c.done 退出，读循环会随对端回帧自然结束。
	c.mu.Lock()
	code, reason := c.closeCode, c.closeReason
	c.mu.Unlock()
	if code == 0 {
		code = websocket.StatusNormalClosure
	}
	closed := make(chan struct{})
	go func() {
		defer close(closed)
		_ = conn.Close(code, reason)
	}()
	select {
	case <-closed:
	case <-time.After(defaultCloseTimeout):
		// 对端不回应：cancel 会强制关闭底层连接，读循环随之退出。
	}

	cancel()
	loops.Wait()
	h.removeClient(c)
}

func (h *Hub) addClient(c *client) {
	h.mu.Lock()
	h.clients[c] = struct{}{}
	h.mu.Unlock()
}

func (h *Hub) removeClient(c *client) {
	h.mu.Lock()
	delete(h.clients, c)
	for room, set := range h.rooms {
		delete(set, c)
		if len(set) == 0 {
			delete(h.rooms, room)
		}
	}
	h.mu.Unlock()

	h.releaseDisplay(c)
}

// join 把连接加入房间；房间用内存 map 维护，不引入实时框架。
func (h *Hub) join(c *client, rooms ...string) {
	h.mu.Lock()
	defer h.mu.Unlock()
	for _, room := range rooms {
		if room == "" {
			continue
		}
		set := h.rooms[room]
		if set == nil {
			set = make(map[*client]struct{})
			h.rooms[room] = set
		}
		set[c] = struct{}{}
	}
}

// roomClients 返回房间成员的快照；room 为空表示全部连接。
func (h *Hub) roomClients(room string) []*client {
	h.mu.RLock()
	defer h.mu.RUnlock()

	if room == "" {
		out := make([]*client, 0, len(h.clients))
		for c := range h.clients {
			out = append(out, c)
		}
		return out
	}
	set := h.rooms[room]
	out := make([]*client, 0, len(set))
	for c := range set {
		out = append(out, c)
	}
	return out
}

// handleHello 只信任会话身份：客户端自报的 role/userId 会被忽略，
// 管理端房间必须由登录会话授予，未登录连接请求管理端直接断开。
func (h *Hub) handleHello(ctx context.Context, c *client, p helloPayload) {
	c.mu.Lock()
	sessionRole, sessionUserID := c.sessionRole, c.sessionUserID
	c.mu.Unlock()

	if p.Role != "" && p.Role != RoleDisplay && p.Role != RoleAdmin {
		h.log.Debug("hello 角色非法", "role", p.Role, "remote", c.remoteAddr)
		c.stop(websocket.StatusPolicyViolation, "role 非法")
		return
	}
	if p.Role == RoleAdmin && sessionRole != RoleAdmin {
		h.log.Warn("未登录连接请求管理端房间，已断开", "remote", c.remoteAddr)
		c.stop(websocket.StatusPolicyViolation, "无权订阅管理端房间")
		return
	}

	// 不写 role 时按会话上限走；显式 role=display 允许管理端用户降级成大屏。
	role := sessionRole
	if p.Role == RoleDisplay {
		role = RoleDisplay
	}

	c.mu.Lock()
	c.role = role
	c.userID = 0
	if role == RoleAdmin {
		c.userID = sessionUserID
	}
	c.mu.Unlock()

	if role == RoleAdmin {
		rooms := []string{RoomAdmin}
		if c.userID > 0 {
			rooms = append(rooms, AdminRoom(c.userID))
		}
		h.join(c, rooms...)
	} else {
		id := p.InstanceID
		if id == "" {
			id = p.DisplayKey
		}
		h.registerDisplay(c, id, p.Page, p.Screen)
		h.join(c, RoomDisplay)
	}

	h.sendInitial(ctx, c, p.LastVersion)
}

// sendInitial 决定首包是 snapshot 还是 catchup：
// 没有 lastVersion、版本超前、或增量条数超阈值，一律回落全量快照。
//
// 两个入口共用这一条路径：hello（首包 / 翻页补的那条）与 resync。
func (h *Hub) sendInitial(ctx context.Context, c *client, lastVersion *int64) {
	h.sendMu.Lock()
	defer h.sendMu.Unlock()

	c.mu.Lock()
	c.greeted = true
	c.mu.Unlock()

	cur, err := h.currentVersion(ctx)
	if err != nil {
		if dbClosed(err) {
			// 恢复/重启窗口（库已经被关掉、文件还没换完）里的新连接不能踢：
			// 现场看到的是「每块屏刚连上就被 1011 打回、日志一屏 ERROR」，像恢复把系统
			// 搞坏了。与 httpapi/realtime.go 的降级口径一致——这类错误只记 Debug，
			// 连接留着（心跳照发，客户端看门狗不会误判），等库回来后的下一次
			// resync / 重连自然就能拿到数据。
			h.log.Debug("跳过首包：库已关闭（恢复/重启中）", "remote", c.remoteAddr, "err", err)
			return
		}
		h.log.Error("读取版本号失败，无法下发首包", "err", err)
		c.stop(websocket.StatusInternalError, "版本号不可用")
		return
	}
	h.setVersion(cur)

	if lastVersion == nil || *lastVersion <= 0 || *lastVersion > cur {
		h.sendSnapshotGuarded(ctx, c)
		return
	}

	n, err := h.countChangesSince(ctx, *lastVersion)
	if err != nil {
		h.log.Warn("统计增量失败，回落快照", "err", err)
		h.sendSnapshotGuarded(ctx, c)
		return
	}
	if n > h.maxCatchupValue() {
		h.sendSnapshotGuarded(ctx, c)
		return
	}

	changes, err := h.changesSince(ctx, *lastVersion, 0)
	if err != nil {
		h.log.Warn("读取增量失败，回落快照", "err", err)
		h.sendSnapshotGuarded(ctx, c)
		return
	}
	// catchup 与 change 一样按连接身份裁剪：重连补齐不能成为读系统配置的后门。
	payload := catchupPayload{
		FromVersion: *lastVersion,
		Version:     cur,
		Changes:     redactChanges(changes, c.settingsAdmin),
	}
	if !c.sendEvent(EventCatchup, payload) {
		return
	}
	c.lastSent.Store(cur)
	c.ackVersion.Store(cur)
}

// sendSnapshotGuarded 是「要发整库快照」的唯一出口（调用方须持有 sendMu，同 sendSnapshotLocked）。
//
// 整库快照的代价是一个 DB 事务 + 8 条查询，而赛事库是单连接（SetMaxOpenConns(1)），
// 读取还在全局发送锁里：不限流的话，一条 15 字节的帧（resync，或重复的不带 lastVersion 的
// hello）就能反复触发，把库连接与所有人的推送一起堵住。这里按连接记账，窗口内已经发过
// 整库快照就不再发第二份。
//
// 只拦「整库快照」这一种：带 lastVersion 的 hello 走 catchup（增量）不受影响；
// 每次断线重连都是新连接（新的 client，窗口为空），第一份快照照发。
//
// 被挡下的请求不丢弃，只在连接上挂一个标记，由心跳循环补发（见 flushPendingSnapshot）：
// 请求方拿到的是「稍微晚一点的一份」，而不是「什么都没有」。
func (h *Hub) sendSnapshotGuarded(ctx context.Context, c *client) bool {
	interval := h.snapshotMinIntervalValue()
	if !c.allowSnapshot(interval) {
		c.snapshotPending.Store(true)
		h.log.Debug("整库快照窗口内，改为心跳补发", "remote", c.remoteAddr, "interval", interval.String())
		return false
	}
	// 这一份已经满足了挂起的请求（否则它会在下个心跳里变成多余的第三份）。
	c.snapshotPending.Store(false)
	h.sendSnapshotLocked(ctx, c)
	return true
}

// flushPendingSnapshot 在心跳里补发被窗口挡下的整库快照；返回 true 表示补发了一份。
//
// 规则：只有在「有待发标记」且「窗口确实已过」时才发，发完清标记——
// 不能做成每次心跳都发一份（那等于把限流从 2.5s 放大成心跳频率）。
// 最迟一个心跳周期（默认 10s）内一定补上，正常客户端感知不到差别。
func (h *Hub) flushPendingSnapshot(ctx context.Context, c *client) bool {
	if !c.snapshotPending.Load() {
		return false
	}
	h.sendMu.Lock()
	defer h.sendMu.Unlock()

	// 加锁后再看一次：期间可能有别的路径已经发过（那时标记已被清掉）。
	if !c.snapshotPending.Load() {
		return false
	}
	if !c.allowSnapshot(h.snapshotMinIntervalValue()) {
		// 窗口还没过（刚有别的请求发过一份）：留给下一个心跳。
		return false
	}
	c.snapshotPending.Store(false)
	h.log.Debug("心跳补发挂起的整库快照", "remote", c.remoteAddr)
	h.sendSnapshotLocked(ctx, c)
	return true
}

// sendSnapshotLocked 必须在持有 sendMu 时调用：快照读取与「已发送版本」的推进
// 必须原子完成，否则并发的 change 会插到快照前面。
func (h *Hub) sendSnapshotLocked(ctx context.Context, c *client) {
	snap, err := h.Snapshot(ctx)
	if err != nil {
		if dbClosed(err) {
			// 同 sendInitial：恢复窗口里「快照读不出来」不是故障，是停机动作的一部分。
			// 这里既不能记 ERROR，也不能拿 1011 把大屏踢回去重连（那正是现场
			// 「恢复把系统搞坏了」的观感来源）；连接留着，心跳继续。
			h.log.Debug("跳过整库快照：库已关闭（恢复/重启中）", "remote", c.remoteAddr, "err", err)
			return
		}
		h.log.Error("构建快照失败", "err", err)
		c.stop(websocket.StatusInternalError, "快照不可用")
		return
	}
	// 快照对所有角色下发，但系统配置只有 admin 能看：非 admin 只保留大屏渲染所需字段。
	if !c.settingsAdmin {
		snap.Settings = snap.Settings.WithoutAdminFields()
	}
	if !c.sendEvent(EventSnapshot, snap) {
		return
	}
	c.lastSent.Store(snap.Version)
	c.ackVersion.Store(snap.Version)
	h.setVersion(snap.Version)
}

// Broadcast 把变更推给所有已握手的连接（大屏与管理端）。
func (h *Hub) Broadcast(changes []Change) { h.fanout("", changes) }

// BroadcastTo 把变更推给指定房间（如 RoomAdmin、AdminRoom(uid)）。
//
// 同样只有测试在调用：生产的推送要么是 Broadcast（所有已握手连接），要么是
// SendCommandTo（大屏指令）。保留它是为了让「房间」这层语义有可用的出口，
// 删掉或接到生产调用方都要连 devdocs/05 §1 一起改（审查清单 C36）。
func (h *Hub) BroadcastTo(room string, changes []Change) { h.fanout(room, changes) }

// fanout 在 sendMu 下按版本去重后下发：已经发过的版本不再重复投递。
func (h *Hub) fanout(room string, changes []Change) {
	if len(changes) == 0 {
		return
	}

	h.sendMu.Lock()
	defer h.sendMu.Unlock()

	version := int64(0)
	for _, ch := range changes {
		if ch.Version > version {
			version = ch.Version
		}
	}
	if version == 0 {
		version = h.currentVersionCached(h.ctx)
	}
	h.setVersion(version)

	for _, c := range h.roomClients(room) {
		c.mu.Lock()
		greeted := c.greeted
		c.mu.Unlock()
		if !greeted {
			continue
		}

		last := c.lastSent.Load()
		settingsAdmin := c.settingsAdmin
		filtered := make([]Change, 0, len(changes))
		for _, ch := range changes {
			if ch.Version > 0 && ch.Version <= last {
				continue
			}
			filtered = append(filtered, redactChange(ch, settingsAdmin))
		}
		if len(filtered) == 0 {
			continue
		}
		if !c.sendEvent(EventChange, changePayload{Version: version, Changes: filtered}) {
			continue
		}
		c.lastSent.Store(version)
		h.checkSlow(c, version)
	}
}

// SendCommand 向所有大屏下发 display:command。
func (h *Hub) SendCommand(cmd DisplayCommand) { h.SendCommandTo(RoomDisplay, cmd) }

// SendCommandTo 向指定房间下发 display:command；非法 action 直接丢弃并记录。
func (h *Hub) SendCommandTo(room string, cmd DisplayCommand) {
	if !ValidDisplayAction(cmd.Action) {
		h.log.Warn("忽略非法的大屏指令", "action", cmd.Action, "room", room)
		return
	}
	if cmd.Action == CmdGoto && (room == "" || room == RoomDisplay) {
		h.setInstancePage(pageFromPayload(cmd.Payload), nil)
	}

	h.sendMu.Lock()
	defer h.sendMu.Unlock()
	for _, c := range h.roomClients(room) {
		c.mu.Lock()
		greeted := c.greeted
		c.mu.Unlock()
		if !greeted {
			continue
		}
		c.sendEvent(EventDisplayCommand, cmd)
	}
}

// ErrUnknownAction 表示指令的 action 不在协议清单内。
// 它不是服务端异常：action 是请求体入参，HTTP 层据此用 errors.Is 把它映射成 400，
// 只有真正的下发失败（写 socket 失败等）才该按 5xx 处理。
var ErrUnknownAction = errors.New("realtime: 未知的大屏指令")

// SendCommandToInstances 按实例定向下发 display:command，返回命中的实例数。
//
// targets 是要控制的 instanceID（GET /api/display/instances 里的 id）；为空时退化成广播，
// 公告这类全屏指令靠的就是广播，这条语义不能改。已离线的实例不计入命中，
// 调用方据此告诉现场「那块屏没收到」。非法 action 返回错误而不是静默丢弃，
// 定向指令需要一个明确的失败信号。
func (h *Hub) SendCommandToInstances(targets []string, cmd DisplayCommand) (int, error) {
	if !ValidDisplayAction(cmd.Action) {
		// 包上哨兵，调用方才能把「入参非法」与「下发失败」分开映射成 4xx / 5xx。
		return 0, fmt.Errorf("未知的大屏指令 %q: %w", cmd.Action, ErrUnknownAction)
	}
	if len(targets) == 0 {
		h.SendCommand(cmd)
		return h.onlineDisplayCount(), nil
	}

	// 归一化只做一次，并且**匹配与写注册表共用同一个值**：管理端从表格里带过来的 id
	// 常有多余空白（粘贴/换行），以前匹配用 TrimSpace、写「当前页」用原值，
	// 结果是指令真的发到了那块屏，注册表里的当前页却停在旧值上，
	// 管理端下一次轮询又把界面上刚切过去的页显示回来。
	wanted := make(map[string]bool, len(targets))
	normalized := make([]string, 0, len(targets))
	for _, id := range targets {
		if id = strings.TrimSpace(id); id != "" {
			if !wanted[id] {
				wanted[id] = true
				normalized = append(normalized, id)
			}
		}
	}

	h.sendMu.Lock()
	matched := make(map[string]bool, len(wanted))
	for _, c := range h.roomClients(RoomDisplay) {
		c.mu.Lock()
		greeted, id := c.greeted, c.instanceID
		c.mu.Unlock()
		if !greeted || !wanted[id] {
			continue
		}
		c.sendEvent(EventDisplayCommand, cmd)
		matched[id] = true
	}
	h.sendMu.Unlock()

	if cmd.Action == CmdGoto {
		h.setInstancePage(pageFromPayload(cmd.Payload), normalized)
	}
	return len(matched), nil
}

// onlineDisplayCount 统计已握手且在线的实例数，用于告知调用方广播命中了几块屏。
func (h *Hub) onlineDisplayCount() int {
	seen := make(map[string]bool)
	for _, c := range h.roomClients(RoomDisplay) {
		c.mu.Lock()
		greeted, id := c.greeted, c.instanceID
		c.mu.Unlock()
		if greeted && id != "" {
			seen[id] = true
		}
	}
	return len(seen)
}

// checkSlow 用 ack 版本差识别慢客户端；超阈值直接断开，由客户端重连后补齐。
func (h *Hub) checkSlow(c *client, version int64) {
	lag := h.slowLagValue()
	if lag <= 0 {
		return
	}
	if version-c.ackVersion.Load() <= lag {
		return
	}
	h.markSlow(c)
}

func (h *Hub) markSlow(c *client) {
	if c.slow.Swap(true) {
		return
	}
	c.mu.Lock()
	id := c.instanceID
	c.mu.Unlock()

	h.mu.Lock()
	if st := h.instances[id]; st != nil {
		st.slow = true
	}
	h.mu.Unlock()

	h.log.Warn("慢客户端：ack 长时间落后，断开后由重连补齐",
		"instance", id, "ackVersion", c.ackVersion.Load(), "remote", c.remoteAddr)
	c.stop(slowCloseCode, "ack 落后")
}

// heartbeatLoop 每 10s 下发一次心跳（version + serverTime）。
func (h *Hub) heartbeatLoop() {
	defer h.hbWG.Done()

	timer := time.NewTimer(h.heartbeatIntervalValue())
	defer timer.Stop()

	for {
		select {
		case <-h.ctx.Done():
			return
		case <-timer.C:
			h.beat()
			timer.Reset(h.heartbeatIntervalValue())
		case <-h.heartbeatReset:
			if !timer.Stop() {
				select {
				case <-timer.C:
				default:
				}
			}
			timer.Reset(h.heartbeatIntervalValue())
		}
	}
}

func (h *Hub) beat() {
	version := h.currentVersionCached(h.ctx)
	now := time.Now()
	data := heartbeatPayload{Version: version, ServerTime: nowISO(now)}

	// 挂起补发单独收集，等心跳全部发完再做：补发要现建一份快照（慢，且要抢 sendMu），
	// 不能让排在后面的大屏连心跳都跟着等（落线判定的看门狗就在客户端那边盯着心跳）。
	var pending []*client
	for _, c := range h.roomClients("") {
		c.mu.Lock()
		greeted := c.greeted
		c.mu.Unlock()
		if !greeted {
			continue
		}
		// 会话失效（撤销 / 过期 / 用户停用）的连接先断开，不再给它下发心跳。
		if h.sessionRevoked(c) {
			continue
		}
		c.sendEvent(EventHeartbeat, data)
		// 只同步 ack / 慢标记，**不刷 lastHeartbeat**：注册表里的「在线」与「最后活跃」
		// 必须来自对端。服务端每 10s 自己给自己盖一个时间戳，会让拔了网线的屏一直
		// 显示在线、指令发过去还报「命中 N 块屏」。真正的存活求证在 livenessLoop。
		h.syncInstance(c)
		h.checkSlow(c, version)

		if c.snapshotPending.Load() {
			pending = append(pending, c)
		}
	}

	for _, c := range pending {
		h.flushPendingSnapshot(h.ctx, c)
	}
}

// sessionRevoked 复查连接背后的会话是否仍然有效，失效即发起断开；返回 true 表示已断开。
//
// 会话撤销（登出 / 改密 / 管理员重置密码 / 停用用户）会删掉会话行或停用用户，
// 但 /ws 的身份在握手时就固化了：不复查的话，那条长连接会一直以原角色收推送，
// 只有写操作被 HTTP 侧拦住。复查挂在心跳上，最迟一个心跳周期（默认 10s）内断开。
//
// 只有带会话的连接需要复查：guest 大屏连接没有会话，撤销语义不适用。
// 查询失败（库忙 / 系统库被长操作占住）一律按「仍有效」处理并留痕——
// 一次查询抖动不该把正在比赛的大屏整片踢下线。
func (h *Hub) sessionRevoked(c *client) bool {
	checker := h.sessionCheckerValue()
	if checker == nil {
		return false
	}
	c.mu.Lock()
	sessionID, role := c.sessionID, c.role
	c.mu.Unlock()
	if sessionID == "" {
		return false
	}

	ctx, cancel := context.WithTimeout(h.ctx, sessionCheckTimeout)
	defer cancel()
	active, err := checker.SessionActive(ctx, sessionID)
	if err != nil {
		h.log.Warn("复查会话失败，本次保留连接", "session", sessionID, "remote", c.remoteAddr, "err", err)
		return false
	}
	if active {
		return false
	}

	// 管理端连接按「登录失效」关闭：前端对所有 1008 会停止重连并引导重新登录，
	// 这正是被撤销的会话应有的表现。大屏连接改用 1001 让它自动重连——
	// 重连时握手会按会话失效降级成 guest（HTTP 侧本来也是这样），
	// 现场大屏不该因为操作员登出而永久冻结。
	code, reason := websocket.StatusGoingAway, "会话已失效，按游客身份重连"
	if role == RoleAdmin {
		code, reason = websocket.StatusPolicyViolation, "会话已失效，请重新登录"
	}
	h.log.Info("会话已失效，断开实时连接", "session", sessionID, "role", role, "remote", c.remoteAddr)
	c.stop(code, reason)
	return true
}