// Package ws implements the websocket endpoint of the game server: // connection upgrade, JOIN handling and the per-connection read loop. package ws import ( "context" "errors" "log/slog" "net/http" "time" "github.com/coder/websocket" "github.com/ruby-sea-battle/game-server/internal/auth" "github.com/ruby-sea-battle/game-server/internal/game" "github.com/ruby-sea-battle/game-server/internal/protocol" ) const ( readLimit = 16 * 1024 // game messages are tiny joinReadTimeout = 10 * time.Second writeTimeout = 5 * time.Second pingInterval = 25 * time.Second pingTimeout = 10 * time.Second ) type writeOp struct { data []byte isClose bool code websocket.StatusCode reason string } // Conn adapts a websocket connection to the game.Sender interface, // delegating all writes to a dedicated goroutine to prevent locking // game mutexes during write timeouts. type Conn struct { ws *websocket.Conn sendCh chan writeOp done chan struct{} // closed to abort writeLoop closed chan struct{} // closed when writeLoop exits } func newConn(ws *websocket.Conn) *Conn { c := &Conn{ ws: ws, sendCh: make(chan writeOp, 64), done: make(chan struct{}), closed: make(chan struct{}), } go c.writeLoop() return c } func (c *Conn) writeLoop() { defer close(c.closed) for { select { case <-c.done: return case op, ok := <-c.sendCh: if !ok { return } if op.isClose { _ = c.ws.Close(op.code, op.reason) return } ctx, cancel := context.WithTimeout(context.Background(), writeTimeout) err := c.ws.Write(ctx, websocket.MessageText, op.data) cancel() if err != nil { _ = c.ws.Close(websocket.StatusGoingAway, "write error") return } } } } func (c *Conn) Send(ctx context.Context, data []byte) error { select { case <-c.done: return errors.New("connection closed") default: } msg := make([]byte, len(data)) copy(msg, data) select { case <-c.done: return errors.New("connection closed") case <-ctx.Done(): return ctx.Err() case c.sendCh <- writeOp{data: msg}: return nil } } func (c *Conn) Close(code int, reason string) error { select { case <-c.closed: return nil default: } op := writeOp{ isClose: true, code: websocket.StatusCode(code), reason: reason, } select { case c.sendCh <- op: select { case <-c.closed: case <-time.After(writeTimeout): _ = c.ws.Close(websocket.StatusCode(code), reason) } default: select { case <-c.done: default: close(c.done) } _ = c.ws.Close(websocket.StatusCode(code), reason) } return nil } // Handler is the http.Handler of the websocket endpoint. type Handler struct { Manager *game.Manager JWTSecret []byte PublicAddr string // must match the "game_server" claim strictly acceptOpts *websocket.AcceptOptions log *slog.Logger } func NewHandler(manager *game.Manager, jwtSecret []byte, publicAddr string, allowedOrigins []string, log *slog.Logger) *Handler { opts := &websocket.AcceptOptions{} if len(allowedOrigins) == 1 && allowedOrigins[0] == "*" { opts.InsecureSkipVerify = true } else { opts.OriginPatterns = allowedOrigins } return &Handler{ Manager: manager, JWTSecret: jwtSecret, PublicAddr: publicAddr, acceptOpts: opts, log: log, } } func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { wsc, err := websocket.Accept(w, r, h.acceptOpts) if err != nil { h.log.Debug("websocket accept failed", "err", err) return } wsc.SetReadLimit(readLimit) ctx, cancel := context.WithCancel(r.Context()) defer cancel() defer wsc.Close(websocket.StatusInternalError, "") // The first message on a fresh connection must be JOIN. readCtx, cancelRead := context.WithTimeout(ctx, joinReadTimeout) _, data, err := wsc.Read(readCtx) cancelRead() if err != nil { wsc.Close(protocol.CloseCodeExpectedJoin, "join timeout") return } opcode, args, err := protocol.ParseMessage(data) if err != nil { wsc.Close(protocol.CloseCodeBadMessage, "malformed message") return } if opcode != protocol.OpJoin { wsc.Close(protocol.CloseCodeExpectedJoin, "expected JOIN") return } gameToken, userToken, err := protocol.ParseJoin(args) if err != nil { wsc.Close(protocol.CloseCodeBadMessage, "malformed JOIN") return } // If player is reconnecting with the same game token used when creating the match, // skip token validation completely. var gc *auth.GameClaims unverified, _ := auth.ParseGameTokenUnverified(gameToken) if unverified != nil { if existing := h.Manager.GetMatch(unverified.MatchID); existing != nil && existing.GameToken() == gameToken { gc = &auth.GameClaims{ MatchID: existing.ID, Player1ID: existing.PlayerIDs()[0], Player2ID: existing.PlayerIDs()[1], RawToken: gameToken, } } } if gc == nil { // Validate the game token and make sure the token was issued for this // particular server instance (strict comparison with the public address). var err error gc, err = auth.ParseGameToken(h.JWTSecret, gameToken) if err != nil { h.log.Debug("bad game token", "err", err) wsc.Close(protocol.CloseCodeBadGameToken, "invalid game token") return } if gc.GameServer != h.PublicAddr { h.log.Debug("game token issued for another server", "want", h.PublicAddr, "got", gc.GameServer) wsc.Close(protocol.CloseCodeBadGameToken, "game server mismatch") return } if gc.Player1ID <= 0 || gc.Player2ID <= 0 || gc.Player1ID == gc.Player2ID { wsc.Close(protocol.CloseCodeBadGameToken, "invalid match") return } } // Validate the user token and make sure the user is a participant. uc, err := auth.ParseUserToken(h.JWTSecret, userToken) if err != nil { h.log.Debug("bad user token", "err", err) wsc.Close(protocol.CloseCodeBadUserToken, "invalid user token") return } if uc.UserID <= 0 || (uc.UserID != gc.Player1ID && uc.UserID != gc.Player2ID) { wsc.Close(protocol.CloseCodeBadUserToken, "not a participant of this match") return } conn := newConn(wsc) match, slotIdx, err := h.Manager.Join(ctx, conn, gc, uc.UserID) if err != nil { // ErrMatchFinished: the END message has already been replayed and // the connection closed by the manager. if err != game.ErrMatchFinished { h.log.Debug("join failed", "err", err) } return } h.log.Info("player joined", "match", match.ID, "user", uc.UserID, "slot", slotIdx) go h.pingLoop(ctx, wsc) // Read loop: process messages until the connection dies. for { _, data, err := wsc.Read(ctx) if err != nil { break } opcode, args, err := protocol.ParseMessage(data) if err != nil { h.log.Debug("malformed message", "match", match.ID, "user", uc.UserID, "err", err) continue } switch opcode { case protocol.OpJoin: // Duplicate JOIN on an established connection is ignored. case protocol.OpAck: seq, err := protocol.ParseAck(args) if err != nil { h.log.Debug("malformed ACK", "match", match.ID, "user", uc.UserID, "err", err) continue } if err := h.Manager.Ack(match, slotIdx, seq); err != nil { h.log.Debug("invalid ACK seq", "match", match.ID, "user", uc.UserID, "seq", seq, "err", err) continue } case protocol.OpPlaceShips: ships, err := protocol.ParsePlaceShips(args) if err != nil { h.log.Debug("malformed PlaceShips", "match", match.ID, "user", uc.UserID, "err", err) continue } if err := protocol.ValidateShips(ships); err != nil { h.log.Debug("invalid ship placement", "match", match.ID, "user", uc.UserID, "err", err) continue } if err := h.Manager.PlaceShips(ctx, match, slotIdx, ships); err != nil { h.log.Debug("place ships rejected", "match", match.ID, "user", uc.UserID, "err", err) continue } case protocol.OpFire: x, y, err := protocol.ParseFire(args) if err != nil { h.log.Debug("malformed Fire", "match", match.ID, "user", uc.UserID, "err", err) continue } if _, err := h.Manager.Fire(ctx, match, slotIdx, x, y); err != nil { h.log.Debug("fire rejected or ignored", "match", match.ID, "user", uc.UserID, "err", err) continue } default: // Unknown opcodes are ignored for forward compatibility. h.log.Debug("unknown opcode", "match", match.ID, "user", uc.UserID, "opcode", opcode) } } h.Manager.Leave(conn, match, slotIdx) h.log.Info("player disconnected", "match", match.ID, "user", uc.UserID, "slot", slotIdx) } // pingLoop keeps the connection alive and detects dead peers. When the ping // fails the connection is closed, which terminates the read loop. func (h *Handler) pingLoop(ctx context.Context, wsc *websocket.Conn) { ticker := time.NewTicker(pingInterval) defer ticker.Stop() for { select { case <-ctx.Done(): return case <-ticker.C: pingCtx, cancel := context.WithTimeout(ctx, pingTimeout) err := wsc.Ping(pingCtx) cancel() if err != nil { wsc.Close(websocket.StatusGoingAway, "ping failed") return } } } }