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 }