Files
Scoreboard/server/internal/realtime/hub_test.go
T
AskaEth 782a38f82f feat: 大屏控制可看局域网设备并逐台切页 + 两种新导出(单页成绩 / 单位积分总明细)
## 大屏控制(用户:管理面板只能看到本机显示器,看不到局域网设备;也不能给某块屏单独切页)
- 服务端:`POST /api/display/command` 新增可选 `targets []string` —— 空 = 广播(公告语义不变),
  非空 = 只发给点名的实例并返回命中数;命中 0 个返回 404 +「目标大屏已离线,请刷新在线列表」
  (不落审计、公告也不落库)。`DisplaySender` 签名随之扩展,新增
  `Hub.SendCommandToInstances`(与按房间的 SendCommand 并存)。
- 服务端:WS `hello` 新增可选 `screen`,`realtime.Instance` 新增 `Screen`(不带 omitempty,
  局域网浏览器恒为空串),这样管理端能把「本机显示器」与「在线实例」精确对上。
- 管理端「大屏控制」:写死的「在线看板」占位换成真实列表(5 秒轮询、页面隐藏停表),
  列出实例 id / 在线离线 / 落后 / 屏号或「局域网浏览器」/ 当前页 / 来源 / 心跳;
  每行可单独切页;本机显示器卡片补「当前页」与逐屏切页;离线时控件置灰并说明原因。
  页面下拉改为共用大屏那份页名清单(管理端两处标签随之变为「个人积分奖牌榜」「分项成绩」)。

## 导出(用户:保留多页工作簿的同时,再要一种全平铺单页;积分榜明细也要能导出)
- 新增 `GET /api/export/all-flat.xlsx` → `赛事成绩总表(单页).xlsx`:一个工作簿一张表,
  列 = 大项/小项/组别/名次/号码/单位/运动员-队/成绩/积分/状态/备注;前三列取
  `EventSheet` 的结构化字段(不解析标题),后 8 列就是 `eventColumns` 本身(append 拼出,
  两处不可能漂移)。**既有「全部」(all.xlsx,多 sheet)一字未动。**
- 新增 `GET /api/export/teams.xlsx` → `单位积分总明细.xlsx`:一张带「单位」列的总表 +
  每个有明细行的单位一张分表;**无明细行的单位不出分表**(空表没信息量,且会把假删除的
  隐藏单位名字重新印进导出件);已隐藏但有流水的单位照常出表(它在榜单上)。
- 积分榜页:页头「导出总明细」、单位明细面板「导出该单位明细」;导出页新增「全部(单页)」。

测试:服务端 12 包全绿 / 347 个测试函数;前端 vue-tsc 通过、184 例全过;Console 18 文件。
docs/ 与 devdocs/ 同步(含新增 Q65–Q70 与两轮进度记录)。
2026-09-11 00:44:03 +08:00

911 lines
30 KiB
Go
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
package realtime
import (
"context"
"database/sql"
"encoding/json"
"fmt"
"io"
"log/slog"
"net/http"
"net/http/httptest"
"path/filepath"
"reflect"
"strings"
"testing"
"time"
"github.com/coder/websocket"
_ "modernc.org/sqlite"
"git.yeij.top/AskaEth/Scoreboard/server/internal/auth"
"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()
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) {
handler.ServeHTTP(w, r.WithContext(middleware.WithUser(r.Context(), u)))
}))
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
}
// 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)
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、在线状态、当前页、最后心跳可查。
func TestDisplayInstancesRegistry(t *testing.T) {
env := newTestEnv(t)
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)
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)
}
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", RoleDisplay, 0, false)
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)
}
}