Scoreboard/server/internal/realtime/client.go

package realtime

import (
	"context"
	"encoding/json"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"github.com/coder/websocket"
)

// client 是一条 WebSocket 连接。读写各一个 goroutine，
// 写通过带缓冲的 send 通道串行化，通道满即视为慢客户端。
type client struct {
	hub        *Hub
	conn       *websocket.Conn
	remoteAddr string

	send chan []byte
	done chan struct{}

	closeOnce   sync.Once
	closeCode   websocket.StatusCode
	closeReason string

	mu      sync.Mutex
	greeted bool
	// role/userID 是 hello 之后实际生效的身份；sessionRole/sessionUserID 是握手时从会话取到的上限，
	// 客户端自报的 role/userId 不能越过它。
	role          string
	userID        int64
	sessionRole   string
	sessionUserID int64
	// sessionID 是握手时的会话 id（未登录为空）：会话撤销后要据此复查并断开本连接。
	sessionID  string
	instanceID string
	// settingsAdmin 表示会话是否为 admin：recorder 也进管理端房间，但读不到系统配置。
	settingsAdmin bool

	ackVersion atomic.Int64 // 客户端已确认应用到的版本
	lastSent   atomic.Int64 // 服务端已发出的最大版本
	// snapshotAt 是本连接最近一次发出整库快照的时间（unix 纳秒）。
	// 它是「整库快照窗口」的记账点：resync 与不带 lastVersion 的 hello 都会走到发快照，
	// 两者共用一个窗口，不能靠交替发把限流绕过去。
	snapshotAt atomic.Int64
	// snapshotPending 表示「有请求要过整库快照，但被窗口挡下了」：
	// 心跳循环会在窗口过后补发一份，见 flushPendingSnapshot。
	// 不能把被挡下的请求直接丢掉——大屏收到它不认识的变更时会主动 resync，
	// 丢了就可能让那块屏一直停在旧数据上，直到下一条变更到来。
	snapshotPending atomic.Bool
	slow            atomic.Bool
}

func newClient(h *Hub, conn *websocket.Conn, remoteAddr string, id wsIdentity) *client {
	return &client{
		hub:           h,
		conn:          conn,
		remoteAddr:    remoteAddr,
		send:          make(chan []byte, defaultSendBuffer),
		done:          make(chan struct{}),
		sessionRole:   id.Role,
		sessionUserID: id.UserID,
		sessionID:     id.SessionID,
		settingsAdmin: id.SettingsAdmin,
	}
}

// stop 只负责让两条 goroutine 退出；真正的关闭动作由 serve 统一执行，
// 避免在持有广播锁时做可能阻塞的关闭握手。
func (c *client) stop(code websocket.StatusCode, reason string) {
	c.closeOnce.Do(func() {
		c.mu.Lock()
		c.closeCode = code
		c.closeReason = reason
		c.mu.Unlock()
		close(c.done)
	})
}

// guardLoop 兜住读写 goroutine 里的 panic：HTTP 中间件的 recover 只覆盖 handler 栈，
// 这两条 goroutine 不兜底会把整个进程带走。
func (c *client) guardLoop(name string) {
	if r := recover(); r != nil {
		c.hub.log.Error("实时循环 panic，已断开连接", "loop", name, "remote", c.remoteAddr, "panic", r)
		c.stop(websocket.StatusInternalError, name+"循环异常")
	}
}

func (c *client) sendEvent(event string, data any) bool {
	msg, err := encode(event, data)
	if err != nil {
		c.hub.log.Error("编码实时消息失败", "event", event, "err", err)
		return false
	}
	return c.sendRaw(msg)
}

// allowSnapshot 按连接给「整库快照」记账：窗口内只放行一次。
//
// 触发整库快照的入口有两个（resync，以及不带 lastVersion 的 hello），它们共用这一个窗口：
// 一条 15 字节的帧就是一次 DB 事务 + 8 条查询，而赛事库是单连接（SetMaxOpenConns(1)），
// 快照读取还在全局发送锁里——不限流的话，一个写坏的客户端就能把库连接与所有人的推送一起堵住。
// 窗口（defaultSnapshotMinInterval）比客户端自己的 3s 冷却短，正常客户端按自己的节奏回来时
// 窗口已经过期，重连补齐行为因此不受影响。
func (c *client) allowSnapshot(minInterval time.Duration) bool {
	if minInterval <= 0 {
		return true
	}
	now := time.Now().UnixNano()
	for {
		last := c.snapshotAt.Load()
		if last != 0 && now-last < int64(minInterval) {
			return false
		}
		if c.snapshotAt.CompareAndSwap(last, now) {
			return true
		}
	}
}

func (c *client) sendRaw(msg []byte) bool {
	select {
	case <-c.done:
		return false
	default:
	}
	select {
	case c.send <- msg:
		return true
	default:
		c.hub.markSlow(c)
		return false
	}
}

// readLoop 解析 C→S 消息；首条消息受 helloTimeout 限制，防止空连接占用资源。
// 读写都在独立 goroutine 里跑，HTTP 中间件的 recover 覆盖不到，必须各自兜底。
func (c *client) readLoop(ctx context.Context) {
	defer c.stop(websocket.StatusNormalClosure, "")
	defer c.guardLoop("读")

	for {
		c.mu.Lock()
		greeted := c.greeted
		c.mu.Unlock()

		readCtx := ctx
		var cancel context.CancelFunc
		if !greeted {
			readCtx, cancel = context.WithTimeout(ctx, c.hub.helloTimeout())
		}
		_, data, err := c.conn.Read(readCtx)
		if cancel != nil {
			cancel()
		}
		if err != nil {
			return
		}
		c.handleMessage(ctx, data)
	}
}

func (c *client) handleMessage(ctx context.Context, data []byte) {
	var env envelope
	if err := json.Unmarshal(data, &env); err != nil {
		c.hub.log.Debug("忽略无法解析的实时消息", "remote", c.remoteAddr, "err", err)
		return
	}

	switch env.Event {
	case EventHello:
		var p helloPayload
		if err := json.Unmarshal(env.Data, &p); err != nil {
			c.hub.log.Debug("hello 载荷不合法", "remote", c.remoteAddr, "err", err)
			return
		}
		c.hub.handleHello(ctx, c, p)

	case EventAck:
		var p ackPayload
		if err := json.Unmarshal(env.Data, &p); err != nil {
			c.hub.log.Debug("ack 载荷不合法", "remote", c.remoteAddr, "err", err)
			return
		}
		if p.Version > c.ackVersion.Load() {
			c.ackVersion.Store(p.Version)
		}
		c.hub.touchInstance(c, time.Now())

	case EventResync:
		// hello 之前的 resync 既没有房间归属，又会提前清掉 hello 超时，直接忽略。
		c.mu.Lock()
		greeted := c.greeted
		c.mu.Unlock()
		if !greeted {
			c.hub.log.Debug("忽略 hello 之前的 resync", "remote", c.remoteAddr)
			return
		}
		// 整库快照的限流统一在 sendInitial 里判定（resync 与重复 hello 共用同一个窗口）。
		c.hub.sendInitial(ctx, c, nil)

	default:
		// 客户端最常见的错法：只发了 payload，没套 `{event,data}` 外壳。
		// 那种帧解出来 Event == ""，以前只在 DEBUG 里留一行 —— 现场表现是
		// 「连上了但一直没数据，15 秒后被断开」，而且客户端拿不到任何提示。
		// 这里升到 WARN 并把「该怎么发」写进日志，省掉一次翻源码。
		if strings.TrimSpace(env.Event) == "" {
			c.hub.log.Warn("实时帧缺少 event 字段，已忽略（C→S 帧同样要套 {event,data} 外壳）",
				"remote", c.remoteAddr)
			return
		}
		c.hub.log.Warn("忽略未知实时事件", "event", env.Event, "remote", c.remoteAddr)
	}
}

// writeLoop 是唯一的写者，保证同一连接上的帧顺序。
func (c *client) writeLoop(ctx context.Context) {
	defer c.guardLoop("写")
	for {
		select {
		case <-ctx.Done():
			return
		case <-c.done:
			return
		case msg := <-c.send:
			writeCtx, cancel := context.WithTimeout(ctx, c.hub.writeTimeout())
			err := c.conn.Write(writeCtx, websocket.MessageText, msg)
			cancel()
			if err != nil {
				c.stop(websocket.StatusInternalError, "写入失败")
				return
			}
		}
	}
}