Scoreboard/server/internal/httpapi/realtime.go

package httpapi

import (
	"context"
	"database/sql"
	"errors"
	"net/http"

	"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
}

// 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)
}

// 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{}
	}
	return list, nil
}