package realtime import ( "context" "database/sql" "encoding/json" "fmt" "git.yeij.top/AskaEth/Scoreboard/server/internal/dto" ) // 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) } // entitySettings 是赛事配置在 change_log.entity_type 里的取值; // 它的载荷是完整的 dto.EventSettings,含只有 admin 能读的系统配置字段。 const entitySettings = "settings" // redactChange 按连接身份裁剪一条变更的载荷,返回不与其它连接共享的新对象。 // // 快照对非 admin 会套 WithoutAdminFields 砍掉 defaultProfileId / requireApproval / // individualToTeamWeight,但 change 与 catchup 走的是另一条路径:不在这里做同样处理, // 任何打开 /ws 的浏览器都能从增量里读到这三个字段,快照上的裁剪等于被旁路。 // 只动 settings 这一个实体,其余实体的载荷本来就是公开的赛事数据。 func redactChange(ch Change, settingsAdmin bool) Change { if settingsAdmin || ch.Entity != entitySettings { return ch } var settings dto.EventSettings if err := json.Unmarshal(ch.Data, &settings); err != nil { // 载荷不是 EventSettings 形状(历史数据 / 手工改库):宁可只下发空对象, // 也不能把可能含系统配置的原文透给非 admin。 ch.Data = json.RawMessage(`{}`) return ch } data, err := json.Marshal(settings.WithoutAdminFields()) if err != nil { ch.Data = json.RawMessage(`{}`) return ch } ch.Data = data return ch } // redactChanges 批量裁剪;admin 连接原样返回,不做无谓的复制。 func redactChanges(changes []Change, settingsAdmin bool) []Change { if settingsAdmin { return changes } out := make([]Change, 0, len(changes)) for _, ch := range changes { out = append(out, redactChange(ch, settingsAdmin)) } return out }