334 lines
8.8 KiB
Go
334 lines
8.8 KiB
Go
// 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
|
|
}
|
|
}
|
|
}
|
|
}
|