Scoreboard/server/internal/realtime/changes.go

package realtime

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"

	"git.yeij.top/AskaEth/Scoreboard/server/internal/dto"
)

// errDBClosed 是 database/sql 在连接池已关闭时返回的错误。
//
// 标准库里这个错误是**未导出**的（database/sql 的 errDBClosed），拿不到变量，只能按原文认。
// httpapi 包里有一份同口径的实现（那一份也导不出来，实时层又依赖不了 httpapi——会成环），
// 两份必须保持同一套语义：库已关闭 = 恢复/重启动作的中间状态，不是故障。
var errDBClosed = errors.New("sql: database is closed")

// dbClosed 判定错误是否只是「库已经被关掉」。
//
// 只认原文完全相等的那一层（含被 %w 包过的），不按子串匹配：别的读库失败（锁超时、
// SQL 写错、库损坏）都得继续按 WARN/ERROR 报出来，不能被这条静默规则顺手吞掉。
func dbClosed(err error) bool {
	for e := err; e != nil; e = errors.Unwrap(e) {
		if e.Error() == errDBClosed.Error() {
			return true
		}
	}
	return false
}

// 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 在数据库不可用时退回内存里的版本，保证心跳与广播不会因一次查询失败而中断。
// 「库已关闭」（恢复/重启窗口）只记 Debug：那不是故障，而且它每个心跳周期都会发生一次，
// 记 WARN 会让现场日志刷成一片红。
func (h *Hub) currentVersionCached(ctx context.Context) int64 {
	if h.db == nil {
		return h.Version()
	}
	v, err := h.currentVersion(ctx)
	if err != nil {
		if dbClosed(err) {
			h.log.Debug("读取版本号：库已关闭（恢复/重启中），沿用内存版本", "version", h.Version())
			return h.Version()
		}
		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
}