AskaEth
7f9701c465
- domain/rules: 计分与排名解耦、破纪录按组别、tieBreak 生效、调整分不丢、移除 aggregate 假能力 - realtime: 新建 internal/dto，快照三项契约对齐，/ws 只认会话身份 - httpapi: 流水幂等、standings 整表载荷、广播接线、强制改密、XFF 防伪造、records 写入、审计、赛事包内容、发布前备份、result.points、实例列表 - web: FormData 上传、YAML 解析、快照归一化、大屏解析层、bootstrap 降级、访客判定、主题切换 - console: 优雅关闭握手、中文化、免验证改管理员账号、选屏与日志轮转修复 - docs: 用户文档三份 + README - 主代理: 破纪录加分接线（levels 未加载导致恒 0）
112 lines
3.1 KiB
Go
112 lines
3.1 KiB
Go
package realtime
|
||
|
||
import (
|
||
"context"
|
||
"database/sql"
|
||
"encoding/json"
|
||
"fmt"
|
||
)
|
||
|
||
// currentVersion 返回服务端版本号，即 change_log 的最大 id；空表为 0。
|
||
// NewHub 允许 db 为 nil（降级/测试），此时退回内存版本，避免空指针 panic。
|
||
func (h *Hub) currentVersion(ctx context.Context) (int64, error) {
|
||
if h.db == nil {
|
||
return h.Version(), nil
|
||
}
|
||
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)
|
||
}
|