Scoreboard/server/internal/httpapi/realtime.go

package httpapi

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"log/slog"
	"net/http"
	"strings"
	"unicode/utf8"

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

// Realtime 是实时层的最小接口；未接入时写操作只落库，不推送。
type Realtime interface {
	Broadcast(changes []realtime.Change)
	LoadChanges(ctx context.Context, sinceVersion int64, limit int) ([]realtime.Change, error)
	Instances() []realtime.Instance
}

// ---------- 大屏别名（机器级设置） ----------
//
// 别名与实例注册表是两套生命周期：
//   - 实例注册表在**内存**里，连接断开后只留 TTL（十分钟）供排查；
//   - 别名是操作员给「这台设备」起的名字，必须活过重启、活过实例过期。
//
// 所以别名存系统库 system.db 的 settings 表（与 PUT /api/system/session 的登录有效期同一套
// 读写机制，见 repo_misc.go 的 writeSessionHours），不新建表、不写迁移。
const (
	// instanceAliasKey 下存一个 JSON 映射：{"<instanceId>": "<别名>"}。
	instanceAliasKey = "display.instance_aliases"
	// maxInstanceAliases 是映射条数上限：别名会在实例过期后继续留着（下次那台设备连上来
	// 还认得出），只增不减会一路涨下去，必须有封顶。
	maxInstanceAliases = 200
	// maxAliasChars 是单个别名的字符数上限（按字符不是字节，中文名才存得下）。
	maxAliasChars = 40
)

// displayAliasPatch 是 PUT /api/display/instances/{id}/alias 的请求体。
type displayAliasPatch struct {
	// Alias 为空串或只空白 = 清除别名、回到自动设备标签。
	Alias     string `json:"alias"`
	RequestID string `json:"requestId"`
}

// displayInstancesHandler 返回在线大屏实例列表（在线看板）。
func (s *server) displayInstancesHandler(w http.ResponseWriter, r *http.Request) (any, error) {
	if s.realtime == nil {
		return []realtime.Instance{}, nil
	}
	list := s.realtime.Instances()
	if list == nil {
		list = []realtime.Instance{}
	}
	// 别名在这里与实例合并：注册表里只剩在线的与刚离线的，而别名可能属于一台早就
	// 被清理掉的设备——那种别名留着不填（没有对应实例），等它连上来自然就填上了。
	aliases, err := s.loadInstanceAliases(r.Context())
	if err != nil {
		// 读别名失败不该让「在线大屏」列表整个打不开：现场靠它切页、排查。
		// 降级成「没有别名」（列表退回显示实例 id）并留下警告，其余字段照常返回。
		s.logger.Warn("读取大屏别名失败，列表按无别名返回", "err", err)
		return list, nil
	}
	for i := range list {
		if alias := aliases[list[i].ID]; alias != "" {
			list[i].Alias = alias
		}
	}
	return list, nil
}

// updateDisplayInstanceAliasHandler 处理 PUT /api/display/instances/{id}/alias：
// 给一台设备设 / 清别名，返回该实例的最新记录（形状与列表元素一致，即 Instance DTO）。
// 权限与大屏切页同档（recorder 及以上），见 server.go 的路由注册。
func (s *server) updateDisplayInstanceAliasHandler(w http.ResponseWriter, r *http.Request) (any, error) {
	id := strings.TrimSpace(r.PathValue("id"))
	var in displayAliasPatch
	if err := decodeJSON(r, &in); err != nil {
		return nil, err
	}
	alias := realtime.CleanUserText(in.Alias)
	if utf8.RuneCountInString(alias) > maxAliasChars {
		return nil, badRequest(
			fmt.Sprintf("别名最长 %d 个字符（当前 %d 个）", maxAliasChars, utf8.RuneCountInString(alias)),
			nil)
	}

	inst, ok := s.findDisplayInstance(id)
	if !ok {
		return nil, notFound("这台大屏当前不在注册表里：它可能已经离线很久、超过了保留时间。请刷新在线列表后重试。")
	}

	return nil, s.idempotent(w, r, in.RequestID, func() (any, error) {
		if err := s.saveInstanceAlias(r.Context(), id, alias); err != nil {
			return nil, err
		}
		s.audit(r.Context(), "display.instance_alias", "display", 0, nil,
			map[string]any{"instanceId": id, "alias": alias})
		inst.Alias = alias
		return inst, nil
	})
}

// findDisplayInstance 在注册表快照里按 id 找实例；找不到说明它已被清理（或压根没连过）。
func (s *server) findDisplayInstance(id string) (realtime.Instance, bool) {
	if id == "" || s.realtime == nil {
		return realtime.Instance{}, false
	}
	for _, inst := range s.realtime.Instances() {
		if inst.ID == id {
			return inst, true
		}
	}
	return realtime.Instance{}, false
}

// loadInstanceAliases 读「实例 id → 别名」映射。
//
// 存量值可能被手工改过、或是别的版本写的，所以读出来还要过一遍清洗：空 id / 空别名等于
// 「没设」，直接丢掉（避免死条目一直占着条数上限），别名本身按同一个口径去控制字符与截断。
func (s *server) loadInstanceAliases(ctx context.Context) (map[string]string, error) {
	return loadInstanceAliasesQ(ctx, s.sys, s.logger)
}

// loadInstanceAliasesQ 与 loadInstanceAliases 相同，但可以在给定连接 / 事务上读：
// saveInstanceAlias 要在同一个事务里「先读后写」，否则两个人同时改不同设备时，
// 后写的那个会把前一个的改动覆盖掉（两边各自读到旧映射，谁也看不见谁）。
func loadInstanceAliasesQ(ctx context.Context, q queryer, logger *slog.Logger) (map[string]string, error) {
	var raw string
	err := q.QueryRowContext(ctx, `SELECT value FROM settings WHERE key = ?`, instanceAliasKey).Scan(&raw)
	switch {
	case errors.Is(err, sql.ErrNoRows):
		return map[string]string{}, nil
	case err != nil:
		return nil, internal(err)
	}
	if strings.TrimSpace(raw) == "" {
		return map[string]string{}, nil
	}
	var stored map[string]string
	if err := json.Unmarshal([]byte(raw), &stored); err != nil {
		// 值被改坏时按「没有别名」处理并留下警告：一条脏设置不该把「在线大屏」列表打不开。
		logger.Warn("大屏别名设置不是合法 JSON，按无别名处理", "key", instanceAliasKey, "err", err)
		return map[string]string{}, nil
	}
	out := make(map[string]string, len(stored))
	for id, alias := range stored {
		id = strings.TrimSpace(id)
		alias = realtime.TruncateChars(realtime.CleanUserText(alias), maxAliasChars)
		if id == "" || alias == "" {
			continue
		}
		out[id] = alias
	}
	return out, nil
}

// saveInstanceAlias 写入 / 清除一个实例的别名。
//
// 清除是真的把那条**删掉**（不是存空串）：实例过期后别名会一直留着，只增不减的映射会一路
// 顶到 maxInstanceAliases，之后所有设备都改不了名。
//
// 超过条数上限时**拒绝**而不是淘汰最旧的：别名是按 id 存的、没有时间戳，淘汰谁都是猜；
// 而悄悄弄丢一台设备的命名比让操作员先清几条更难查。拒绝会带中文提示（409）。
func (s *server) saveInstanceAlias(ctx context.Context, id, alias string) error {
	// 读—改—写必须在一个事务里（系统库只有一条连接，事务把并发改名串起来）。
	tx, err := s.sys.BeginTx(ctx, nil)
	if err != nil {
		return internal(err)
	}
	defer func() { _ = tx.Rollback() }()

	aliases, err := loadInstanceAliasesQ(ctx, tx, s.logger)
	if err != nil {
		return err
	}
	if alias == "" {
		if _, ok := aliases[id]; !ok {
			return nil // 本来就没有别名：不必写库
		}
		delete(aliases, id)
	} else {
		if _, exists := aliases[id]; !exists && len(aliases) >= maxInstanceAliases {
			s.logger.Warn("大屏别名数量已达上限，拒绝新增",
				"max", maxInstanceAliases, "instanceId", id)
			return conflict(fmt.Sprintf(
				"大屏别名最多 %d 条，已经存满。请先给不再使用的设备清除别名后重试。", maxInstanceAliases))
		}
		aliases[id] = alias
	}

	raw, err := json.Marshal(aliases)
	if err != nil {
		return internal(err)
	}
	if _, err := tx.ExecContext(ctx,
		`INSERT INTO settings (key, value, updated_at) VALUES (?, ?, datetime('now'))
		 ON CONFLICT(key) DO UPDATE SET value = excluded.value, updated_at = datetime('now')`,
		instanceAliasKey, string(raw)); err != nil {
		return internal(err)
	}
	if err := tx.Commit(); err != nil {
		return internal(err)
	}
	return nil
}

// broadcastChanges 在写请求成功返回后，把本次请求写入 change_log 的变更推给实时层。
// 版本号取请求开始时的最大 id，因此只推送本次写入，不会重复补发历史变更。
func (s *server) broadcastChanges(next http.Handler) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if s.realtime == nil || s.ev == nil || !isWriteMethod(r.Method) {
			next.ServeHTTP(w, r)
			return
		}
		before, ok := s.changeVersion(r.Context())
		if !ok {
			next.ServeHTTP(w, r)
			return
		}

		rec := &statusRecorder{ResponseWriter: w, status: http.StatusOK}
		next.ServeHTTP(rec, r)
		if rec.status >= http.StatusBadRequest {
			return
		}

		// 请求结束后 r.Context() 可能已被取消，推送不应因此中断。
		ctx := context.WithoutCancel(r.Context())
		changes, err := s.realtime.LoadChanges(ctx, before, 0)
		if err != nil {
			if dbClosed(err) {
				// 恢复流程会在请求末尾关掉目标库再换文件，广播中间件在这之后才读 change_log，
				// 必然撞上「库已关闭」。这是停机动作的一部分，不是故障——
				// 记 ERROR 只会让现场以为恢复把系统搞坏了。降级成 Debug，其余错误照旧。
				s.logger.Debug("跳过实时推送：库已关闭（恢复/重启中）", "err", err)
				return
			}
			s.logger.Error("读取实时变更失败", "err", err)
			return
		}
		if len(changes) > 0 {
			s.realtime.Broadcast(changes)
		}
	})
}

// errDBClosed 是 database/sql 在连接池已关闭时返回的错误。
//
// 标准库里这个错误是**未导出**的（database/sql 的 errDBClosed），拿不到它的变量，
// 只能按原文认；这里独立声明一份，比对时逐层走错误链看有没有同文本的那一层。
var errDBClosed = errors.New("sql: database is closed")

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

// changeVersion 读取 change_log 当前最大 id；失败时调用方跳过本次推送。
func (s *server) changeVersion(ctx context.Context) (int64, bool) {
	var v sql.NullInt64
	if err := s.ev.QueryRowContext(ctx, `SELECT MAX(id) FROM change_log`).Scan(&v); err != nil {
		if dbClosed(err) {
			// 同 broadcastChanges：库已关闭是恢复/重启的中间状态，不是故障。
			s.logger.Debug("跳过变更版本：库已关闭（恢复/重启中）", "err", err)
			return 0, false
		}
		s.logger.Error("读取变更版本失败", "err", err)
		return 0, false
	}
	return v.Int64, true
}

func isWriteMethod(method string) bool {
	switch method {
	case http.MethodGet, http.MethodHead, http.MethodOptions:
		return false
	default:
		return true
	}
}

// statusRecorder 记录响应状态码，用于判断写请求是否成功。
type statusRecorder struct {
	http.ResponseWriter
	status  int
	written bool
}

func (w *statusRecorder) WriteHeader(code int) {
	if !w.written {
		w.status = code
		w.written = true
	}
	w.ResponseWriter.WriteHeader(code)
}

func (w *statusRecorder) Write(b []byte) (int, error) {
	if !w.written {
		w.written = true
	}
	return w.ResponseWriter.Write(b)
}