create a websocket game server

2026-09-22 09:08:13 +03:00
parent 4758ffeeb0
commit d6efb46f87
13 changed files with 1634 additions and 0 deletions
@@ -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)
+		}
+	}
+}
@@ -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
+}