diff --git a/game-server/README.md b/game-server/README.md new file mode 100644 index 0000000..21c6d96 --- /dev/null +++ b/game-server/README.md @@ -0,0 +1,170 @@ +# game-server + +WebSocket-сервер партий морского боя. Принимает подключения игроков, +проверяет JWT-токены, выданные main-server, создаёт и хранит в памяти +состояние партий. + +## Сборка и запуск + +```sh +go build ./cmd/server +JWT_SECRET=... PUBLIC_ADDR=ws://localhost:8081/ws ./server +``` + +Конфигурация через переменные окружения: + +| Переменная | По умолчанию | Описание | +|--------------------|--------------|----------| +| `LISTEN_ADDR` | `:8081` | Адрес, на котором открывается HTTP/WebSocket-сервер (может быть за reverse proxy). | +| `PUBLIC_ADDR` | — (обяз.) | Публичный адрес сервера, например `ws://localhost:8081/ws`. Используется **только** для идентификации: сравнивается строго (побайтово) с полем `game_server` игрового токена. Main-server должен подписывать в токены ровно эту строку. | +| `JWT_SECRET` | — (обяз.) | Общий с main-server HS256-секрет (`secret_key_base`). Им подписаны и игровой, и пользовательский токены. | +| `JOIN_TIMEOUT` | `2m` | Сколько первый игрок ждёт второго, прежде чем партия будет отменена. Go-duration (`90s`, `2m`) или число секунд. | +| `ALLOWED_ORIGINS` | `*` | Список Origin-паттернов через запятую для проверки заголовка `Origin`; `*` отключает проверку. | + +Эндпоинты: `GET /ws` — WebSocket, `GET /healthz` — проверка живости. + +## Протокол + +Все сообщения — JSON-массивы, первый элемент — числовой опкод. + +| Опкод | Направление | Название | Формат | +|-------|-------------|----------|--------| +| 1 | C -> S | JOIN | `[1, "", ""]` | +| 2 | S -> C | END | `[2, , , ""]` | +| 3 | S -> C | STATE | `[3, , ]` | +| 4 | C <-> S | ACK | `[4, ]` | + +### JOIN (1) + +Первое и единственное сообщение, которое сервер принимает на свежем +соединении (в течение 10 секунд после апгрейда). Параметры: + +1. `game_token` — JWT партии, полезная нагрузка: + ```json + { + "match_id": "uuid партии", + "player1_id": 1, + "player2_id": 2, + "game_server": "ws://...", + "exp": 1234567890 + } + ``` +2. `user_token` — личный JWT игрока: `{"user_id": 1, "exp": ...}`. + +Проверки при JOIN: + +- подпись (HS256) и срок действия обоих токенов; +- `game_server` строго равен `PUBLIC_ADDR` этого сервера; +- `player1_id != player2_id`; +- `user_id` из пользовательского токена совпадает с `player1_id` или + `player2_id` игрового токена; +- партия не находится в реестре завершённых (см. ниже). + +При неудаче соединение закрывается с кодом из диапазона 4xxx (см. ниже). + +### END (2) + +Партия завершена. Формат: `[2, , , ""]`. + +Коды завершения: + +| Код | Причина | +|-----|---------| +| -1 | `join timeout exceeded` — второй игрок не присоединился вовремя | + +### STATE (3) + +Переход игры в новое состояние. Код состояния всегда отсылается **с точки +зрения получателя** — клиент не знает, «игрок 1» он или «игрок 2»: + +| Код | Состояние | +|-----|-----------| +| 1 | оба игрока расставляют корабли | +| 2 | соперник готов | +| 3 | ожидайте готовности соперника | +| 4 | ваш ход | +| 5 | ожидайте ход соперника | + +Сейчас рассылается только состояние `1` — при старте партии (когда оба +игрока впервые присоединились). + +### ACK (4) + +Каждое изменяющее состояние сообщение S->C (END, STATE) несёт `seq` — +монотонно возрастающий номер в рамках пары (партия, игрок). Клиент +подтверждает доставку сообщением `[4, ]`; ACK кумулятивен +(подтверждает все сообщения с `seq` <= указанного). + +Неподтверждённые сообщения хранятся на сервере в очереди игрока и +переотправляются при его переподключении. Если очередь пуста, но игра уже +идёт, при переподключении текущее состояние отправляется заново (с новым +`seq`), чтобы клиент мог восстановить UI. + +Направление S->C зарезервировано для подтверждения будущих сообщений +клиента (у которых пока нет номеров). + +## Коды закрытия соединения + +| Код | Значение | +|------|----------| +| 1000 | нормальное завершение (партия окончена, END уже отправлен) | +| 4000 | соединение заменено новым соединением того же игрока | +| 4001 | первое сообщение не JOIN | +| 4002 | сообщение не соответствует формату | +| 4003 | недействительный игровой токен / чужой `game_server` | +| 4004 | недействительный пользовательский токен / игрок не участник партии | + +## Обработка краевых случаев + +- **Переподключение.** Игрок идентифицируется токенами, а не соединением. + Новое JOIN от того же игрока закрывает старое соединение (код 4000) и + занимает его слот; очередь неподтверждённых сообщений переживает + разрыв и переотправляется. Одновременно существует не более одного + соединения на игрока. +- **Повторный JOIN после завершения партии.** Завершённые партии хранятся + в реестре (`match_id` -> END-сообщения обоих игроков) до истечения срока + действия игрового токена. Поздний JOIN получает в ответ сохранённое END + и закрытие соединения — фантомная партия не создаётся, и будущий репорт + итогов на main-server не задвоится. Реестр очищается лениво: при каждом + обращении к нему стираются записи с истёкшим токеном, фоновых задач нет. +- **Дедлайн присоединения** — минимум из `JOIN_TIMEOUT` и `exp` игрового + токена: после истечения токена второй игрок всё равно не сможет войти. +- **JOIN с токеном, указывающим на другой сервер** (неверный + `game_server`) отклоняется с кодом 4003. +- **`player1_id == player2_id`** отклоняется (код 4003) — иначе один игрок + мог бы занять оба слота. +- Неизвестные опкоды после JOIN игнорируются (совместимость с будущими + версиями протокола); мусорные/не-JSON сообщения логируются и пропускаются. +- Keepalive: сервер пингует соединение каждые 25 с; размер входящего + сообщения ограничен 16 КБ; все записи в сокет ограничены таймаутом 5 с. + +## Структура кода + +``` +cmd/server/main.go — точка входа, graceful shutdown +internal/config/ — конфигурация из env +internal/protocol/ — опкоды, коды состояний/завершения, сборка/разбор сообщений +internal/auth/ — разбор и валидация JWT (HS256, exp обязателен) +internal/game/ — менеджер партий: слоты игроков, очереди неподтверждённых + сообщений, таймер присоединения, реестр завершённых партий +internal/ws/ — websocket-эндпоинт: апгрейд, JOIN, read-цикл, ping +``` + +## Тесты + +```sh +go test -race ./... +``` + +Покрыты: валидация токенов, разбор/сборка сообщений, сценарии менеджера +(ожидание, старт, таймаут, повторный вход, замена соединения, ACK) и +сквозные интеграционные тесты через настоящий websocket. + +## TODO (следующие этапы) + +- игровая логика (расстановка кораблей, ходы) и остальные состояния; +- завершение партии по итогам игры с персонифицированными кодами + победы/поражения; +- репорт итогов партии на main-server (реестр завершённых партий уже + гарантирует, что партия завершается ровно один раз); +- очистка состояния начавшихся партий после их завершения. diff --git a/game-server/cmd/server/main.go b/game-server/cmd/server/main.go new file mode 100644 index 0000000..140d4a3 --- /dev/null +++ b/game-server/cmd/server/main.go @@ -0,0 +1,63 @@ +package main + +import ( + "context" + "errors" + "log/slog" + "net/http" + "os" + "os/signal" + "syscall" + "time" + + "github.com/ruby-sea-battle/game-server/internal/config" + "github.com/ruby-sea-battle/game-server/internal/game" + "github.com/ruby-sea-battle/game-server/internal/ws" +) + +func main() { + log := slog.New(slog.NewTextHandler(os.Stdout, nil)) + + cfg, err := config.Load() + if err != nil { + log.Error("configuration error", "err", err) + os.Exit(1) + } + + manager := game.NewManager(cfg.JoinTimeout) + handler := ws.NewHandler(manager, cfg.JWTSecret, cfg.PublicAddr, cfg.AllowedOrigins, log) + + mux := http.NewServeMux() + mux.Handle("/ws", handler) + mux.HandleFunc("/healthz", func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte("ok")) + }) + + srv := &http.Server{ + Addr: cfg.ListenAddr, + Handler: mux, + ReadHeaderTimeout: 10 * time.Second, + } + + ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) + defer stop() + + go func() { + log.Info("game server listening", + "listen_addr", cfg.ListenAddr, + "public_addr", cfg.PublicAddr, + "join_timeout", cfg.JoinTimeout, + ) + if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) { + log.Error("server failed", "err", err) + os.Exit(1) + } + }() + + <-ctx.Done() + log.Info("shutting down") + shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + _ = srv.Shutdown(shutdownCtx) +} diff --git a/game-server/go.mod b/game-server/go.mod new file mode 100644 index 0000000..6b8b9bf --- /dev/null +++ b/game-server/go.mod @@ -0,0 +1,8 @@ +module github.com/ruby-sea-battle/game-server + +go 1.27.1 + +require ( + github.com/coder/websocket v1.8.15 + github.com/golang-jwt/jwt/v5 v5.3.1 +) diff --git a/game-server/go.sum b/game-server/go.sum new file mode 100644 index 0000000..c6bffb7 --- /dev/null +++ b/game-server/go.sum @@ -0,0 +1,4 @@ +github.com/coder/websocket v1.8.15 h1:6B2JPeOGlpff2Uz6vOEH1Vzpi0iUz20A+lPVhPHtNUA= +github.com/coder/websocket v1.8.15/go.mod h1:NX3SzP+inril6yawo5CQXx8+fk145lPDC6pumgx0mVg= +github.com/golang-jwt/jwt/v5 v5.3.1 h1:kYf81DTWFe7t+1VvL7eS+jKFVWaUnK9cB1qbwn63YCY= +github.com/golang-jwt/jwt/v5 v5.3.1/go.mod h1:fxCRLWMO43lRc8nhHWY6LGqRcf+1gQWArsqaEUEa5bE= diff --git a/game-server/internal/auth/jwt.go b/game-server/internal/auth/jwt.go new file mode 100644 index 0000000..3efc3f7 --- /dev/null +++ b/game-server/internal/auth/jwt.go @@ -0,0 +1,66 @@ +// Package auth implements parsing and validation of the JWT tokens used by +// the game server. Both the game token and the user token are HS256-signed +// with the same shared secret (the main server's secret_key_base). +package auth + +import ( + "fmt" + + "github.com/golang-jwt/jwt/v5" +) + +// GameClaims is the payload of the game token issued by the main server +// when a match is created. +type GameClaims struct { + MatchID string `json:"match_id"` + Player1ID int64 `json:"player1_id"` + Player2ID int64 `json:"player2_id"` + GameServer string `json:"game_server"` + jwt.RegisteredClaims +} + +// UserClaims is the payload of the personal token of a player. +type UserClaims struct { + UserID int64 `json:"user_id"` + jwt.RegisteredClaims +} + +func keyFunc(secret []byte) jwt.Keyfunc { + return func(t *jwt.Token) (any, error) { + if t.Method.Alg() != jwt.SigningMethodHS256.Alg() { + return nil, fmt.Errorf("unexpected signing method: %s", t.Method.Alg()) + } + return secret, nil + } +} + +// ParseGameToken verifies the signature and the expiration time of a game +// token and returns its claims. +func ParseGameToken(secret []byte, tokenString string) (*GameClaims, error) { + claims := &GameClaims{} + _, err := jwt.ParseWithClaims(tokenString, claims, keyFunc(secret), + jwt.WithValidMethods([]string{jwt.SigningMethodHS256.Alg()}), + jwt.WithExpirationRequired(), + ) + if err != nil { + return nil, fmt.Errorf("invalid game token: %w", err) + } + if claims.MatchID == "" { + return nil, fmt.Errorf("invalid game token: empty match_id") + } + return claims, nil +} + +// ParseUserToken verifies the signature and the expiration time of a user +// token and returns its claims. +func ParseUserToken(secret []byte, tokenString string) (*UserClaims, error) { + claims := &UserClaims{} + _, err := jwt.ParseWithClaims(tokenString, claims, keyFunc(secret), + jwt.WithValidMethods([]string{jwt.SigningMethodHS256.Alg()}), + jwt.WithExpirationRequired(), + ) + if err != nil { + return nil, fmt.Errorf("invalid user token: %w", err) + } + return claims, nil +} diff --git a/game-server/internal/auth/jwt_test.go b/game-server/internal/auth/jwt_test.go new file mode 100644 index 0000000..86f439d --- /dev/null +++ b/game-server/internal/auth/jwt_test.go @@ -0,0 +1,78 @@ +package auth + +import ( + "testing" + "time" + + "github.com/golang-jwt/jwt/v5" +) + +var secret = []byte("test-secret") + +func sign(t *testing.T, claims jwt.Claims, key []byte) string { + t.Helper() + token, err := jwt.NewWithClaims(jwt.SigningMethodHS256, claims).SignedString(key) + if err != nil { + t.Fatalf("sign: %v", err) + } + return token +} + +func gameToken(t *testing.T, key []byte, exp time.Time) string { + t.Helper() + return sign(t, &GameClaims{ + MatchID: "m1", + Player1ID: 1, + Player2ID: 2, + GameServer: "ws://game.example.com/ws", + RegisteredClaims: jwt.RegisteredClaims{ + ExpiresAt: jwt.NewNumericDate(exp), + }, + }, key) +} + +func TestParseGameToken(t *testing.T) { + token := gameToken(t, secret, time.Now().Add(time.Hour)) + claims, err := ParseGameToken(secret, token) + if err != nil { + t.Fatalf("parse: %v", err) + } + if claims.MatchID != "m1" || claims.Player1ID != 1 || claims.Player2ID != 2 { + t.Fatalf("unexpected claims: %+v", claims) + } +} + +func TestParseGameTokenExpired(t *testing.T) { + token := gameToken(t, secret, time.Now().Add(-time.Hour)) + if _, err := ParseGameToken(secret, token); err == nil { + t.Fatal("expected error for expired token") + } +} + +func TestParseGameTokenWrongSecret(t *testing.T) { + token := gameToken(t, []byte("other-secret"), time.Now().Add(time.Hour)) + if _, err := ParseGameToken(secret, token); err == nil { + t.Fatal("expected error for wrong secret") + } +} + +func TestParseGameTokenMissingExp(t *testing.T) { + token := sign(t, &GameClaims{MatchID: "m1", Player1ID: 1, Player2ID: 2}, secret) + if _, err := ParseGameToken(secret, token); err == nil { + t.Fatal("expected error for token without exp") + } +} + +func TestParseUserToken(t *testing.T) { + token := sign(t, &UserClaims{ + UserID: 42, + RegisteredClaims: jwt.RegisteredClaims{ExpiresAt: jwt.NewNumericDate(time.Now().Add(time.Hour))}, + }, secret) + claims, err := ParseUserToken(secret, token) + if err != nil { + t.Fatalf("parse: %v", err) + } + if claims.UserID != 42 { + t.Fatalf("user_id = %d, want 42", claims.UserID) + } +} diff --git a/game-server/internal/config/config.go b/game-server/internal/config/config.go new file mode 100644 index 0000000..b6c8c85 --- /dev/null +++ b/game-server/internal/config/config.go @@ -0,0 +1,82 @@ +// Package config loads the game server configuration from environment +// variables. +package config + +import ( + "fmt" + "os" + "strconv" + "strings" + "time" +) + +type Config struct { + // ListenAddr is the address the websocket HTTP server binds to + // (e.g. ":8081"). The server may sit behind a reverse proxy. + ListenAddr string + // PublicAddr is the public endpoint address of this server + // (e.g. "ws://example.com:8081/ws"). It is used only to identify this + // server: the "game_server" claim of a game token must match it + // strictly. The main server must be configured to put exactly this + // string into the tokens it issues. + PublicAddr string + // JWTSecret is the HS256 secret shared with the main server; both the + // game token and the user token are verified with it. + JWTSecret []byte + // JoinTimeout is how long the first player waits for the second one + // before the match is cancelled. + JoinTimeout time.Duration + // AllowedOrigins restricts the Origin header of websocket upgrade + // requests; ["*"] disables the check. + AllowedOrigins []string +} + +func Load() (Config, error) { + var cfg Config + + cfg.ListenAddr = getenv("LISTEN_ADDR", ":8081") + + cfg.PublicAddr = os.Getenv("PUBLIC_ADDR") + if cfg.PublicAddr == "" { + return cfg, fmt.Errorf("PUBLIC_ADDR is required (public endpoint address, e.g. ws://example.com:8081/ws)") + } + + secret := os.Getenv("JWT_SECRET") + if secret == "" { + return cfg, fmt.Errorf("JWT_SECRET is required (HS256 secret shared with the main server)") + } + cfg.JWTSecret = []byte(secret) + + joinTimeout, err := parseDuration(getenv("JOIN_TIMEOUT", "2m")) + if err != nil { + return cfg, fmt.Errorf("invalid JOIN_TIMEOUT: %w", err) + } + cfg.JoinTimeout = joinTimeout + + origins := getenv("ALLOWED_ORIGINS", "*") + cfg.AllowedOrigins = strings.Split(origins, ",") + for i := range cfg.AllowedOrigins { + cfg.AllowedOrigins[i] = strings.TrimSpace(cfg.AllowedOrigins[i]) + } + + return cfg, nil +} + +func getenv(key, fallback string) string { + if v := os.Getenv(key); v != "" { + return v + } + return fallback +} + +// parseDuration accepts either a Go duration string ("90s", "2m") or a plain +// number of seconds ("120"). +func parseDuration(s string) (time.Duration, error) { + if d, err := time.ParseDuration(s); err == nil { + return d, nil + } + if secs, err := strconv.Atoi(s); err == nil { + return time.Duration(secs) * time.Second, nil + } + return 0, fmt.Errorf("not a duration: %q", s) +} diff --git a/game-server/internal/game/manager.go b/game-server/internal/game/manager.go new file mode 100644 index 0000000..03f192a --- /dev/null +++ b/game-server/internal/game/manager.go @@ -0,0 +1,308 @@ +// Package game keeps the in-memory state of all matches running on this +// game server: player connections, per-player unacknowledged message queues +// and a registry of recently finished matches. +package game + +import ( + "context" + "errors" + "sync" + "time" + + "github.com/ruby-sea-battle/game-server/internal/auth" + "github.com/ruby-sea-battle/game-server/internal/protocol" +) + +// ErrMatchFinished is returned by Join when the player tries to join a match +// that has already ended. The stored END message (if any) is replayed to the +// connection before Join returns. +var ErrMatchFinished = errors.New("match already finished") + +// Sender abstracts a player connection so the game logic stays testable and +// independent of the websocket implementation. +type Sender interface { + // Send writes one message to the connection. Implementations must be + // safe for concurrent use and must bound the write with a timeout. + Send(ctx context.Context, data []byte) error + // Close closes the connection with the given websocket close code. + Close(code int, reason string) error +} + +// queuedMessage is a state-changing message that has been sent to a player +// but not yet acknowledged with an ACK. +type queuedMessage struct { + seq int64 + data []byte +} + +// PlayerSlot is the state of one of the two players of a match. +type PlayerSlot struct { + UserID int64 + + conn Sender // nil while the player is disconnected + seq int64 // last issued sequence number + unacked []queuedMessage + lastState int // last state sent to this player, 0 if none + everConnected bool // true after the first successful join +} + +// Match is the state of a single game of sea battle. +type Match struct { + ID string + tokenExp time.Time // game token expiration; bounds the join deadline + playerIDs [2]int64 // slot 0 = player1, slot 1 = player2 (internal only) + slots [2]*PlayerSlot + + mu sync.Mutex + started bool // both players joined at least once + ended bool + state int // current global game state + joinTimer *time.Timer +} + +func (m *Match) slotIndex(userID int64) int { + switch userID { + case m.playerIDs[0]: + return 0 + case m.playerIDs[1]: + return 1 + default: + return -1 + } +} + +// finishedMatch remembers how a match ended so that late join attempts with +// a still-valid game token get the same END message instead of creating a +// phantom match (and, later, a duplicate result report to the main server). +type finishedMatch struct { + ends map[int64][]byte // userID -> END message sent to that player + expiresAt time.Time // game token expiration time +} + +// Manager owns all matches of this server instance. +type Manager struct { + joinTimeout time.Duration + + mu sync.Mutex + matches map[string]*Match + finished map[string]*finishedMatch +} + +func NewManager(joinTimeout time.Duration) *Manager { + return &Manager{ + joinTimeout: joinTimeout, + matches: make(map[string]*Match), + finished: make(map[string]*finishedMatch), + } +} + +// Join attaches conn to the match identified by the game token claims. +// userID must already be validated against the claims by the caller. +// +// If the player was already connected, the old connection is closed and +// replaced. Unacknowledged messages are replayed to the new connection; if +// there is nothing to replay but the game is already running, the current +// state is re-sent so the client can restore its UI. +// +// Returns the match and the player's slot index (0 or 1). +func (m *Manager) Join(ctx context.Context, conn Sender, gc *auth.GameClaims, userID int64) (*Match, int, error) { + now := time.Now() + + m.mu.Lock() + m.purgeFinishedLocked(now) + if fin, ok := m.finished[gc.MatchID]; ok { + m.mu.Unlock() + if data, ok := fin.ends[userID]; ok { + _ = conn.Send(ctx, data) + } + _ = conn.Close(protocol.CloseCodeNormal, "match already finished") + return nil, -1, ErrMatchFinished + } + match, ok := m.matches[gc.MatchID] + if !ok { + match = &Match{ + ID: gc.MatchID, + tokenExp: gc.ExpiresAt.Time, + playerIDs: [2]int64{gc.Player1ID, gc.Player2ID}, + slots: [2]*PlayerSlot{ + {UserID: gc.Player1ID}, + {UserID: gc.Player2ID}, + }, + } + // The second player must join before the join timeout expires, but + // never later than the game token itself expires — after that a + // join is impossible anyway. + deadline := now.Add(m.joinTimeout) + if match.tokenExp.Before(deadline) { + deadline = match.tokenExp + } + match.joinTimer = time.AfterFunc(time.Until(deadline), func() { + m.onJoinTimeout(match.ID) + }) + m.matches[gc.MatchID] = match + } + m.mu.Unlock() + + match.mu.Lock() + defer match.mu.Unlock() + + if match.ended { + // Lost race with the join timeout; the match is being torn down. + _ = conn.Close(protocol.CloseCodeNormal, "match already finished") + return nil, -1, ErrMatchFinished + } + + idx := match.slotIndex(userID) + if idx < 0 { + // Should be impossible: the caller validated membership. + _ = conn.Close(protocol.CloseCodeBadUserToken, "not a participant of this match") + return nil, -1, errors.New("user is not a participant") + } + slot := match.slots[idx] + + if slot.conn != nil && slot.conn != conn { + // Only one live connection per player: replace the old one. + _ = slot.conn.Close(protocol.CloseCodeReplaced, "connection replaced by a new one") + } + slot.conn = conn + slot.everConnected = true + + switch { + case !match.started && match.slots[0].everConnected && match.slots[1].everConnected: + // Both players joined for the first time: the game begins. + match.started = true + if match.joinTimer != nil { + match.joinTimer.Stop() + } + match.state = protocol.StatePlacingShips + m.setStateLocked(ctx, match, match.state) + case match.started: + // Reconnect of a player while the game is running. + m.replayLocked(ctx, slot) + } + // Otherwise: first player waiting for the opponent, nothing to send. + + return match, idx, nil +} + +// Leave detaches conn from the match if it is still the current connection +// of the player's slot. Messages queued for the player survive the +// disconnect and will be replayed on rejoin. +func (m *Manager) Leave(conn Sender, match *Match, idx int) { + match.mu.Lock() + defer match.mu.Unlock() + if slot := match.slots[idx]; slot.conn == conn { + slot.conn = nil + } +} + +// Ack confirms delivery of all messages with seq <= the given sequence +// number for the given player. +func (m *Manager) Ack(match *Match, idx int, seq int64) { + match.mu.Lock() + defer match.mu.Unlock() + slot := match.slots[idx] + kept := slot.unacked[:0] + for _, qm := range slot.unacked { + if qm.seq > seq { + kept = append(kept, qm) + } + } + slot.unacked = kept +} + +// setStateLocked moves the match into a new state and notifies both players. +// match.mu must be held. +func (m *Manager) setStateLocked(ctx context.Context, match *Match, state int) { + for _, slot := range match.slots { + m.queueStateLocked(ctx, slot, state) + } +} + +// queueStateLocked appends a STATE message to the player's unacknowledged +// queue and sends it right away if the player is connected. +// match.mu must be held. +func (m *Manager) queueStateLocked(ctx context.Context, slot *PlayerSlot, state int) { + slot.seq++ + data := protocol.BuildState(slot.seq, state) + slot.unacked = append(slot.unacked, queuedMessage{seq: slot.seq, data: data}) + slot.lastState = state + if slot.conn != nil { + _ = slot.conn.Send(ctx, data) + } +} + +// replayLocked re-sends everything the connected player has not +// acknowledged yet; if there is nothing pending, the current state is sent +// anew so a reconnected client can catch up. +// match.mu must be held. +func (m *Manager) replayLocked(ctx context.Context, slot *PlayerSlot) { + if len(slot.unacked) == 0 { + if slot.lastState != 0 { + m.queueStateLocked(ctx, slot, slot.lastState) + } + return + } + for _, qm := range slot.unacked { + _ = slot.conn.Send(ctx, qm.data) + } +} + +// onJoinTimeout finishes a match whose second player never joined. +func (m *Manager) onJoinTimeout(matchID string) { + m.mu.Lock() + match, ok := m.matches[matchID] + if !ok { + m.mu.Unlock() + return + } + match.mu.Lock() + if match.started || match.ended { + match.mu.Unlock() + m.mu.Unlock() + return + } + match.ended = true + delete(m.matches, matchID) + + ends := make(map[int64][]byte, 2) + type notification struct { + conn Sender + data []byte + } + var notify []notification + for _, slot := range match.slots { + slot.seq++ + data := protocol.BuildEnd(slot.seq, protocol.EndCodeJoinTimeout, "join timeout exceeded") + slot.unacked = append(slot.unacked, queuedMessage{seq: slot.seq, data: data}) + ends[slot.UserID] = data + if slot.conn != nil { + notify = append(notify, notification{conn: slot.conn, data: data}) + slot.conn = nil + } + } + match.mu.Unlock() + + // Remember the outcome until the game token expires so that late join + // attempts are answered with the same END message instead of creating a + // phantom match. + m.finished[matchID] = &finishedMatch{ends: ends, expiresAt: match.tokenExp} + m.purgeFinishedLocked(time.Now()) + m.mu.Unlock() + + // Network I/O happens outside of any lock. + for _, n := range notify { + _ = n.conn.Send(context.Background(), n.data) + _ = n.conn.Close(protocol.CloseCodeNormal, "match ended") + } +} + +// purgeFinishedLocked drops registry entries whose game token has expired. +// Called lazily whenever the registry is touched. m.mu must be held. +func (m *Manager) purgeFinishedLocked(now time.Time) { + for id, fin := range m.finished { + if !now.Before(fin.expiresAt) { + delete(m.finished, id) + } + } +} diff --git a/game-server/internal/game/manager_test.go b/game-server/internal/game/manager_test.go new file mode 100644 index 0000000..4a553e6 --- /dev/null +++ b/game-server/internal/game/manager_test.go @@ -0,0 +1,265 @@ +package game + +import ( + "context" + "encoding/json" + "sync" + "testing" + "time" + + "github.com/golang-jwt/jwt/v5" + + "github.com/ruby-sea-battle/game-server/internal/auth" + "github.com/ruby-sea-battle/game-server/internal/protocol" +) + +type fakeConn struct { + mu sync.Mutex + sent [][]byte + closed bool + closeCode int +} + +func (f *fakeConn) Send(_ context.Context, data []byte) error { + f.mu.Lock() + defer f.mu.Unlock() + cp := make([]byte, len(data)) + copy(cp, data) + f.sent = append(f.sent, cp) + return nil +} + +func (f *fakeConn) Close(code int, _ string) error { + f.mu.Lock() + defer f.mu.Unlock() + f.closed = true + f.closeCode = code + return nil +} + +func (f *fakeConn) messages() [][]byte { + f.mu.Lock() + defer f.mu.Unlock() + out := make([][]byte, len(f.sent)) + copy(out, f.sent) + return out +} + +func (f *fakeConn) isClosed() (bool, int) { + f.mu.Lock() + defer f.mu.Unlock() + return f.closed, f.closeCode +} + +func claims(exp time.Time) *auth.GameClaims { + return &auth.GameClaims{ + MatchID: "match-1", + Player1ID: 10, + Player2ID: 20, + GameServer: "ws://game.example.com/ws", + RegisteredClaims: jwt.RegisteredClaims{ + ExpiresAt: jwt.NewNumericDate(exp), + }, + } +} + +func parseMsg(t *testing.T, data []byte) []any { + t.Helper() + var v []any + if err := json.Unmarshal(data, &v); err != nil { + t.Fatalf("unmarshal %s: %v", data, err) + } + return v +} + +func TestJoinFirstPlayerWaits(t *testing.T) { + m := NewManager(time.Minute) + c1 := &fakeConn{} + gc := claims(time.Now().Add(time.Hour)) + + _, idx, err := m.Join(context.Background(), c1, gc, gc.Player1ID) + if err != nil { + t.Fatalf("join: %v", err) + } + if idx != 0 { + t.Fatalf("slot = %d, want 0", idx) + } + if got := len(c1.messages()); got != 0 { + t.Fatalf("first player got %d messages, want 0", got) + } +} + +func TestSecondJoinStartsMatch(t *testing.T) { + m := NewManager(time.Minute) + gc := claims(time.Now().Add(time.Hour)) + c1, c2 := &fakeConn{}, &fakeConn{} + + if _, _, err := m.Join(context.Background(), c1, gc, gc.Player1ID); err != nil { + t.Fatalf("join p1: %v", err) + } + if _, idx, err := m.Join(context.Background(), c2, gc, gc.Player2ID); err != nil { + t.Fatalf("join p2: %v", err) + } else if idx != 1 { + t.Fatalf("slot = %d, want 1", idx) + } + + for i, c := range []*fakeConn{c1, c2} { + msgs := c.messages() + if len(msgs) != 1 { + t.Fatalf("player %d got %d messages, want 1", i, len(msgs)) + } + v := parseMsg(t, msgs[0]) + // [3, seq, StatePlacingShips] + if v[0] != float64(protocol.OpState) || v[1] != float64(1) || v[2] != float64(protocol.StatePlacingShips) { + t.Fatalf("player %d got unexpected message: %v", i, v) + } + } +} + +func TestJoinTimeoutEndsMatch(t *testing.T) { + m := NewManager(50 * time.Millisecond) + gc := claims(time.Now().Add(time.Hour)) + c1 := &fakeConn{} + + if _, _, err := m.Join(context.Background(), c1, gc, gc.Player1ID); err != nil { + t.Fatalf("join: %v", err) + } + + deadline := time.Now().Add(2 * time.Second) + for { + if closed, _ := c1.isClosed(); closed { + break + } + if time.Now().After(deadline) { + t.Fatal("timed out waiting for the match to end") + } + time.Sleep(10 * time.Millisecond) + } + + msgs := c1.messages() + if len(msgs) != 1 { + t.Fatalf("got %d messages, want 1", len(msgs)) + } + v := parseMsg(t, msgs[0]) + // [2, seq, -1, "join timeout exceeded"] + if v[0] != float64(protocol.OpEnd) || v[1] != float64(1) || v[2] != float64(protocol.EndCodeJoinTimeout) || v[3] != "join timeout exceeded" { + t.Fatalf("unexpected END message: %v", v) + } + if closed, code := c1.isClosed(); !closed || code != protocol.CloseCodeNormal { + t.Fatalf("conn closed=%v code=%d, want closed with %d", closed, code, protocol.CloseCodeNormal) + } +} + +func TestJoinAfterFinishReplaysEnd(t *testing.T) { + m := NewManager(50 * time.Millisecond) + gc := claims(time.Now().Add(time.Hour)) + c1 := &fakeConn{} + + if _, _, err := m.Join(context.Background(), c1, gc, gc.Player1ID); err != nil { + t.Fatalf("join: %v", err) + } + time.Sleep(200 * time.Millisecond) // let the join timer fire + + // A late join with the same (still valid) token must not create a new + // match; the stored END is replayed instead. This also works for the + // player who never connected at all. + for _, userID := range []int64{gc.Player1ID, gc.Player2ID} { + c := &fakeConn{} + _, _, err := m.Join(context.Background(), c, gc, userID) + if err != ErrMatchFinished { + t.Fatalf("join after finish: err = %v, want ErrMatchFinished", err) + } + msgs := c.messages() + if len(msgs) != 1 { + t.Fatalf("replayed %d messages, want 1", len(msgs)) + } + v := parseMsg(t, msgs[0]) + if v[0] != float64(protocol.OpEnd) || v[2] != float64(protocol.EndCodeJoinTimeout) { + t.Fatalf("unexpected replayed message: %v", v) + } + if closed, _ := c.isClosed(); !closed { + t.Fatal("connection was not closed after replaying END") + } + } +} + +func TestReconnectReplacesConnection(t *testing.T) { + m := NewManager(time.Minute) + gc := claims(time.Now().Add(time.Hour)) + old, c2 := &fakeConn{}, &fakeConn{} + + if _, _, err := m.Join(context.Background(), old, gc, gc.Player1ID); err != nil { + t.Fatalf("join p1: %v", err) + } + if _, _, err := m.Join(context.Background(), c2, gc, gc.Player2ID); err != nil { + t.Fatalf("join p2: %v", err) + } + // Game started: both got STATE, nobody acknowledged it yet. + + // Player 1 reconnects: old connection must be closed, the unacknowledged + // STATE must be replayed to the new one. + fresh := &fakeConn{} + if _, idx, err := m.Join(context.Background(), fresh, gc, gc.Player1ID); err != nil { + t.Fatalf("rejoin p1: %v", err) + } else if idx != 0 { + t.Fatalf("slot = %d, want 0", idx) + } + + if closed, code := old.isClosed(); !closed || code != protocol.CloseCodeReplaced { + t.Fatalf("old conn closed=%v code=%d, want closed with %d", closed, code, protocol.CloseCodeReplaced) + } + msgs := fresh.messages() + if len(msgs) != 1 { + t.Fatalf("replayed %d messages, want 1", len(msgs)) + } + v := parseMsg(t, msgs[0]) + if v[0] != float64(protocol.OpState) || v[2] != float64(protocol.StatePlacingShips) { + t.Fatalf("unexpected replayed message: %v", v) + } + + // Leave with the old connection must not detach the new one. + m.Leave(old, mustMatch(t, m, gc.MatchID), 0) + m.Leave(fresh, mustMatch(t, m, gc.MatchID), 0) +} + +func TestAckThenReconnectResendsCurrentState(t *testing.T) { + m := NewManager(time.Minute) + gc := claims(time.Now().Add(time.Hour)) + c1, c2 := &fakeConn{}, &fakeConn{} + + match, _, err := m.Join(context.Background(), c1, gc, gc.Player1ID) + if err != nil { + t.Fatalf("join p1: %v", err) + } + if _, _, err := m.Join(context.Background(), c2, gc, gc.Player2ID); err != nil { + t.Fatalf("join p2: %v", err) + } + + // Player 1 acknowledges the STATE message, then reconnects. The queue + // is empty, so the server re-sends the current state with a fresh seq. + m.Ack(match, 0, 1) + + fresh := &fakeConn{} + if _, _, err := m.Join(context.Background(), fresh, gc, gc.Player1ID); err != nil { + t.Fatalf("rejoin p1: %v", err) + } + msgs := fresh.messages() + if len(msgs) != 1 { + t.Fatalf("got %d messages, want 1", len(msgs)) + } + v := parseMsg(t, msgs[0]) + if v[0] != float64(protocol.OpState) || v[1] != float64(2) || v[2] != float64(protocol.StatePlacingShips) { + t.Fatalf("unexpected message: %v", v) + } +} + +func mustMatch(t *testing.T, m *Manager, id string) *Match { + t.Helper() + m.mu.Lock() + defer m.mu.Unlock() + match, ok := m.matches[id] + if !ok { + t.Fatalf("match %q not found", id) + } + return match +} diff --git a/game-server/internal/protocol/protocol.go b/game-server/internal/protocol/protocol.go new file mode 100644 index 0000000..1a2b05b --- /dev/null +++ b/game-server/internal/protocol/protocol.go @@ -0,0 +1,128 @@ +// Package protocol defines the wire format of messages exchanged between +// game clients and the game server. +// +// All messages are JSON arrays whose first element is a numeric opcode: +// +// [opcode, param1, param2, ...] +// +// Opcodes: +// +// 1 C->S JOIN [1, "", ""] +// 2 S->C END [2, , , ""] +// 3 S->C STATE [3, , ] +// 4 C<->S ACK [4, ] +// +// Every state-changing server->client message (END, STATE) carries a +// per-player monotonically increasing sequence number. The client confirms +// delivery with an ACK message containing the sequence number of the last +// received message (ACK is cumulative). Unacknowledged messages are kept in +// a per-player queue and re-sent when the player reconnects. +// +// ACK in the S->C direction is reserved for confirming future client->server +// messages once the protocol grows (client messages do not carry sequence +// numbers yet). +package protocol + +import ( + "encoding/json" + "fmt" +) + +// Opcodes. +const ( + OpJoin = 1 // C->S: join a match + OpEnd = 2 // S->C: match ended + OpState = 3 // S->C: game state transition + OpAck = 4 // C<->S: delivery acknowledgement +) + +// Game state codes. States are always sent from the receiver's point of +// view: a player never learns whether they are "player 1" or "player 2". +const ( + StatePlacingShips = 1 // both players are placing their ships + StateOpponentReady = 2 // opponent finished placing ships + StateWaitOpponentReady = 3 // you are ready, waiting for the opponent + StateYourTurn = 4 // make your move + StateWaitOpponentTurn = 5 // wait for the opponent's move +) + +// Match end codes. +const ( + EndCodeJoinTimeout = -1 // the second player never joined +) + +// WebSocket close codes used by the server (application range 4000+). +const ( + CloseCodeNormal = 1000 // normal closure + CloseCodeReplaced = 4000 // connection replaced by a newer one of the same player + CloseCodeExpectedJoin = 4001 // first message was not a JOIN + CloseCodeBadMessage = 4002 // malformed message + CloseCodeBadGameToken = 4003 // invalid/expired game token, server mismatch, invalid match + CloseCodeBadUserToken = 4004 // invalid/expired user token or user is not a participant +) + +// ParseMessage decodes a raw message into its opcode and the remaining +// arguments. +func ParseMessage(data []byte) (opcode int, args []json.RawMessage, err error) { + var raw []json.RawMessage + if err := json.Unmarshal(data, &raw); err != nil { + return 0, nil, fmt.Errorf("not a JSON array: %w", err) + } + if len(raw) == 0 { + return 0, nil, fmt.Errorf("empty message") + } + if err := json.Unmarshal(raw[0], &opcode); err != nil { + return 0, nil, fmt.Errorf("opcode is not a number: %w", err) + } + return opcode, raw[1:], nil +} + +// ParseJoin extracts the game token and the user token from JOIN arguments. +func ParseJoin(args []json.RawMessage) (gameToken, userToken string, err error) { + if len(args) != 2 { + return "", "", fmt.Errorf("JOIN expects 2 arguments, got %d", len(args)) + } + if err := json.Unmarshal(args[0], &gameToken); err != nil { + return "", "", fmt.Errorf("game token is not a string: %w", err) + } + if err := json.Unmarshal(args[1], &userToken); err != nil { + return "", "", fmt.Errorf("user token is not a string: %w", err) + } + return gameToken, userToken, nil +} + +// ParseAck extracts the acknowledged sequence number from ACK arguments. +func ParseAck(args []json.RawMessage) (seq int64, err error) { + if len(args) != 1 { + return 0, fmt.Errorf("ACK expects 1 argument, got %d", len(args)) + } + if err := json.Unmarshal(args[0], &seq); err != nil { + return 0, fmt.Errorf("seq is not a number: %w", err) + } + return seq, nil +} + +func build(v []any) []byte { + data, err := json.Marshal(v) + if err != nil { + // Only possible with non-serializable values, which never happens + // with the fixed shapes below. + panic(fmt.Sprintf("protocol: marshal message: %v", err)) + } + return data +} + +// BuildEnd builds an END message: [2, seq, code, reason]. +func BuildEnd(seq int64, code int, reason string) []byte { + return build([]any{OpEnd, seq, code, reason}) +} + +// BuildState builds a STATE message: [3, seq, state]. +func BuildState(seq int64, state int) []byte { + return build([]any{OpState, seq, state}) +} + +// BuildAck builds an ACK message: [4, seq]. +func BuildAck(seq int64) []byte { + return build([]any{OpAck, seq}) +} diff --git a/game-server/internal/protocol/protocol_test.go b/game-server/internal/protocol/protocol_test.go new file mode 100644 index 0000000..c24131c --- /dev/null +++ b/game-server/internal/protocol/protocol_test.go @@ -0,0 +1,71 @@ +package protocol + +import ( + "encoding/json" + "testing" +) + +func TestParseJoinRoundtrip(t *testing.T) { + opcode, args, err := ParseMessage([]byte(`[1,"game-token","user-token"]`)) + if err != nil { + t.Fatalf("parse: %v", err) + } + if opcode != OpJoin { + t.Fatalf("opcode = %d, want %d", opcode, OpJoin) + } + gt, ut, err := ParseJoin(args) + if err != nil { + t.Fatalf("parse join: %v", err) + } + if gt != "game-token" || ut != "user-token" { + t.Fatalf("got (%q, %q)", gt, ut) + } +} + +func TestParseMessageErrors(t *testing.T) { + for _, data := range []string{`{}`, `[]`, `["1"]`, `not json`} { + if _, _, err := ParseMessage([]byte(data)); err == nil { + t.Fatalf("expected error for %s", data) + } + } +} + +func TestBuildEnd(t *testing.T) { + data := BuildEnd(7, EndCodeJoinTimeout, "join timeout exceeded") + var v []any + if err := json.Unmarshal(data, &v); err != nil { + t.Fatalf("unmarshal: %v", err) + } + want := []any{float64(2), float64(7), float64(-1), "join timeout exceeded"} + if len(v) != len(want) { + t.Fatalf("got %v, want %v", v, want) + } + for i := range want { + if v[i] != want[i] { + t.Fatalf("got %v, want %v", v, want) + } + } +} + +func TestBuildStateAndAck(t *testing.T) { + if got := string(BuildState(3, StatePlacingShips)); got != `[3,3,1]` { + t.Fatalf("BuildState = %s", got) + } + if got := string(BuildAck(3)); got != `[4,3]` { + t.Fatalf("BuildAck = %s", got) + } +} + +func TestParseAck(t *testing.T) { + _, args, err := ParseMessage([]byte(`[4,12]`)) + if err != nil { + t.Fatalf("parse: %v", err) + } + seq, err := ParseAck(args) + if err != nil { + t.Fatalf("parse ack: %v", err) + } + if seq != 12 { + t.Fatalf("seq = %d, want 12", seq) + } +} diff --git a/game-server/internal/ws/handler.go b/game-server/internal/ws/handler.go new file mode 100644 index 0000000..d5f5971 --- /dev/null +++ b/game-server/internal/ws/handler.go @@ -0,0 +1,199 @@ +// Package ws implements the websocket endpoint of the game server: +// connection upgrade, JOIN handling and the per-connection read loop. +package ws + +import ( + "context" + "log/slog" + "net/http" + "sync" + "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 +) + +// Conn adapts a websocket connection to the game.Sender interface. +type Conn struct { + ws *websocket.Conn + wmu sync.Mutex // serializes writes +} + +func (c *Conn) Send(_ context.Context, data []byte) error { + ctx, cancel := context.WithTimeout(context.Background(), writeTimeout) + defer cancel() + c.wmu.Lock() + defer c.wmu.Unlock() + return c.ws.Write(ctx, websocket.MessageText, data) +} + +func (c *Conn) Close(code int, reason string) error { + return c.ws.Close(websocket.StatusCode(code), reason) +} + +// 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 + } + + // Validate the game token and make sure the token was issued for this + // particular server instance (strict comparison with the public address). + 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 == 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 != gc.Player1ID && uc.UserID != gc.Player2ID { + wsc.Close(protocol.CloseCodeBadUserToken, "not a participant of this match") + return + } + + conn := &Conn{ws: 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.OpAck: + seq, err := protocol.ParseAck(args) + if err != nil { + h.log.Debug("malformed ACK", "match", match.ID, "user", uc.UserID, "err", err) + continue + } + h.Manager.Ack(match, slotIdx, seq) + 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 + } + } + } +} diff --git a/game-server/internal/ws/handler_test.go b/game-server/internal/ws/handler_test.go new file mode 100644 index 0000000..06a943a --- /dev/null +++ b/game-server/internal/ws/handler_test.go @@ -0,0 +1,192 @@ +package ws_test + +import ( + "context" + "encoding/json" + "log/slog" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/coder/websocket" + "github.com/golang-jwt/jwt/v5" + + "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/ws" +) + +var secret = []byte("integration-secret") + +const publicAddr = "ws://game.example.com/ws" + +func gameToken(t *testing.T, matchID string, p1, p2 int64, gameServer string, exp time.Time) string { + t.Helper() + token, err := jwt.NewWithClaims(jwt.SigningMethodHS256, &auth.GameClaims{ + MatchID: matchID, + Player1ID: p1, + Player2ID: p2, + GameServer: gameServer, + RegisteredClaims: jwt.RegisteredClaims{ + ExpiresAt: jwt.NewNumericDate(exp), + }, + }).SignedString(secret) + if err != nil { + t.Fatalf("sign game token: %v", err) + } + return token +} + +func userToken(t *testing.T, userID int64) string { + t.Helper() + token, err := jwt.NewWithClaims(jwt.SigningMethodHS256, &auth.UserClaims{ + UserID: userID, + RegisteredClaims: jwt.RegisteredClaims{ + ExpiresAt: jwt.NewNumericDate(time.Now().Add(time.Hour)), + }, + }).SignedString(secret) + if err != nil { + t.Fatalf("sign user token: %v", err) + } + return token +} + +func startServer(t *testing.T, joinTimeout time.Duration) (url string) { + t.Helper() + manager := game.NewManager(joinTimeout) + log := slog.New(slog.NewTextHandler(testWriter{t}, nil)) + handler := ws.NewHandler(manager, secret, publicAddr, []string{"*"}, log) + srv := httptest.NewServer(handler) + t.Cleanup(srv.Close) + return "ws" + strings.TrimPrefix(srv.URL, "http") +} + +type testWriter struct{ t *testing.T } + +func (w testWriter) Write(p []byte) (int, error) { + w.t.Log(strings.TrimSpace(string(p))) + return len(p), nil +} + +func dial(t *testing.T, url string) *websocket.Conn { + t.Helper() + c, _, err := websocket.Dial(context.Background(), url, nil) + if err != nil { + t.Fatalf("dial: %v", err) + } + t.Cleanup(func() { c.Close(websocket.StatusNormalClosure, "") }) + return c +} + +func join(t *testing.T, c *websocket.Conn, gt, ut string) { + t.Helper() + msg, _ := json.Marshal([]any{1, gt, ut}) + if err := c.Write(context.Background(), websocket.MessageText, msg); err != nil { + t.Fatalf("write JOIN: %v", err) + } +} + +func readMsg(t *testing.T, c *websocket.Conn) []any { + t.Helper() + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + _, data, err := c.Read(ctx) + if err != nil { + t.Fatalf("read: %v", err) + } + var v []any + if err := json.Unmarshal(data, &v); err != nil { + t.Fatalf("unmarshal %s: %v", data, err) + } + return v +} + +func TestTwoPlayersStartMatch(t *testing.T) { + url := startServer(t, time.Minute) + exp := time.Now().Add(time.Hour) + gt := gameToken(t, "m-1", 1, 2, publicAddr, exp) + + c1, c2 := dial(t, url), dial(t, url) + join(t, c1, gt, userToken(t, 1)) + join(t, c2, gt, userToken(t, 2)) + + for i, c := range []*websocket.Conn{c1, c2} { + v := readMsg(t, c) + if v[0] != float64(3) || v[1] != float64(1) || v[2] != float64(1) { + t.Fatalf("player %d: unexpected message %v", i, v) + } + // ACK the state. + msg, _ := json.Marshal([]any{4, 1}) + if err := c.Write(context.Background(), websocket.MessageText, msg); err != nil { + t.Fatalf("write ACK: %v", err) + } + } +} + +func TestWrongServerAddressRejected(t *testing.T) { + url := startServer(t, time.Minute) + gt := gameToken(t, "m-2", 1, 2, "ws://other.example.com/ws", time.Now().Add(time.Hour)) + + c := dial(t, url) + join(t, c, gt, userToken(t, 1)) + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + _, _, err := c.Read(ctx) + if err == nil { + t.Fatal("expected connection to be closed") + } + if code := websocket.CloseStatus(err); code != 4003 { + t.Fatalf("close code = %d, want 4003", code) + } +} + +func TestJoinTimeoutEndsMatchAndReplay(t *testing.T) { + url := startServer(t, 200*time.Millisecond) + exp := time.Now().Add(time.Hour) + gt := gameToken(t, "m-3", 1, 2, publicAddr, exp) + + c1 := dial(t, url) + join(t, c1, gt, userToken(t, 1)) + + v := readMsg(t, c1) // END + if v[0] != float64(2) || v[1] != float64(1) || v[2] != float64(-1) || v[3] != "join timeout exceeded" { + t.Fatalf("unexpected END: %v", v) + } + + // Late join with the same token replays END and closes. + c2 := dial(t, url) + join(t, c2, gt, userToken(t, 2)) + v = readMsg(t, c2) + if v[0] != float64(2) || v[2] != float64(-1) { + t.Fatalf("unexpected replayed END: %v", v) + } + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if _, _, err := c2.Read(ctx); err == nil { + t.Fatal("expected connection to be closed after END replay") + } +} + +func TestReconnectDuringGame(t *testing.T) { + url := startServer(t, time.Minute) + exp := time.Now().Add(time.Hour) + gt := gameToken(t, "m-4", 1, 2, publicAddr, exp) + + c1, c2 := dial(t, url), dial(t, url) + join(t, c1, gt, userToken(t, 1)) + join(t, c2, gt, userToken(t, 2)) + readMsg(t, c1) // STATE, not acknowledged + + // Simulate a dropped connection. + c1.Close(websocket.StatusAbnormalClosure, "network gone") + + // Reconnect with the same tokens: unacknowledged STATE is replayed. + c1b := dial(t, url) + join(t, c1b, gt, userToken(t, 1)) + v := readMsg(t, c1b) + if v[0] != float64(3) || v[1] != float64(1) || v[2] != float64(1) { + t.Fatalf("unexpected replayed message: %v", v) + } +}