feat: M3-M7 全模块落地 + 集成接线
- M3 认证/权限/审计/限流 + 全部 REST 接口 - M4 管理面板（MIUIx + 成绩录入/积分榜下钻/大屏控制） - M5a 大屏页面（六页 + 明细覆盖层 + 缩放适配） - M6 WebSocket 实时同步中心 - M7 xlsx 导出 + 赛事包 zip + 备份 - 规则包 YAML 导入导出（差异预览） - 字体 vendor 子集化（4 套 x2 字重） - 集成：realtime/export/rules 接入 main+httpapi；backup 子命令 - 修复：中间件包装 ResponseWriter 丢失 Hijacker，导致 WS 握手 501
This commit is contained in:
AskaEth committed
1 parent
d0e3b93e76
commit
bc8b1ee6c0
145 files changed
+25159
-321
No files matched your search
@@ -0,0 +1,107 @@
|
||||
package realtime
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
)
|
||||
|
||||
// currentVersion 返回服务端版本号，即 change_log 的最大 id；空表为 0。
|
||||
func (h *Hub) currentVersion(ctx context.Context) (int64, error) {
|
||||
var v sql.NullInt64
|
||||
if err := h.db.QueryRowContext(ctx, `SELECT MAX(id) FROM change_log`).Scan(&v); err != nil {
|
||||
return 0, fmt.Errorf("读取 change_log 版本号: %w", err)
|
||||
}
|
||||
if !v.Valid {
|
||||
return 0, nil
|
||||
}
|
||||
return v.Int64, nil
|
||||
}
|
||||
|
||||
// currentVersionCached 在数据库不可用时退回内存里的版本，保证心跳与广播不会因一次查询失败而中断。
|
||||
func (h *Hub) currentVersionCached(ctx context.Context) int64 {
|
||||
if h.db == nil {
|
||||
return h.Version()
|
||||
}
|
||||
v, err := h.currentVersion(ctx)
|
||||
if err != nil {
|
||||
h.log.Warn("读取版本号失败，沿用内存版本", "err", err, "version", h.Version())
|
||||
return h.Version()
|
||||
}
|
||||
h.setVersion(v)
|
||||
return v
|
||||
}
|
||||
|
||||
func (h *Hub) countChangesSince(ctx context.Context, from int64) (int, error) {
|
||||
var n int
|
||||
err := h.db.QueryRowContext(ctx, `SELECT COUNT(*) FROM change_log WHERE id > ?`, from).Scan(&n)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("统计增量变更: %w", err)
|
||||
}
|
||||
return n, nil
|
||||
}
|
||||
|
||||
// changesSince 按版本升序取增量；limit <= 0 表示不限量。
|
||||
func (h *Hub) changesSince(ctx context.Context, from int64, limit int) ([]Change, error) {
|
||||
q := `SELECT id, entity_type, entity_id, action, payload_json
|
||||
FROM change_log WHERE id > ? ORDER BY id`
|
||||
args := []any{from}
|
||||
if limit > 0 {
|
||||
q += ` LIMIT ?`
|
||||
args = append(args, limit)
|
||||
}
|
||||
|
||||
rows, err := h.db.QueryContext(ctx, q, args...)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("读取增量变更: %w", err)
|
||||
}
|
||||
defer func() { _ = rows.Close() }()
|
||||
|
||||
out := make([]Change, 0, 16)
|
||||
for rows.Next() {
|
||||
var (
|
||||
ch Change
|
||||
entity string
|
||||
entityID sql.NullInt64
|
||||
action string
|
||||
payload string
|
||||
)
|
||||
if err := rows.Scan(&ch.Version, &entity, &entityID, &action, &payload); err != nil {
|
||||
return nil, fmt.Errorf("解析变更行: %w", err)
|
||||
}
|
||||
ch.Entity = entity
|
||||
ch.ID = entityID.Int64
|
||||
ch.Action = action
|
||||
ch.Data = payloadJSON(payload)
|
||||
out = append(out, ch)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, fmt.Errorf("遍历变更行: %w", err)
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// LoadChanges 供写操作完成后读取「自 sinceVersion 起的变更」，再交给 Broadcast 推送。
|
||||
func (h *Hub) LoadChanges(ctx context.Context, sinceVersion int64, limit int) ([]Change, error) {
|
||||
if h.db == nil {
|
||||
return nil, nil
|
||||
}
|
||||
return h.changesSince(ctx, sinceVersion, limit)
|
||||
}
|
||||
|
||||
// payloadJSON 保证 data 永远是合法 JSON：change_log 允许写入任意文本，
|
||||
// 空串或坏 JSON 一律降级成空对象，避免把坏数据传播给客户端。
|
||||
func payloadJSON(payload string) json.RawMessage {
|
||||
if payload == "" {
|
||||
return json.RawMessage(`{}`)
|
||||
}
|
||||
if !json.Valid([]byte(payload)) {
|
||||
quoted, err := json.Marshal(payload)
|
||||
if err != nil {
|
||||
return json.RawMessage(`{}`)
|
||||
}
|
||||
return quoted
|
||||
}
|
||||
return json.RawMessage(payload)
|
||||
}
|
||||
@@ -0,0 +1,162 @@
|
||||
package realtime
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"github.com/coder/websocket"
|
||||
)
|
||||
|
||||
// client 是一条 WebSocket 连接。读写各一个 goroutine，
|
||||
// 写通过带缓冲的 send 通道串行化，通道满即视为慢客户端。
|
||||
type client struct {
|
||||
hub *Hub
|
||||
conn *websocket.Conn
|
||||
remoteAddr string
|
||||
|
||||
send chan []byte
|
||||
done chan struct{}
|
||||
|
||||
closeOnce sync.Once
|
||||
closeCode websocket.StatusCode
|
||||
closeReason string
|
||||
|
||||
mu sync.Mutex
|
||||
greeted bool
|
||||
role string
|
||||
userID int64
|
||||
instanceID string
|
||||
|
||||
ackVersion atomic.Int64 // 客户端已确认应用到的版本
|
||||
lastSent atomic.Int64 // 服务端已发出的最大版本
|
||||
slow atomic.Bool
|
||||
}
|
||||
|
||||
func newClient(h *Hub, conn *websocket.Conn, remoteAddr string) *client {
|
||||
return &client{
|
||||
hub: h,
|
||||
conn: conn,
|
||||
remoteAddr: remoteAddr,
|
||||
send: make(chan []byte, defaultSendBuffer),
|
||||
done: make(chan struct{}),
|
||||
}
|
||||
}
|
||||
|
||||
// stop 只负责让两条 goroutine 退出；真正的关闭动作由 serve 统一执行，
|
||||
// 避免在持有广播锁时做可能阻塞的关闭握手。
|
||||
func (c *client) stop(code websocket.StatusCode, reason string) {
|
||||
c.closeOnce.Do(func() {
|
||||
c.mu.Lock()
|
||||
c.closeCode = code
|
||||
c.closeReason = reason
|
||||
c.mu.Unlock()
|
||||
close(c.done)
|
||||
})
|
||||
}
|
||||
|
||||
func (c *client) sendEvent(event string, data any) bool {
|
||||
msg, err := encode(event, data)
|
||||
if err != nil {
|
||||
c.hub.log.Error("编码实时消息失败", "event", event, "err", err)
|
||||
return false
|
||||
}
|
||||
return c.sendRaw(msg)
|
||||
}
|
||||
|
||||
func (c *client) sendRaw(msg []byte) bool {
|
||||
select {
|
||||
case <-c.done:
|
||||
return false
|
||||
default:
|
||||
}
|
||||
select {
|
||||
case c.send <- msg:
|
||||
return true
|
||||
default:
|
||||
c.hub.markSlow(c)
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
// readLoop 解析 C→S 消息；首条消息受 helloTimeout 限制，防止空连接占用资源。
|
||||
func (c *client) readLoop(ctx context.Context) {
|
||||
defer c.stop(websocket.StatusNormalClosure, "")
|
||||
|
||||
for {
|
||||
c.mu.Lock()
|
||||
greeted := c.greeted
|
||||
c.mu.Unlock()
|
||||
|
||||
readCtx := ctx
|
||||
var cancel context.CancelFunc
|
||||
if !greeted {
|
||||
readCtx, cancel = context.WithTimeout(ctx, c.hub.helloTimeout())
|
||||
}
|
||||
_, data, err := c.conn.Read(readCtx)
|
||||
if cancel != nil {
|
||||
cancel()
|
||||
}
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
c.handleMessage(ctx, data)
|
||||
}
|
||||
}
|
||||
|
||||
func (c *client) handleMessage(ctx context.Context, data []byte) {
|
||||
var env envelope
|
||||
if err := json.Unmarshal(data, &env); err != nil {
|
||||
c.hub.log.Debug("忽略无法解析的实时消息", "remote", c.remoteAddr, "err", err)
|
||||
return
|
||||
}
|
||||
|
||||
switch env.Event {
|
||||
case EventHello:
|
||||
var p helloPayload
|
||||
if err := json.Unmarshal(env.Data, &p); err != nil {
|
||||
c.hub.log.Debug("hello 载荷不合法", "remote", c.remoteAddr, "err", err)
|
||||
return
|
||||
}
|
||||
c.hub.handleHello(ctx, c, p)
|
||||
|
||||
case EventAck:
|
||||
var p ackPayload
|
||||
if err := json.Unmarshal(env.Data, &p); err != nil {
|
||||
c.hub.log.Debug("ack 载荷不合法", "remote", c.remoteAddr, "err", err)
|
||||
return
|
||||
}
|
||||
if p.Version > c.ackVersion.Load() {
|
||||
c.ackVersion.Store(p.Version)
|
||||
}
|
||||
c.hub.touchInstance(c, time.Now())
|
||||
|
||||
case EventResync:
|
||||
c.hub.sendInitial(ctx, c, nil)
|
||||
|
||||
default:
|
||||
c.hub.log.Debug("忽略未知实时事件", "event", env.Event, "remote", c.remoteAddr)
|
||||
}
|
||||
}
|
||||
|
||||
// writeLoop 是唯一的写者，保证同一连接上的帧顺序。
|
||||
func (c *client) writeLoop(ctx context.Context) {
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-c.done:
|
||||
return
|
||||
case msg := <-c.send:
|
||||
writeCtx, cancel := context.WithTimeout(ctx, c.hub.writeTimeout())
|
||||
err := c.conn.Write(writeCtx, websocket.MessageText, msg)
|
||||
cancel()
|
||||
if err != nil {
|
||||
c.stop(websocket.StatusInternalError, "写入失败")
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,542 @@
|
||||
package realtime
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/coder/websocket"
|
||||
)
|
||||
|
||||
// 默认参数；现场可通过 Set* 调整。
|
||||
const (
|
||||
defaultHeartbeatInterval = 10 * time.Second
|
||||
defaultMaxCatchup = 500
|
||||
defaultSlowLag = 500
|
||||
defaultHelloTimeout = 15 * time.Second
|
||||
defaultWriteTimeout = 10 * time.Second
|
||||
defaultSendBuffer = 64
|
||||
)
|
||||
|
||||
// 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
|
||||
}
|
||||
|
||||
// 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,
|
||||
}
|
||||
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 调整心跳间隔并立即生效；用于现场调优与测试。
|
||||
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()
|
||||
}
|
||||
|
||||
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
|
||||
}
|
||||
|
||||
// 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{
|
||||
// 大屏常以 IP/别名访问且与管理端同端口，Origin 校验交给反向代理与鉴权层。
|
||||
InsecureSkipVerify: true,
|
||||
CompressionMode: websocket.CompressionDisabled,
|
||||
})
|
||||
if err != nil {
|
||||
h.log.Debug("WebSocket 握手失败", "remote", r.RemoteAddr, "err", err)
|
||||
return
|
||||
}
|
||||
|
||||
h.mu.Lock()
|
||||
if h.closed {
|
||||
h.mu.Unlock()
|
||||
_ = conn.CloseNow()
|
||||
return
|
||||
}
|
||||
h.connWG.Add(1)
|
||||
h.mu.Unlock()
|
||||
defer h.connWG.Done()
|
||||
|
||||
h.serve(conn, r.RemoteAddr)
|
||||
})
|
||||
}
|
||||
|
||||
func (h *Hub) serve(conn *websocket.Conn, remoteAddr string) {
|
||||
conn.SetReadLimit(1 << 20)
|
||||
|
||||
c := newClient(h, conn, remoteAddr)
|
||||
h.addClient(c)
|
||||
|
||||
ctx, cancel := context.WithCancel(h.ctx)
|
||||
defer cancel()
|
||||
|
||||
var loops sync.WaitGroup
|
||||
loops.Add(2)
|
||||
go func() { defer loops.Done(); c.readLoop(ctx) }()
|
||||
go func() { defer loops.Done(); c.writeLoop(ctx) }()
|
||||
|
||||
<-c.done
|
||||
cancel()
|
||||
loops.Wait()
|
||||
h.removeClient(c)
|
||||
// 客户端会自动重连并走 catchup，这里直接断开，避免关闭握手阻塞。
|
||||
_ = conn.CloseNow()
|
||||
}
|
||||
|
||||
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
|
||||
}
|
||||
|
||||
func (h *Hub) handleHello(ctx context.Context, c *client, p helloPayload) {
|
||||
if p.Role != RoleDisplay && p.Role != RoleAdmin {
|
||||
h.log.Debug("hello 角色非法", "role", p.Role, "remote", c.remoteAddr)
|
||||
c.stop(websocket.StatusPolicyViolation, "role 非法")
|
||||
return
|
||||
}
|
||||
|
||||
c.mu.Lock()
|
||||
c.role = p.Role
|
||||
c.userID = p.UserID
|
||||
c.mu.Unlock()
|
||||
|
||||
if p.Role == RoleAdmin {
|
||||
rooms := []string{RoomAdmin}
|
||||
if p.UserID > 0 {
|
||||
rooms = append(rooms, AdminRoom(p.UserID))
|
||||
}
|
||||
h.join(c, rooms...)
|
||||
} else {
|
||||
id := p.InstanceID
|
||||
if id == "" {
|
||||
id = p.DisplayKey
|
||||
}
|
||||
h.registerDisplay(c, id, p.Page)
|
||||
h.join(c, RoomDisplay)
|
||||
}
|
||||
|
||||
h.sendInitial(ctx, c, p.LastVersion)
|
||||
}
|
||||
|
||||
// sendInitial 决定首包是 snapshot 还是 catchup：
|
||||
// 没有 lastVersion、版本超前、或增量条数超阈值，一律回落全量快照。
|
||||
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 {
|
||||
h.log.Error("读取版本号失败，无法下发首包", "err", err)
|
||||
c.stop(websocket.StatusInternalError, "版本号不可用")
|
||||
return
|
||||
}
|
||||
h.setVersion(cur)
|
||||
|
||||
if lastVersion == nil || *lastVersion <= 0 || *lastVersion > cur {
|
||||
h.sendSnapshotLocked(ctx, c)
|
||||
return
|
||||
}
|
||||
|
||||
n, err := h.countChangesSince(ctx, *lastVersion)
|
||||
if err != nil {
|
||||
h.log.Warn("统计增量失败，回落快照", "err", err)
|
||||
h.sendSnapshotLocked(ctx, c)
|
||||
return
|
||||
}
|
||||
if n > h.maxCatchupValue() {
|
||||
h.sendSnapshotLocked(ctx, c)
|
||||
return
|
||||
}
|
||||
|
||||
changes, err := h.changesSince(ctx, *lastVersion, 0)
|
||||
if err != nil {
|
||||
h.log.Warn("读取增量失败，回落快照", "err", err)
|
||||
h.sendSnapshotLocked(ctx, c)
|
||||
return
|
||||
}
|
||||
if !c.sendEvent(EventCatchup, catchupPayload{FromVersion: *lastVersion, Version: cur, Changes: changes}) {
|
||||
return
|
||||
}
|
||||
c.lastSent.Store(cur)
|
||||
c.ackVersion.Store(cur)
|
||||
}
|
||||
|
||||
// sendSnapshotLocked 必须在持有 sendMu 时调用：快照读取与「已发送版本」的推进
|
||||
// 必须原子完成，否则并发的 change 会插到快照前面。
|
||||
func (h *Hub) sendSnapshotLocked(ctx context.Context, c *client) {
|
||||
snap, err := h.Snapshot(ctx)
|
||||
if err != nil {
|
||||
h.log.Error("构建快照失败", "err", err)
|
||||
c.stop(websocket.StatusInternalError, "快照不可用")
|
||||
return
|
||||
}
|
||||
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)）。
|
||||
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()
|
||||
filtered := make([]Change, 0, len(changes))
|
||||
for _, ch := range changes {
|
||||
if ch.Version > 0 && ch.Version <= last {
|
||||
continue
|
||||
}
|
||||
filtered = append(filtered, ch)
|
||||
}
|
||||
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))
|
||||
}
|
||||
|
||||
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)
|
||||
}
|
||||
}
|
||||
|
||||
// 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(websocket.StatusPolicyViolation, "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)}
|
||||
|
||||
for _, c := range h.roomClients("") {
|
||||
c.mu.Lock()
|
||||
greeted := c.greeted
|
||||
c.mu.Unlock()
|
||||
if !greeted {
|
||||
continue
|
||||
}
|
||||
c.sendEvent(EventHeartbeat, data)
|
||||
h.touchInstance(c, now)
|
||||
h.checkSlow(c, version)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,502 @@
|
||||
package realtime
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"log/slog"
|
||||
"net/http/httptest"
|
||||
"path/filepath"
|
||||
"reflect"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/coder/websocket"
|
||||
_ "modernc.org/sqlite"
|
||||
|
||||
"git.yeij.top/AskaEth/Scoreboard/server/internal/store"
|
||||
)
|
||||
|
||||
// ---------- 测试脚手架 ----------
|
||||
|
||||
type testEnv struct {
|
||||
hub *Hub
|
||||
db *sql.DB
|
||||
srv *httptest.Server
|
||||
}
|
||||
|
||||
func newTestEnv(t *testing.T) *testEnv {
|
||||
t.Helper()
|
||||
|
||||
db := newTestDB(t)
|
||||
h := NewHub(db, slog.New(slog.NewTextHandler(io.Discard, &slog.HandlerOptions{Level: slog.LevelError})))
|
||||
srv := httptest.NewServer(h.Handler())
|
||||
|
||||
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
|
||||
}
|
||||
|
||||
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}
|
||||
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
|
||||
}
|
||||
|
||||
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> 只发给对应管理端。
|
||||
func TestAdminRoomTargeting(t *testing.T) {
|
||||
env := newTestEnv(t)
|
||||
display := env.dial(t, &helloPayload{Role: RoleDisplay, InstanceID: "d1"})
|
||||
display.recvEvent(EventSnapshot)
|
||||
admin := env.dial(t, &helloPayload{Role: RoleAdmin, UserID: 7})
|
||||
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)
|
||||
}
|
||||
}
|
||||
|
||||
// 慢客户端：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
|
||||
})
|
||||
}
|
||||
@@ -0,0 +1,158 @@
|
||||
package realtime
|
||||
|
||||
import (
|
||||
"crypto/rand"
|
||||
"encoding/hex"
|
||||
"sort"
|
||||
"strconv"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Instance 是大屏实例的对外快照，供管理端查询「谁在线、在哪一页」。
|
||||
type Instance struct {
|
||||
ID string `json:"id"`
|
||||
Online bool `json:"online"`
|
||||
Page string `json:"page,omitempty"`
|
||||
LastHeartbeat time.Time `json:"lastHeartbeat"`
|
||||
ConnectedAt time.Time `json:"connectedAt"`
|
||||
RemoteAddr string `json:"remoteAddr,omitempty"`
|
||||
AckVersion int64 `json:"ackVersion"`
|
||||
Slow bool `json:"slow"`
|
||||
}
|
||||
|
||||
// instanceState 是注册表的内部记录；同一 instanceId 的多条连接合并成一个实例。
|
||||
type instanceState struct {
|
||||
id string
|
||||
page string
|
||||
connectedAt time.Time
|
||||
lastHeartbeat time.Time
|
||||
remoteAddr string
|
||||
ackVersion int64
|
||||
slow bool
|
||||
conns int
|
||||
}
|
||||
|
||||
// registerDisplay 把连接挂到大屏实例上；未带 instanceId 时由服务端生成，
|
||||
// 保证「刷新页面 = 同一实例」之外的连接也能被登记和查询。
|
||||
func (h *Hub) registerDisplay(c *client, id, page string) string {
|
||||
if id == "" {
|
||||
id = "display-" + randomID()
|
||||
}
|
||||
now := time.Now()
|
||||
|
||||
c.mu.Lock()
|
||||
prev := c.instanceID
|
||||
c.instanceID = id
|
||||
c.mu.Unlock()
|
||||
|
||||
// 同一连接重复 hello 或换了 instanceId 时，先归还旧实例的引用，避免计数虚高。
|
||||
if prev != "" && prev != id {
|
||||
h.releaseDisplayID(prev, c)
|
||||
}
|
||||
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
st := h.instances[id]
|
||||
if st == nil {
|
||||
st = &instanceState{id: id, connectedAt: now}
|
||||
h.instances[id] = st
|
||||
}
|
||||
if prev != id {
|
||||
st.conns++
|
||||
}
|
||||
if page != "" {
|
||||
st.page = page
|
||||
}
|
||||
st.lastHeartbeat = now
|
||||
st.remoteAddr = c.remoteAddr
|
||||
return id
|
||||
}
|
||||
|
||||
// releaseDisplay 断开时只减少引用计数：实例保留在注册表里并标记离线，方便事后排查。
|
||||
func (h *Hub) releaseDisplay(c *client) {
|
||||
c.mu.Lock()
|
||||
id := c.instanceID
|
||||
c.mu.Unlock()
|
||||
if id == "" {
|
||||
return
|
||||
}
|
||||
h.releaseDisplayID(id, c)
|
||||
}
|
||||
|
||||
func (h *Hub) releaseDisplayID(id string, c *client) {
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
st := h.instances[id]
|
||||
if st == nil {
|
||||
return
|
||||
}
|
||||
if st.conns > 0 {
|
||||
st.conns--
|
||||
}
|
||||
st.lastHeartbeat = time.Now()
|
||||
st.ackVersion = c.ackVersion.Load()
|
||||
st.slow = c.slow.Load()
|
||||
}
|
||||
|
||||
// touchInstance 在收到客户端消息或下发心跳时刷新「最后心跳」。
|
||||
func (h *Hub) touchInstance(c *client, now time.Time) {
|
||||
c.mu.Lock()
|
||||
id := c.instanceID
|
||||
c.mu.Unlock()
|
||||
if id == "" {
|
||||
return
|
||||
}
|
||||
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
if st := h.instances[id]; st != nil {
|
||||
st.lastHeartbeat = now
|
||||
st.ackVersion = c.ackVersion.Load()
|
||||
st.slow = c.slow.Load()
|
||||
}
|
||||
}
|
||||
|
||||
// setInstancePage 记录服务端刚下发的目标页，让注册表里的「当前页」跟得上 goto 指令。
|
||||
func (h *Hub) setInstancePage(page string) {
|
||||
if page == "" {
|
||||
return
|
||||
}
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
for _, st := range h.instances {
|
||||
if st.conns > 0 {
|
||||
st.page = page
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Instances 返回注册表快照，按 instanceId 排序，便于管理端稳定展示。
|
||||
func (h *Hub) Instances() []Instance {
|
||||
h.mu.RLock()
|
||||
defer h.mu.RUnlock()
|
||||
|
||||
out := make([]Instance, 0, len(h.instances))
|
||||
for _, st := range h.instances {
|
||||
out = append(out, Instance{
|
||||
ID: st.id,
|
||||
Online: st.conns > 0,
|
||||
Page: st.page,
|
||||
LastHeartbeat: st.lastHeartbeat,
|
||||
ConnectedAt: st.connectedAt,
|
||||
RemoteAddr: st.remoteAddr,
|
||||
AckVersion: st.ackVersion,
|
||||
Slow: st.slow,
|
||||
})
|
||||
}
|
||||
sort.Slice(out, func(i, j int) bool { return out[i].ID < out[j].ID })
|
||||
return out
|
||||
}
|
||||
|
||||
func randomID() string {
|
||||
var b [8]byte
|
||||
if _, err := rand.Read(b[:]); err != nil {
|
||||
// 随机源不可用时退回时间戳，实例 id 只需唯一，不承担安全职责。
|
||||
return strconv.FormatInt(time.Now().UnixNano(), 16)
|
||||
}
|
||||
return hex.EncodeToString(b[:])
|
||||
}
|
||||
@@ -0,0 +1,171 @@
|
||||
// Package realtime 实现 /ws 实时同步中心。
|
||||
//
|
||||
// 协议以 devdocs/05-接口与实时协议.md 第 1–5 节为唯一规范：
|
||||
// 版本号取自 change_log.id，客户端按 version 去重，增量补不上就回落全量快照。
|
||||
package realtime
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"time"
|
||||
)
|
||||
|
||||
// 事件名与协议第 3 节的表格一一对应。
|
||||
const (
|
||||
EventHello = "hello"
|
||||
EventSnapshot = "snapshot"
|
||||
EventCatchup = "catchup"
|
||||
EventChange = "change"
|
||||
EventResync = "resync"
|
||||
EventAck = "ack"
|
||||
EventHeartbeat = "heartbeat"
|
||||
EventDisplayCommand = "display:command"
|
||||
)
|
||||
|
||||
// 房间名。admin:<userId> 由 AdminRoom 生成，用于定向通知。
|
||||
const (
|
||||
RoomDisplay = "display"
|
||||
RoomAdmin = "admin"
|
||||
)
|
||||
|
||||
// AdminRoom 返回某个管理端用户的定向房间名。
|
||||
func AdminRoom(userID int64) string { return fmt.Sprintf("admin:%d", userID) }
|
||||
|
||||
// 连接角色。
|
||||
const (
|
||||
RoleDisplay = "display"
|
||||
RoleAdmin = "admin"
|
||||
)
|
||||
|
||||
// change.action 取值。
|
||||
const (
|
||||
ActionUpsert = "upsert"
|
||||
ActionVoid = "void"
|
||||
ActionReplace = "replace"
|
||||
)
|
||||
|
||||
// display:command 支持的 action。
|
||||
const (
|
||||
CmdGoto = "goto"
|
||||
CmdLock = "lock"
|
||||
CmdUnlock = "unlock"
|
||||
CmdReload = "reload"
|
||||
CmdAnnounce = "announce"
|
||||
CmdShowDetail = "show-detail"
|
||||
CmdHideDetail = "hide-detail"
|
||||
)
|
||||
|
||||
var displayActions = map[string]bool{
|
||||
CmdGoto: true,
|
||||
CmdLock: true,
|
||||
CmdUnlock: true,
|
||||
CmdReload: true,
|
||||
CmdAnnounce: true,
|
||||
CmdShowDetail: true,
|
||||
CmdHideDetail: true,
|
||||
}
|
||||
|
||||
// ValidDisplayAction 判断 display:command 的 action 是否在协议清单内。
|
||||
func ValidDisplayAction(action string) bool { return displayActions[action] }
|
||||
|
||||
// envelope 是所有消息的统一外壳：{ "event": "...", "data": { ... } }。
|
||||
type envelope struct {
|
||||
Event string `json:"event"`
|
||||
Data json.RawMessage `json:"data,omitempty"`
|
||||
}
|
||||
|
||||
// encode 把事件名与载荷打包成一条文本帧。
|
||||
func encode(event string, data any) ([]byte, error) {
|
||||
raw, err := json.Marshal(data)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("编码 %s 载荷: %w", event, err)
|
||||
}
|
||||
return json.Marshal(envelope{Event: event, Data: raw})
|
||||
}
|
||||
|
||||
// helloPayload 是 C→S 的 hello 载荷。
|
||||
// instanceId / page 是协议字段之外的可选扩展：注册表要记「谁在线、在哪一页」，
|
||||
// 老客户端不带这两个字段也能正常工作。
|
||||
type helloPayload struct {
|
||||
Role string `json:"role"`
|
||||
LastVersion *int64 `json:"lastVersion,omitempty"`
|
||||
DisplayKey string `json:"displayKey,omitempty"`
|
||||
InstanceID string `json:"instanceId,omitempty"`
|
||||
UserID int64 `json:"userId,omitempty"`
|
||||
Page string `json:"page,omitempty"`
|
||||
}
|
||||
|
||||
// ackPayload 是 C→S 的 ack 载荷；服务端据此识别慢客户端。
|
||||
type ackPayload struct {
|
||||
Version int64 `json:"version"`
|
||||
}
|
||||
|
||||
// Change 是变更的最小单元；Version 只用于服务端排序与去重，不上线路。
|
||||
type Change struct {
|
||||
Entity string `json:"entity"`
|
||||
ID int64 `json:"id,omitempty"`
|
||||
Action string `json:"action"`
|
||||
Data json.RawMessage `json:"data,omitempty"`
|
||||
Version int64 `json:"-"`
|
||||
}
|
||||
|
||||
// changePayload 是 change 事件的载荷。
|
||||
type changePayload struct {
|
||||
Version int64 `json:"version"`
|
||||
Changes []Change `json:"changes"`
|
||||
}
|
||||
|
||||
// catchupPayload 是 catchup 事件的载荷。
|
||||
type catchupPayload struct {
|
||||
FromVersion int64 `json:"fromVersion"`
|
||||
Version int64 `json:"version"`
|
||||
Changes []Change `json:"changes"`
|
||||
}
|
||||
|
||||
// heartbeatPayload 是 heartbeat 事件的载荷。
|
||||
type heartbeatPayload struct {
|
||||
Version int64 `json:"version"`
|
||||
ServerTime string `json:"serverTime"`
|
||||
}
|
||||
|
||||
// DisplayCommand 是下发给大屏的 display:command 载荷。
|
||||
type DisplayCommand struct {
|
||||
Action string `json:"action"`
|
||||
Payload json.RawMessage `json:"payload,omitempty"`
|
||||
}
|
||||
|
||||
// NewDisplayCommand 构造一条大屏指令；action 不在协议清单内时返回错误。
|
||||
func NewDisplayCommand(action string, payload any) (DisplayCommand, error) {
|
||||
if !ValidDisplayAction(action) {
|
||||
return DisplayCommand{}, fmt.Errorf("未知的大屏指令 %q", action)
|
||||
}
|
||||
cmd := DisplayCommand{Action: action}
|
||||
if payload != nil {
|
||||
raw, err := json.Marshal(payload)
|
||||
if err != nil {
|
||||
return DisplayCommand{}, fmt.Errorf("编码大屏指令 %s 的 payload: %w", action, err)
|
||||
}
|
||||
cmd.Payload = raw
|
||||
}
|
||||
return cmd, nil
|
||||
}
|
||||
|
||||
// pageFromPayload 从 goto 指令的 payload 里取目标页名，兼容字符串与对象两种写法。
|
||||
func pageFromPayload(payload json.RawMessage) string {
|
||||
if len(payload) == 0 {
|
||||
return ""
|
||||
}
|
||||
var s string
|
||||
if err := json.Unmarshal(payload, &s); err == nil {
|
||||
return s
|
||||
}
|
||||
var obj struct {
|
||||
Page string `json:"page"`
|
||||
}
|
||||
if err := json.Unmarshal(payload, &obj); err == nil {
|
||||
return obj.Page
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
func nowISO(t time.Time) string { return t.Format(time.RFC3339Nano) }
|
||||
@@ -0,0 +1,348 @@
|
||||
package realtime
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
|
||||
"git.yeij.top/AskaEth/Scoreboard/server/internal/domain"
|
||||
)
|
||||
|
||||
// latestResultsLimit 控制快照里「最新成绩」的条数，避免大屏首包过大。
|
||||
const latestResultsLimit = 20
|
||||
|
||||
// Snapshot 是 snapshot 事件的 data，也是客户端冷启动的完整状态。
|
||||
// 字段名与协议第 3 节一致。
|
||||
type Snapshot struct {
|
||||
Version int64 `json:"version"`
|
||||
Settings map[string]json.RawMessage `json:"settings"`
|
||||
Teams []domain.Team `json:"teams"`
|
||||
Events []domain.Event `json:"events"`
|
||||
Standings []domain.StandingsRow `json:"standings"`
|
||||
Records []Record `json:"records"`
|
||||
LatestResults []LatestResult `json:"latestResults"`
|
||||
}
|
||||
|
||||
// Record 是纪录榜的一行。
|
||||
type Record struct {
|
||||
ID int64 `json:"id"`
|
||||
EventID int64 `json:"eventId"`
|
||||
Level string `json:"level"`
|
||||
ResultValue *float64 `json:"resultValue,omitempty"`
|
||||
ResultText string `json:"resultText,omitempty"`
|
||||
HolderName string `json:"holderName,omitempty"`
|
||||
HolderTeamID int64 `json:"holderTeamId,omitempty"`
|
||||
AchievedAt string `json:"achievedAt,omitempty"`
|
||||
IsCurrent bool `json:"isCurrent"`
|
||||
Source string `json:"source"`
|
||||
}
|
||||
|
||||
// LatestResult 是已发布成绩的展示行，供大屏滚动与详情页使用。
|
||||
type LatestResult struct {
|
||||
ID int64 `json:"id"`
|
||||
EventID int64 `json:"eventId"`
|
||||
EntryID int64 `json:"entryId"`
|
||||
TeamID int64 `json:"teamId"`
|
||||
AthleteID int64 `json:"athleteId,omitempty"`
|
||||
Rank int `json:"rank,omitempty"`
|
||||
ResultText string `json:"resultText,omitempty"`
|
||||
ResultValue *float64 `json:"resultValue,omitempty"`
|
||||
Unit string `json:"unit,omitempty"`
|
||||
IsTie bool `json:"isTie,omitempty"`
|
||||
Status string `json:"status"`
|
||||
State string `json:"state"`
|
||||
UpdatedAt string `json:"updatedAt,omitempty"`
|
||||
}
|
||||
|
||||
// Snapshot 读取一份一致的完整状态。
|
||||
//
|
||||
// 全部查询放在同一个只读事务里，且版本号取自同一事务，
|
||||
// 否则快照里的数据与版本号会错位，重连补齐时就会漏变更。
|
||||
func (h *Hub) Snapshot(ctx context.Context) (*Snapshot, error) {
|
||||
if h.db == nil {
|
||||
return nil, fmt.Errorf("实时中心未绑定赛事库")
|
||||
}
|
||||
|
||||
tx, err := h.db.BeginTx(ctx, nil)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("开启快照事务: %w", err)
|
||||
}
|
||||
defer func() { _ = tx.Rollback() }()
|
||||
|
||||
snap := &Snapshot{
|
||||
Settings: map[string]json.RawMessage{},
|
||||
Teams: []domain.Team{},
|
||||
Events: []domain.Event{},
|
||||
Standings: []domain.StandingsRow{},
|
||||
Records: []Record{},
|
||||
LatestResults: []LatestResult{},
|
||||
}
|
||||
|
||||
if err := tx.QueryRowContext(ctx, `SELECT COALESCE(MAX(id), 0) FROM change_log`).Scan(&snap.Version); err != nil {
|
||||
return nil, fmt.Errorf("读取快照版本号: %w", err)
|
||||
}
|
||||
if err := loadSettings(ctx, tx, snap); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := loadTeams(ctx, tx, snap); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := loadEvents(ctx, tx, snap); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := loadStandings(ctx, tx, snap); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := loadRecords(ctx, tx, snap); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := loadLatestResults(ctx, tx, snap); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if err := tx.Commit(); err != nil {
|
||||
return nil, fmt.Errorf("提交快照事务: %w", err)
|
||||
}
|
||||
return snap, nil
|
||||
}
|
||||
|
||||
func loadSettings(ctx context.Context, tx *sql.Tx, snap *Snapshot) error {
|
||||
rows, err := tx.QueryContext(ctx, `SELECT key, value FROM settings ORDER BY key`)
|
||||
if err != nil {
|
||||
return fmt.Errorf("读取赛事配置: %w", err)
|
||||
}
|
||||
defer func() { _ = rows.Close() }()
|
||||
|
||||
for rows.Next() {
|
||||
var key, value string
|
||||
if err := rows.Scan(&key, &value); err != nil {
|
||||
return fmt.Errorf("解析赛事配置: %w", err)
|
||||
}
|
||||
snap.Settings[key] = rawJSONOrString(value)
|
||||
}
|
||||
return rows.Err()
|
||||
}
|
||||
|
||||
func loadTeams(ctx context.Context, tx *sql.Tx, snap *Snapshot) error {
|
||||
rows, err := tx.QueryContext(ctx, `
|
||||
SELECT id, code, name, COALESCE(short_name, ''), COALESCE(logo_path, ''),
|
||||
COALESCE(color, ''), sort_order, active
|
||||
FROM teams WHERE active = 1
|
||||
ORDER BY sort_order, code`)
|
||||
if err != nil {
|
||||
return fmt.Errorf("读取单位: %w", err)
|
||||
}
|
||||
defer func() { _ = rows.Close() }()
|
||||
|
||||
for rows.Next() {
|
||||
var (
|
||||
t domain.Team
|
||||
active int
|
||||
)
|
||||
if err := rows.Scan(&t.ID, &t.Code, &t.Name, &t.ShortName, &t.LogoPath, &t.Color, &t.SortOrder, &active); err != nil {
|
||||
return fmt.Errorf("解析单位: %w", err)
|
||||
}
|
||||
t.Active = active != 0
|
||||
snap.Teams = append(snap.Teams, t)
|
||||
}
|
||||
return rows.Err()
|
||||
}
|
||||
|
||||
func loadEvents(ctx context.Context, tx *sql.Tx, snap *Snapshot) error {
|
||||
rows, err := tx.QueryContext(ctx, `
|
||||
SELECT id, sport_id, code, name, event_type, result_mode, result_type, compare,
|
||||
record_eligible, scored, multiplier, status,
|
||||
COALESCE(scheduled_at, ''), COALESCE(venue, ''), sort_order, COALESCE(remark, '')
|
||||
FROM events
|
||||
ORDER BY sort_order, id`)
|
||||
if err != nil {
|
||||
return fmt.Errorf("读取小项: %w", err)
|
||||
}
|
||||
defer func() { _ = rows.Close() }()
|
||||
|
||||
for rows.Next() {
|
||||
var (
|
||||
e domain.Event
|
||||
eventType string
|
||||
resultMode string
|
||||
resultType string
|
||||
compare string
|
||||
status string
|
||||
recordEligible int
|
||||
scored int
|
||||
)
|
||||
if err := rows.Scan(
|
||||
&e.ID, &e.SportID, &e.Code, &e.Name, &eventType, &resultMode, &resultType, &compare,
|
||||
&recordEligible, &scored, &e.Multiplier, &status,
|
||||
&e.ScheduledAt, &e.Venue, &e.SortOrder, &e.Remark,
|
||||
); err != nil {
|
||||
return fmt.Errorf("解析小项: %w", err)
|
||||
}
|
||||
e.EventType = domain.EventType(eventType)
|
||||
e.ResultMode = domain.ResultMode(resultMode)
|
||||
e.ResultType = domain.ResultType(resultType)
|
||||
e.Compare = domain.CompareMode(compare)
|
||||
e.Status = domain.EventStatus(status)
|
||||
e.RecordEligible = recordEligible != 0
|
||||
e.Scored = scored != 0
|
||||
snap.Events = append(snap.Events, e)
|
||||
}
|
||||
return rows.Err()
|
||||
}
|
||||
|
||||
// loadStandings 复用 domain 的排名口径：积分来自流水，奖牌来自已发布名次，
|
||||
// 保证大屏看到的榜单与 REST /api/standings 一致。
|
||||
func loadStandings(ctx context.Context, tx *sql.Tx, snap *Snapshot) error {
|
||||
ledger, err := loadLedger(ctx, tx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
ranked, err := loadRanked(ctx, tx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
snap.Standings = domain.BuildTeamStandings(domain.StandingsInput{
|
||||
Teams: snap.Teams,
|
||||
Ledger: ledger,
|
||||
Ranked: ranked,
|
||||
})
|
||||
return nil
|
||||
}
|
||||
|
||||
func loadLedger(ctx context.Context, tx *sql.Tx) ([]domain.LedgerEntry, error) {
|
||||
rows, err := tx.QueryContext(ctx, `
|
||||
SELECT id, team_id, COALESCE(athlete_id, 0), COALESCE(event_id, 0),
|
||||
COALESCE(entry_id, 0), COALESCE(result_id, 0), source_type, source_id, points
|
||||
FROM score_ledger WHERE voided = 0 ORDER BY id`)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("读取积分流水: %w", err)
|
||||
}
|
||||
defer func() { _ = rows.Close() }()
|
||||
|
||||
var out []domain.LedgerEntry
|
||||
for rows.Next() {
|
||||
var (
|
||||
e domain.LedgerEntry
|
||||
sourceType string
|
||||
)
|
||||
if err := rows.Scan(&e.ID, &e.TeamID, &e.AthleteID, &e.EventID, &e.EntryID, &e.ResultID, &sourceType, &e.SourceID, &e.Points); err != nil {
|
||||
return nil, fmt.Errorf("解析积分流水: %w", err)
|
||||
}
|
||||
e.SourceType = domain.SourceType(sourceType)
|
||||
out = append(out, e)
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
// loadRanked 只取已发布的前三名，用于统计奖牌；名次明细的完整计算仍由成绩模块负责。
|
||||
func loadRanked(ctx context.Context, tx *sql.Tx) ([]domain.RankedResult, error) {
|
||||
rows, err := tx.QueryContext(ctx, `
|
||||
SELECT e.team_id, COALESCE(e.athlete_id, 0), r.rank
|
||||
FROM results r JOIN entries e ON e.id = r.entry_id
|
||||
WHERE r.state = 'published' AND r.status = 'ok' AND r.rank BETWEEN 1 AND 3
|
||||
ORDER BY r.id`)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("读取奖牌名次: %w", err)
|
||||
}
|
||||
defer func() { _ = rows.Close() }()
|
||||
|
||||
var out []domain.RankedResult
|
||||
for rows.Next() {
|
||||
var (
|
||||
teamID, athleteID int64
|
||||
rank int
|
||||
)
|
||||
if err := rows.Scan(&teamID, &athleteID, &rank); err != nil {
|
||||
return nil, fmt.Errorf("解析奖牌名次: %w", err)
|
||||
}
|
||||
out = append(out, domain.RankedResult{
|
||||
TeamID: teamID,
|
||||
AthleteID: athleteID,
|
||||
Rank: rank,
|
||||
Scope: domain.ScopeTeam,
|
||||
Shares: []domain.TeamShare{{TeamID: teamID, Weight: 1}},
|
||||
})
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
func loadRecords(ctx context.Context, tx *sql.Tx, snap *Snapshot) error {
|
||||
rows, err := tx.QueryContext(ctx, `
|
||||
SELECT id, event_id, level, result_value, COALESCE(result_text, ''),
|
||||
COALESCE(holder_name, ''), COALESCE(holder_team_id, 0),
|
||||
COALESCE(achieved_at, ''), is_current, source
|
||||
FROM records WHERE is_current = 1
|
||||
ORDER BY event_id, level, id`)
|
||||
if err != nil {
|
||||
return fmt.Errorf("读取纪录榜: %w", err)
|
||||
}
|
||||
defer func() { _ = rows.Close() }()
|
||||
|
||||
for rows.Next() {
|
||||
var (
|
||||
r Record
|
||||
value sql.NullFloat64
|
||||
isCurrent int
|
||||
)
|
||||
if err := rows.Scan(&r.ID, &r.EventID, &r.Level, &value, &r.ResultText,
|
||||
&r.HolderName, &r.HolderTeamID, &r.AchievedAt, &isCurrent, &r.Source); err != nil {
|
||||
return fmt.Errorf("解析纪录: %w", err)
|
||||
}
|
||||
if value.Valid {
|
||||
v := value.Float64
|
||||
r.ResultValue = &v
|
||||
}
|
||||
r.IsCurrent = isCurrent != 0
|
||||
snap.Records = append(snap.Records, r)
|
||||
}
|
||||
return rows.Err()
|
||||
}
|
||||
|
||||
func loadLatestResults(ctx context.Context, tx *sql.Tx, snap *Snapshot) error {
|
||||
rows, err := tx.QueryContext(ctx, `
|
||||
SELECT r.id, r.event_id, r.entry_id, e.team_id, COALESCE(e.athlete_id, 0),
|
||||
COALESCE(r.rank, 0), COALESCE(r.result_text, ''), r.result_value,
|
||||
COALESCE(r.unit, ''), r.is_tie, r.status, r.state,
|
||||
COALESCE(r.verified_at, r.recorded_at, r.updated_at, r.created_at, '')
|
||||
FROM results r JOIN entries e ON e.id = r.entry_id
|
||||
WHERE r.state = 'published'
|
||||
ORDER BY COALESCE(r.verified_at, r.recorded_at, r.updated_at) DESC, r.id DESC
|
||||
LIMIT ?`, latestResultsLimit)
|
||||
if err != nil {
|
||||
return fmt.Errorf("读取最新成绩: %w", err)
|
||||
}
|
||||
defer func() { _ = rows.Close() }()
|
||||
|
||||
for rows.Next() {
|
||||
var (
|
||||
lr LatestResult
|
||||
value sql.NullFloat64
|
||||
isTie int
|
||||
)
|
||||
if err := rows.Scan(&lr.ID, &lr.EventID, &lr.EntryID, &lr.TeamID, &lr.AthleteID,
|
||||
&lr.Rank, &lr.ResultText, &value, &lr.Unit, &isTie, &lr.Status, &lr.State, &lr.UpdatedAt); err != nil {
|
||||
return fmt.Errorf("解析最新成绩: %w", err)
|
||||
}
|
||||
if value.Valid {
|
||||
v := value.Float64
|
||||
lr.ResultValue = &v
|
||||
}
|
||||
lr.IsTie = isTie != 0
|
||||
snap.LatestResults = append(snap.LatestResults, lr)
|
||||
}
|
||||
return rows.Err()
|
||||
}
|
||||
|
||||
// rawJSONOrString 让 settings 的值保持原样：本来就是 JSON 的直接透传，
|
||||
// 纯文本则编码成 JSON 字符串，客户端统一按 JSON 解析。
|
||||
func rawJSONOrString(value string) json.RawMessage {
|
||||
if json.Valid([]byte(value)) {
|
||||
return json.RawMessage(value)
|
||||
}
|
||||
quoted, err := json.Marshal(value)
|
||||
if err != nil {
|
||||
return json.RawMessage(`null`)
|
||||
}
|
||||
return quoted
|
||||
}
|
||||
@@ -0,0 +1,103 @@
|
||||
package realtime
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"encoding/json"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"git.yeij.top/AskaEth/Scoreboard/server/internal/domain"
|
||||
)
|
||||
|
||||
// 快照必须与数据库同源：版本号取自 change_log，各列表取真实表，
|
||||
// 且空库时序列化为 [] 而不是 null，避免前端判空逻辑踩坑。
|
||||
func TestSnapshotContents(t *testing.T) {
|
||||
env := newTestEnv(t)
|
||||
db := env.db
|
||||
|
||||
mustExec(t, db, `UPDATE settings SET value = ? WHERE key = 'event.name'`, `"测试运动会"`)
|
||||
mustExec(t, db, `INSERT INTO settings(key, value) VALUES ('display.theme', 'dark')`)
|
||||
mustExec(t, db, `INSERT INTO teams(code, name, short_name, sort_order, active) VALUES ('T1', '第一单位', '一队', 1, 1)`)
|
||||
mustExec(t, db, `INSERT INTO sports(code, name) VALUES ('S1', '田径')`)
|
||||
mustExec(t, db, `INSERT INTO events(sport_id, code, name, event_type, result_mode, result_type, compare, status)
|
||||
VALUES (1, 'E1', '100米', 'individual', 'auto', 'time', 'asc', 'published')`)
|
||||
mustExec(t, db, `INSERT INTO event_categories(event_id, name) VALUES (1, '男子')`)
|
||||
mustExec(t, db, `INSERT INTO athletes(name, gender, team_id) VALUES ('张三', '男', 1)`)
|
||||
mustExec(t, db, `INSERT INTO entries(event_id, category_id, team_id, athlete_id, bib) VALUES (1, 1, 1, 1, '101')`)
|
||||
mustExec(t, db, `INSERT INTO results(event_id, entry_id, rank, result_text, result_value, unit, state, status)
|
||||
VALUES (1, 1, 1, '11.58', 11.58, 's', 'published', 'ok')`)
|
||||
mustExec(t, db, `INSERT INTO score_ledger(team_id, athlete_id, event_id, entry_id, result_id, source_type, source_id, points)
|
||||
VALUES (1, 1, 1, 1, 1, 'result', 1, 9)`)
|
||||
mustExec(t, db, `INSERT INTO records(event_id, level, result_value, result_text, holder_name, holder_team_id, achieved_at, is_current, source)
|
||||
VALUES (1, 'meet', 11.2, '11.20', '李四', 1, '2025-09-01', 1, 'history')`)
|
||||
version := insertChange(t, db, "result", 1, ActionUpsert, `{"rank":1}`)
|
||||
|
||||
snap, err := env.hub.Snapshot(context.Background())
|
||||
if err != nil {
|
||||
t.Fatalf("构建快照: %v", err)
|
||||
}
|
||||
if snap.Version != version {
|
||||
t.Fatalf("快照版本应为 %d，得到 %d", version, snap.Version)
|
||||
}
|
||||
if got := string(snap.Settings["event.name"]); got != `"测试运动会"` {
|
||||
t.Fatalf("settings JSON 值应透传，得到 %s", got)
|
||||
}
|
||||
if got := string(snap.Settings["display.theme"]); got != `"dark"` {
|
||||
t.Fatalf("settings 纯文本应编码为 JSON 字符串，得到 %s", got)
|
||||
}
|
||||
|
||||
if len(snap.Teams) != 1 || snap.Teams[0].Code != "T1" || !snap.Teams[0].Active {
|
||||
t.Fatalf("单位不符: %+v", snap.Teams)
|
||||
}
|
||||
if len(snap.Events) != 1 {
|
||||
t.Fatalf("小项数应为 1，得到 %d", len(snap.Events))
|
||||
}
|
||||
ev := snap.Events[0]
|
||||
if ev.Name != "100米" || ev.EventType != domain.EventTypeIndividual || !ev.Scored || ev.Compare != domain.CompareAsc {
|
||||
t.Fatalf("小项不符: %+v", ev)
|
||||
}
|
||||
|
||||
if len(snap.Standings) != 1 {
|
||||
t.Fatalf("榜单行数应为 1，得到 %d", len(snap.Standings))
|
||||
}
|
||||
row := snap.Standings[0]
|
||||
if row.TeamID != 1 || row.Points != 9 || row.Gold != 1 || row.Rank != 1 {
|
||||
t.Fatalf("榜单不符: %+v", row)
|
||||
}
|
||||
|
||||
if len(snap.Records) != 1 || snap.Records[0].HolderName != "李四" || snap.Records[0].ResultValue == nil {
|
||||
t.Fatalf("纪录不符: %+v", snap.Records)
|
||||
}
|
||||
if len(snap.LatestResults) != 1 {
|
||||
t.Fatalf("最新成绩数应为 1，得到 %d", len(snap.LatestResults))
|
||||
}
|
||||
lr := snap.LatestResults[0]
|
||||
if lr.ResultText != "11.58" || lr.TeamID != 1 || lr.AthleteID != 1 || lr.Rank != 1 || lr.State != "published" {
|
||||
t.Fatalf("最新成绩不符: %+v", lr)
|
||||
}
|
||||
}
|
||||
|
||||
func TestEmptySnapshotMarshalsEmptyLists(t *testing.T) {
|
||||
env := newTestEnv(t)
|
||||
snap, err := env.hub.Snapshot(context.Background())
|
||||
if err != nil {
|
||||
t.Fatalf("构建空快照: %v", err)
|
||||
}
|
||||
raw, err := json.Marshal(snap)
|
||||
if err != nil {
|
||||
t.Fatalf("序列化快照: %v", err)
|
||||
}
|
||||
for _, want := range []string{`"teams":[]`, `"events":[]`, `"standings":[]`, `"records":[]`, `"latestResults":[]`} {
|
||||
if !strings.Contains(string(raw), want) {
|
||||
t.Fatalf("空快照应包含 %s，实际 %s", want, raw)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func mustExec(t *testing.T, db *sql.DB, q string, args ...any) {
|
||||
t.Helper()
|
||||
if _, err := db.Exec(q, args...); err != nil {
|
||||
t.Fatalf("执行 SQL %q: %v", q, err)
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user