// 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) } } }