noitu/server/internal/wsapi/session.go

package wsapi

import (
	"context"
	"crypto/rand"
	"encoding/base64"
	"errors"
	"log/slog"
	"sync"
	"time"

	"github.com/coder/websocket"
	noituv1 "github.com/tiennm99dev/noitu/server/gen/noitu/v1"
	"github.com/tiennm99dev/noitu/server/internal/game"
)

const (
	// outboxCap buffers writes. A client that cannot keep up with this many
	// pending frames is not going to catch up, so the session is closed rather
	// than grown without bound.
	outboxCap = 32

	// pingEvery / pingTimeout are the liveness check. Reads carry no deadline
	// of their own: a player idling in the lobby between games is healthy and
	// silent, and a read timeout cannot tell that apart from a dead socket.
	// A ping can.
	pingEvery   = 20 * time.Second
	pingTimeout = 10 * time.Second
	pingMisses  = 2

	// writeTimeout bounds a single frame write, and drainTimeout the final
	// flush of whatever is still queued when the session ends.
	writeTimeout = 10 * time.Second
	drainTimeout = 2 * time.Second

	// submitsPerSecond bounds word submissions. Each one is a dictionary
	// lookup and a possible engine mutation; a human types far below this.
	submitsPerSecond = 5
	submitBurst      = 10

	// Chat gets its own budget so talking never costs a move. It can afford to
	// be humane about a burst — two people typing at each other is normal —
	// because the danger a limiter would otherwise be holding down is handled
	// where it actually lives: chat is delivered with trySend, so a recipient
	// who cannot keep up drops a line rather than losing their session.
	chatsPerSecond = 2.0
	chatBurst      = 5

	joinsPerSecond = 1
	joinBurst      = 5

	// Room creation is far more expensive than a join: each one is a
	// goroutine, an engine and a registry entry held until the game ends.
	roomsPerSecond = 0.2
	roomBurst      = 5
	limiterIdleFor = 5 * time.Minute

	// framesPerSecond bounds every frame a connection sends, before it is
	// routed. The per-action limiters above only meter the actions they know
	// about; a Ping, or a ClientMessage with no payload set, matched none of
	// them and cost the reader a decode at line rate. A client past this is
	// not a player typing, so the connection is closed rather than throttled.
	framesPerSecond = 20
	frameBurst      = 40
)

var (
	errHandshake = errors.New("wsapi: first message must be Hello")
	errFlood     = errors.New("wsapi: frame rate exceeded")
)

// session is one WebSocket connection.
//
// Exactly one reader goroutine and one writer goroutine touch the socket. The
// mutex below guards only the small mutable identity — nickname, room, seat —
// that the reader sets and the room goroutine reads.
type session struct {
	id          string
	resumeToken string
	conn        *websocket.Conn
	hub         *hub

	// remoteIP keys the join limiter. A session id is minted per connection,
	// so keying on it would let anyone brute-force room codes by reconnecting
	// between attempts — the limiter has to outlive the socket.
	remoteIP string

	// ctx is the teardown signal for everything except the read.
	ctx    context.Context
	cancel context.CancelFunc

	// readCtx is cancelled only after the writer has finished flushing.
	//
	// coder/websocket arms a context.AfterFunc on the context passed to Read
	// that hard-closes the underlying socket when it fires, so cancelling the
	// read context is the same as destroying the connection. Every frame
	// queued at teardown — the shutdown notice above all — would be written
	// into a socket that is already gone.
	//
	// It is rooted at Background rather than at the server context on purpose.
	// A child of the server context would be cancelled by Shutdown at the same
	// instant as ctx, which is precisely the ordering this exists to prevent.
	// Nothing leaks: run always signals ctx, and the goroutine watching it
	// always cancels this one.
	readCtx    context.Context
	cancelRead context.CancelFunc

	out chan []byte

	// flushed closes when the writer has drained, so teardown can wait for the
	// last frames to leave before tearing the socket down.
	flushed chan struct{}

	mu       sync.Mutex
	nick     string
	room     *room
	playerID game.PlayerID

	submitLimiter *bucket
	roomLimiter   *bucket
	chatLimiter   *bucket
	frameLimiter  *bucket

	// greeted marks the handshake done. It is a one-shot transition: a second
	// Hello would re-register the session and rewrite its nickname mid-game.
	greeted bool

	closeOnce sync.Once
}

func newSession(ctx context.Context, conn *websocket.Conn, h *hub, remoteIP string) *session {
	readCtx, cancelRead := context.WithCancel(context.Background())
	ctx, cancel := context.WithCancel(ctx)
	return &session{
		readCtx:       readCtx,
		cancelRead:    cancelRead,
		flushed:       make(chan struct{}),
		id:            randomToken(),
		resumeToken:   randomToken(),
		remoteIP:      remoteIP,
		conn:          conn,
		hub:           h,
		ctx:           ctx,
		cancel:        cancel,
		out:           make(chan []byte, outboxCap),
		submitLimiter: newBucket(submitsPerSecond, submitBurst, time.Now()),
		roomLimiter:   newBucket(roomsPerSecond, roomBurst, time.Now()),
		chatLimiter:   newBucket(chatsPerSecond, chatBurst, time.Now()),
		frameLimiter:  newBucket(framesPerSecond, frameBurst, time.Now()),
	}
}

func (s *session) nickname() string {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.nick
}

func (s *session) setNickname(n string) {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.nick = n
}

// attach binds this connection to a room seat.
func (s *session) attach(r *room, seatName string) {
	s.mu.Lock()
	previous, previousID := s.room, s.playerID
	s.room = r
	s.playerID = playerIDFor(seatName)
	s.mu.Unlock()

	// Releasing the old room is not tidiness. Nothing else tells it this
	// connection has gone: leaveRoom only ever notifies the current room, so an
	// unreleased room parks in select forever, holding a goroutine and a room
	// code for the life of the process. One connection asking for several rooms
	// is all it takes.
	if previous != nil && previous != r {
		previous.send(disconnectInput{player: previousID, sess: s})
	}
}

// release forgets a room this connection is no longer seated in, because it
// left or was kicked. The connection itself stays open.
//
// Guarded by identity: a release from a room the connection has already moved
// on from must not detach it from the one it is sitting in now.
func (s *session) release(r *room) {
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.room == r {
		s.room = nil
		s.playerID = ""
	}
}

func (s *session) currentRoom() (*room, game.PlayerID) {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.room, s.playerID
}

// send queues a message for the writer goroutine.
//
// Never blocks: the room goroutine calls this, and one unresponsive client
// must not be able to stall the game its opponent is still playing. A full
// outbox closes the session instead.
func (s *session) send(m *noituv1.ServerMessage) {
	raw, err := Encode(m)
	if err != nil {
		slog.Error("encode failed", "session", s.id, "err", err)
		return
	}

	select {
	case s.out <- raw:
	case <-s.ctx.Done():
	default:
		slog.Warn("outbox full, closing session", "session", s.id)
		s.close()
	}
}

// close signals teardown. It does not cancel the read context: that is done by
// run once the writer has flushed, so a client is told why it is being
// disconnected before the socket goes.
func (s *session) close() {
	s.closeOnce.Do(func() {
		s.cancel()
	})
}

// trySend queues a message and reports whether it fit.
//
// The difference from send is what a full outbox means: send closes the
// session, on the grounds that a client this far behind will not catch up.
// That is right for a game frame and wrong for a chat line, because it hands
// one player a way to disconnect the other into losing by abandonment.
//
// A line is droppable because the next replay carries it. A ChatHistory is
// not — it is the frame that corrects a whole panel, and there is nothing
// behind it — so that one still goes through send. This is for ChatMessage.
func (s *session) trySend(m *noituv1.ServerMessage) bool {
	raw, err := Encode(m)
	if err != nil {
		slog.Error("encode failed", "session", s.id, "err", err)
		return false
	}

	select {
	case s.out <- raw:
		return true
	case <-s.ctx.Done():
		return false
	default:
		slog.Warn("outbox full, dropping chat", "session", s.id)
		return false
	}
}

// run drives the connection until it closes.
func (s *session) run() {
	defer s.close()
	defer s.leaveRoom()
	// A connection that ends while queued must not leave a ghost in line: the
	// next two strangers to ask are paired with each other, not with a socket
	// that is already gone.
	defer s.hub.cancelQuickMatch(s)

	s.conn.SetReadLimit(maxFrameBytes)

	var wg sync.WaitGroup
	wg.Add(3)
	go func() { defer wg.Done(); s.writeLoop() }()
	go func() { defer wg.Done(); s.keepalive() }()

	// Teardown ordering lives in its own goroutine because readLoop below is
	// blocked in Read and cannot run it. Whoever signals the close — a
	// protocol error here, a dead peer in keepalive, or Shutdown cancelling
	// the server context — gets the same sequence: flush, then drop the
	// socket.
	go func() {
		defer wg.Done()
		<-s.ctx.Done()
		select {
		case <-s.flushed:
		case <-time.After(drainTimeout):
		}
		s.cancelRead()
	}()

	err := s.readLoop()
	s.close()
	wg.Wait()

	// readLoop only ever returns an error; a cancelled context is the one
	// that means the server chose to close.
	status, reason := websocket.StatusNormalClosure, ""
	if !errors.Is(err, context.Canceled) {
		status, reason = websocket.StatusPolicyViolation, "protocol error"
	}
	_ = s.conn.Close(status, reason)
}

// readLoop is the only reader of the socket.
func (s *session) readLoop() error {
	for {
		typ, raw, err := s.conn.Read(s.readCtx)
		if err != nil {
			return err
		}
		if !s.frameLimiter.allow(time.Now()) {
			s.send(errorMsg("too_fast"))
			return errFlood
		}

		msg, err := Decode(typ, raw)
		if err != nil {
			s.send(errorMsg("bad_frame"))
			return err
		}
		if err := s.dispatch(msg); err != nil {
			return err
		}
	}
}

// writeLoop is the only writer of the socket. A single owner keeps frame order
// deterministic even though coder/websocket tolerates concurrent writes.
func (s *session) writeLoop() {
	defer close(s.flushed)

	for {
		select {
		case <-s.ctx.Done():
			s.drain()
			return
		case raw := <-s.out:
			if !s.write(raw) {
				return
			}
		}
	}
}

// write sends one frame.
//
// The deadline is its own, not derived from the session context. A frame that
// has already been dequeued must still reach the peer even when the session is
// ending — refusals are sent immediately before a close, and inheriting the
// cancelled context would fail every one of them. The timeout is what protects
// against a peer that has stopped reading.
func (s *session) write(raw []byte) bool {
	ctx, cancel := context.WithTimeout(context.Background(), writeTimeout)
	defer cancel()

	if err := s.conn.Write(ctx, websocket.MessageBinary, raw); err != nil {
		s.close()
		return false
	}
	return true
}

// drain flushes what is already queued after the session is cancelled.
//
// Refusals are the reason this exists: the server answers a bad handshake with
// a UI key and then closes, and without this the close wins the race and the
// client is left to guess why it was dropped. The context is fresh because
// s.ctx is by definition already cancelled here, and bounded because a peer
// that is not reading must not delay teardown.
func (s *session) drain() {
	for {
		select {
		case raw := <-s.out:
			ctx, cancel := context.WithTimeout(context.Background(), drainTimeout)
			err := s.conn.Write(ctx, websocket.MessageBinary, raw)
			cancel()
			if err != nil {
				return
			}
		default:
			return
		}
	}
}

// keepalive is what actually detects a dead peer, since reads have no deadline.
func (s *session) keepalive() {
	ticker := time.NewTicker(pingEvery)
	defer ticker.Stop()

	misses := 0
	for {
		select {
		case <-s.ctx.Done():
			return
		case <-ticker.C:
			ctx, cancel := context.WithTimeout(s.ctx, pingTimeout)
			err := s.conn.Ping(ctx)
			cancel()

			if err == nil {
				misses = 0
				continue
			}
			if misses++; misses >= pingMisses {
				slog.Info("peer unresponsive, closing", "session", s.id)
				s.close()
				return
			}
		}
	}
}

// dispatch routes one client message.
//
// Hello must come first: everything else needs a sanitized nickname and a
// registered resume token, and accepting them before the handshake would mean
// carrying "maybe not greeted yet" through every branch below.
func (s *session) dispatch(msg *noituv1.ClientMessage) error {
	if _, isHello := msg.GetPayload().(*noituv1.ClientMessage_Hello); !isHello && s.nickname() == "" {
		s.send(errorMsg("handshake_required"))
		return errHandshake
	}

	switch p := msg.GetPayload().(type) {
	case *noituv1.ClientMessage_Hello:
		return s.handleHello(p.Hello)

	case *noituv1.ClientMessage_StartBotGame:
		// The limiter is charged before the payload is inspected, so a bad
		// difficulty costs the same as a good one and cannot be used to probe
		// for free.
		if !s.roomLimiter.allow(time.Now()) {
			s.send(errorMsg("too_many_rooms"))
			return nil
		}
		difficulty, ok := Difficulty(p.StartBotGame.GetDifficulty())
		if !ok {
			s.send(errorMsg("unknown_difficulty"))
			return nil
		}
		if err := s.hub.startBotRoom(s, difficulty); err != nil {
			s.send(roomCreateError(s.id, err))
		}

	case *noituv1.ClientMessage_CreateRoom:
		// Creating a room allocates a goroutine and an engine, so one
		// connection must not be able to mint them without limit.
		if !s.roomLimiter.allow(time.Now()) {
			s.send(errorMsg("too_many_rooms"))
			return nil
		}
		if err := s.hub.createRoom(s); err != nil {
			s.send(roomCreateError(s.id, err))
		}

	case *noituv1.ClientMessage_JoinRoom:
		if !s.hub.joinLimiter.allow(s.remoteIP, time.Now()) {
			metrics.joinsRefused.Add("too_many_attempts", 1)
			s.send(errorMsg("too_many_attempts"))
			return nil
		}
		if err := s.hub.joinRoom(p.JoinRoom.GetRoomCode(), s); err != nil {
			metrics.joinsRefused.Add("room_not_found", 1)
			s.send(errorMsg("room_not_found"))
		}

	case *noituv1.ClientMessage_QuickMatch:
		if r, _ := s.currentRoom(); r != nil {
			s.send(errorMsg("already_in_a_room"))
			return nil
		}
		// A match mints a room exactly as CreateRoom does, so it is charged
		// the same way and for the same reason.
		if !s.roomLimiter.allow(time.Now()) {
			s.send(errorMsg("too_many_rooms"))
			return nil
		}
		if err := s.hub.quickMatch(s); err != nil {
			if errors.Is(err, errAlreadyQueued) {
				s.send(errorMsg("already_queued"))
			} else {
				s.send(roomCreateError(s.id, err))
			}
		}

	case *noituv1.ClientMessage_CancelQuickMatch:
		// Idempotent by design: a cancel that finds nothing queued is not an
		// error, it is the answer the player wanted.
		s.hub.cancelQuickMatch(s)
		s.send(quickMatchStatusMsg(false))

	case *noituv1.ClientMessage_SubmitWord:
		s.handleSubmit(p.SubmitWord)

	case *noituv1.ClientMessage_Resign:
		// A silently dropped resignation leaves the player staring at a board
		// they thought they had left.
		if r, id := s.currentRoom(); r != nil {
			if !r.send(resignInput{sess: s, player: id}) {
				s.send(errorMsg("game_already_over"))
			}
		} else {
			s.send(errorMsg("not_in_a_game"))
		}

	case *noituv1.ClientMessage_SetReady:
		s.toRoom(lobbyInput{sess: s, action: lobbyReady, ready: p.SetReady.GetReady()})

	case *noituv1.ClientMessage_StartGame:
		s.toRoom(lobbyInput{sess: s, action: lobbyStart})

	case *noituv1.ClientMessage_KickPlayer:
		s.toRoom(lobbyInput{sess: s, action: lobbyKick, target: playerIDFor(p.KickPlayer.GetPlayerId())})

	case *noituv1.ClientMessage_LeaveRoom:
		s.toRoom(lobbyInput{sess: s, action: lobbyLeave})

	case *noituv1.ClientMessage_SendChat:
		// Its own budget, so a talkative player never runs out of moves. The
		// seat itself is checked by the room, which is the only place that
		// knows whether this connection still holds one.
		if !s.chatLimiter.allow(time.Now()) {
			s.send(errorMsg("too_fast"))
			return nil
		}
		r, id := s.currentRoom()
		if r == nil {
			s.send(errorMsg("not_in_a_room"))
			return nil
		}
		// A dropped line would leave the player watching their own message
		// fail to appear with no reason given.
		if !r.send(chatInput{sess: s, player: id, text: p.SendChat.GetText()}) {
			s.send(errorMsg("busy"))
		}

	case *noituv1.ClientMessage_Ping:
		s.send(pongMsg(p.Ping.GetClientTimeMs(), time.Now().UnixMilli()))
	}
	return nil
}

// roomCreateError names the refusal a room could not be opened for. A full
// server is the player's business — they should wait, not retry at once — and
// anything else is the server's, logged here because the client is only told
// that it failed.
func roomCreateError(sessionID string, err error) *noituv1.ServerMessage {
	if errors.Is(err, errServerFull) {
		return errorMsg("server_full")
	}
	if errors.Is(err, errDraining) {
		// The same key Shutdown sends to everyone already seated: a room
		// refused for this reason will not open a moment later the way a full
		// one might, so the client is told the same thing either way.
		return errorMsg("server_restarting")
	}
	slog.Error("open room", "session", sessionID, "err", err)
	return errorMsg("room_start_failed")
}

// toRoom forwards one lobby action to the room this connection is seated in.
//
// Rate-limited like a submission: every accepted action is broadcast to every
// seat, so an unbounded one lets a player flood the other's outbox until
// their session is closed for falling behind. A dropped action would leave a
// button that did nothing and no reason why, so every failure answers.
func (s *session) toRoom(in lobbyInput) {
	if !s.submitLimiter.allow(time.Now()) {
		s.send(errorMsg("too_fast"))
		return
	}
	r, id := s.currentRoom()
	if r == nil {
		s.send(errorMsg("not_in_a_room"))
		return
	}
	in.player = id
	if !r.send(in) {
		s.send(errorMsg("not_in_a_room"))
	}
}

// handleHello completes the handshake, resuming a prior game when the client
// presents a token that is still live.
func (s *session) handleHello(h *noituv1.Hello) error {
	if v := h.GetProtocolVersion(); v != ProtocolVersion {
		s.send(errorMsg("protocol_version_mismatch"))
		return errors.New("wsapi: protocol version mismatch")
	}

	// The handshake is a one-shot transition. A second Hello would re-register
	// the session and rewrite the nickname of a player already seated in a
	// game, which nothing downstream expects.
	s.mu.Lock()
	repeat := s.greeted
	s.greeted = true
	s.mu.Unlock()
	if repeat {
		s.send(errorMsg("already_greeted"))
		return errors.New("wsapi: repeated hello")
	}

	s.setNickname(sanitizeNickname(h.GetNickname()))
	s.hub.register(s)
	s.send(welcomeMsg(s.id, s.resumeToken, s.nickname()))

	if prior, ok := s.hub.resumable(h.GetResumeToken()); ok && prior != s {
		s.resumeFrom(prior)
	}
	return nil
}

// resumeFrom takes over the seat a previous connection held.
//
// Every failing branch has to say so. A token can outlive its game — the turn
// clock keeps running through the grace window, so a player who dropped on
// their own turn loses before the window closes — and a client that got a
// Welcome and then silence has nothing to render and no reason to stop
// waiting.
func (s *session) resumeFrom(prior *session) {
	metrics.resumesAttempted.Add(1)
	r, id := prior.currentRoom()
	if r == nil {
		s.send(errorMsg("game_already_over"))
		return
	}
	if !r.send(resumeInput{player: id, sess: s, prior: prior}) {
		s.send(errorMsg("game_already_over"))
		return
	}
	// Deliberately no attach and no close here. The room has not decided yet,
	// and a refused resume that had already closed the old connection would end
	// the game it was trying to rejoin.
}

func (s *session) handleSubmit(w *noituv1.SubmitWord) {
	if !s.submitLimiter.allow(time.Now()) {
		s.send(errorMsg("too_fast"))
		return
	}
	r, id := s.currentRoom()
	if r == nil {
		s.send(errorMsg("not_in_a_game"))
		return
	}
	// A dropped submission would otherwise leave the player waiting out the
	// turn clock with no idea their word never arrived.
	if !r.send(submitInput{sess: s, player: id, word: w.GetWord(), turnSeq: w.GetTurnSeq()}) {
		s.send(errorMsg("busy"))
	}
}

// leaveRoom tells the room this connection is gone, so the seat enters its
// grace window rather than the game simply stalling.
func (s *session) leaveRoom() {
	r, id := s.currentRoom()
	if r == nil {
		return
	}
	r.send(disconnectInput{player: id, sess: s})
}

func randomToken() string {
	raw := make([]byte, 16)
	_, _ = rand.Read(raw)
	return base64.RawURLEncoding.EncodeToString(raw)
}