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
	// lastSeen 是最近一次真正收到对端帧的时间（unix 纳秒），含普通数据帧与 ping 的 pong。
	// 「这条连接还活着吗」只认它：服务端自己发过什么、写过几次 socket 都不算证据。
	lastSeen atomic.Int64
}

func newClient(h *Hub, conn *websocket.Conn, remoteAddr string, id wsIdentity) *client {
	c := &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,
	}
	// 建立连接本身就是一次「听到对端」：不初始化的话，存活探测会把刚连上的
	// 大屏当成已经静默很久，第一条 ping 立刻发出去。
	c.markSeen(time.Now())
	return c
}

// markSeen 记下「刚刚听到了对端」。
func (c *client) markSeen(now time.Time) { c.lastSeen.Store(now.UnixNano()) }

// lastSeenAt 返回最近一次听到对端的时间。
func (c *client) lastSeenAt() time.Time { return time.Unix(0, c.lastSeen.Load()) }

// instanceIDValue 读实例 id；调用方拿到的是快照值，不持有锁。
func (c *client) instanceIDValue() string {
	c.mu.Lock()
	defer c.mu.Unlock()
	return c.instanceID
}

// isDone 判断本连接是否已经进入关闭流程。
func (c *client) isDone() bool {
	select {
	case <-c.done:
		return true
	default:
		return false
	}
}

// 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 与 fd。
// 读写都在独立 goroutine 里跑，HTTP 中间件的 recover 覆盖不到，必须各自兜底。
func (c *client) readLoop(ctx context.Context) {
	defer c.stop(websocket.StatusNormalClosure, "")
	defer c.guardLoop("读")

	// 握手 deadline 在进入循环时算一次，整个「未握手」阶段共用它。
	//
	// 以前是每轮 Read 都新建一个 WithTimeout：只要对端在超时前随便发一帧（哪怕是个
	// 客户端不认识的事件），计时就被重置，未握手的连接可以无限期挂着——
	// 现场用脚本反复重连就能堆出一堆永不释放的连接。绝对 deadline 让「打开连接却不
	// hello」在固定时间后必然被断开；已握手的连接不受它限制。
	helloDeadline := time.Now().Add(c.hub.helloTimeout())

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

		readCtx := ctx
		var cancel context.CancelFunc
		if !greeted {
			readCtx, cancel = context.WithDeadline(ctx, helloDeadline)
		}
		_, data, err := c.conn.Read(readCtx)
		if cancel != nil {
			cancel()
		}
		if err != nil {
			return
		}
		// 收到任何一帧都是「对端还活着」的证据，与它是不是合法事件无关。
		c.markSeen(time.Now())
		c.handleMessage(ctx, data)
	}
}

// livenessLoop 定期求证对端还在不在：静默（没有任何 C→S 帧）超过一个探测周期就发一次
// WebSocket ping 并等 pong，连续 maxProbeFailures 次等不到就断开这条连接。
//
// 为什么必须单独探：大屏在数据没变时只收不发（hello 只在连接与翻页时发、ack 只在应用
// 变更时发），所以「没有上行帧」不等于掉线，不能当判据；而 TCP 层面「连接对象还在」
// 什么也证明不了——拔网线、断电、机器休眠都不会让服务端立刻收到错误，内核缓冲会让写
// 一直“成功”，注册表里的「在线」就会永远挂着那块已经不在的屏。
// ping/pong 是唯一的真实往返，且由浏览器在网络层自动应答，不需要前端改代码。
func (c *client) livenessLoop(ctx context.Context) {
	defer c.guardLoop("存活探测")

	failures := 0
	for {
		interval := c.hub.livenessIntervalValue()
		if interval <= 0 {
			// <= 0 表示关闭探测，只给「单独验证某个判定口径」的用例用：
			// 生产路径上 NewHub 一定给了默认值，不会走到这里。
			return
		}
		timer := time.NewTimer(interval)
		select {
		case <-ctx.Done():
			timer.Stop()
			return
		case <-c.done:
			timer.Stop()
			return
		case <-timer.C:
		}

		if time.Since(c.lastSeenAt()) < interval {
			// 静默期内收到过真实上行帧：这条连接有人，不必多问一句。
			failures = 0
			continue
		}
		if c.probePeer(ctx) {
			failures = 0
			continue
		}
		if ctx.Err() != nil || c.isDone() {
			return
		}
		failures++
		if failures < maxProbeFailures {
			// 一次不回应可能只是对端正忙（例如正在收一份大快照，写锁被占住），
			// 再给一个周期；连续失败才说明这条连接上已经没有人了。
			continue
		}
		c.hub.log.Info("实时连接长时间无响应（未回 pong），按掉线断开",
			"remote", c.remoteAddr, "instance", c.instanceIDValue(),
			"silent", time.Since(c.lastSeenAt()).Truncate(time.Second).String())
		c.stop(unresponsiveCloseCode, "连接无响应")
		return
	}
}

// probePeer 发一次 ping 并等 pong；返回 true 表示对端应答了。
func (c *client) probePeer(ctx context.Context) bool {
	pingCtx, cancel := context.WithTimeout(ctx, c.hub.pongTimeoutValue())
	defer cancel()

	if err := c.conn.Ping(pingCtx); err != nil {
		c.hub.log.Debug("存活探测未收到 pong", "remote", c.remoteAddr, "err", err)
		return false
	}
	now := time.Now()
	c.markSeen(now)
	// 这次往返是真实的客户端应答，注册表的「最后活跃」也该跟着走。
	c.hub.touchInstance(c, now)
	return true
}

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