f452e69999
- TurnTimeoutService with percentage-based warnings (35 tests) - ConnectionManager enhancements for spectators and reconnection - GameService with timer integration, spectator support, handle_timeout - GameNamespace with spectate/leave_spectate handlers, reconnection - WebSocket message schemas for spectator events - WinConditionsConfig additions for turn timer thresholds - 83 GameService tests, 37 ConnectionManager tests, 37 GameNamespace tests Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
1225 lines
44 KiB
Python
1225 lines
44 KiB
Python
"""Game service for orchestrating game lifecycle in Mantimon TCG.
|
|
|
|
This service is the bridge between WebSocket communication and the core
|
|
GameEngine. It handles:
|
|
- Game creation with deck loading and validation
|
|
- Action execution with persistence
|
|
- Game state retrieval with visibility filtering
|
|
- Game lifecycle (join, resign, end)
|
|
|
|
IMPORTANT: This service is stateless. All game-specific configuration
|
|
(RulesConfig) is stored in the GameState itself, not in this service.
|
|
Rules come from the frontend request at game creation time.
|
|
|
|
Architecture:
|
|
WebSocket Layer -> GameService -> GameEngine + GameStateManager
|
|
-> DeckService + CardService
|
|
|
|
Example:
|
|
from app.services.game_service import GameService, game_service
|
|
|
|
# Create a new game (rules from frontend)
|
|
result = await game_service.create_game(
|
|
player1_id=user1.id,
|
|
player2_id=user2.id,
|
|
deck1_id=deck1.id,
|
|
deck2_id=deck2.id,
|
|
rules_config=rules_from_request, # Frontend provides this
|
|
)
|
|
|
|
# Execute an action (uses rules stored in game state)
|
|
result = await game_service.execute_action(
|
|
game_id=game.id,
|
|
player_id=user1.id,
|
|
action=AttackAction(attack_index=0),
|
|
)
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import uuid as uuid_module
|
|
from collections.abc import Callable
|
|
from dataclasses import dataclass, field
|
|
from typing import TYPE_CHECKING, Any
|
|
from uuid import UUID
|
|
|
|
if TYPE_CHECKING:
|
|
from app.services.deck_service import DeckService
|
|
|
|
from app.core.config import RulesConfig
|
|
from app.core.engine import ActionResult, GameCreationResult, GameEngine
|
|
from app.core.enums import GameEndReason, TurnPhase
|
|
from app.core.models.actions import Action, ResignAction
|
|
from app.core.models.card import CardInstance
|
|
from app.core.models.game_state import GameState
|
|
from app.core.rng import create_rng
|
|
from app.core.visibility import VisibleGameState, get_spectator_state, get_visible_state
|
|
from app.db.models.game import EndReason, GameType
|
|
from app.services.card_service import CardService, get_card_service
|
|
from app.services.game_state_manager import GameStateManager, game_state_manager
|
|
from app.services.turn_timeout_service import TurnTimeoutService, turn_timeout_service
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Type alias for engine factory - takes GameState, returns GameEngine (for execute_action)
|
|
EngineFactory = Callable[[GameState], GameEngine]
|
|
|
|
# Type alias for creation factory - takes RulesConfig, returns GameEngine (for create_game)
|
|
CreationEngineFactory = Callable[[RulesConfig], GameEngine]
|
|
|
|
|
|
def _map_end_reason(core_reason: GameEndReason) -> EndReason:
|
|
"""Map core GameEndReason to database EndReason.
|
|
|
|
The core module uses its own enum for offline independence,
|
|
while the database has a separate enum for persistence. The DB enum
|
|
also includes service-layer reasons (like DISCONNECTION) that the
|
|
core engine doesn't know about.
|
|
|
|
Args:
|
|
core_reason: The GameEndReason from the core module.
|
|
|
|
Returns:
|
|
The corresponding EndReason for database storage.
|
|
|
|
Raises:
|
|
ValueError: If core_reason is not in the mapping (indicates missing enum sync).
|
|
"""
|
|
mapping = {
|
|
GameEndReason.PRIZES_TAKEN: EndReason.PRIZES_TAKEN,
|
|
GameEndReason.NO_POKEMON: EndReason.NO_POKEMON,
|
|
GameEndReason.DECK_EMPTY: EndReason.CANNOT_DRAW,
|
|
GameEndReason.RESIGNATION: EndReason.RESIGNATION,
|
|
GameEndReason.TIMEOUT: EndReason.TIMEOUT,
|
|
GameEndReason.TURN_LIMIT: EndReason.TURN_LIMIT,
|
|
GameEndReason.DRAW: EndReason.DRAW,
|
|
}
|
|
result = mapping.get(core_reason)
|
|
if result is None:
|
|
raise ValueError(
|
|
f"Unknown GameEndReason: {core_reason}. "
|
|
"Update _map_end_reason when adding new end reasons to core."
|
|
)
|
|
return result
|
|
|
|
|
|
# =============================================================================
|
|
# Exceptions
|
|
# =============================================================================
|
|
|
|
|
|
class GameServiceError(Exception):
|
|
"""Base exception for GameService errors."""
|
|
|
|
pass
|
|
|
|
|
|
class GameNotFoundError(GameServiceError):
|
|
"""Raised when a game cannot be found."""
|
|
|
|
def __init__(self, game_id: str) -> None:
|
|
self.game_id = game_id
|
|
super().__init__(f"Game not found: {game_id}")
|
|
|
|
|
|
class NotPlayerTurnError(GameServiceError):
|
|
"""Raised when a player tries to act out of turn."""
|
|
|
|
def __init__(self, game_id: str, player_id: str, current_player_id: str) -> None:
|
|
self.game_id = game_id
|
|
self.player_id = player_id
|
|
self.current_player_id = current_player_id
|
|
super().__init__(
|
|
f"Not player's turn: {player_id} tried to act, but it's {current_player_id}'s turn"
|
|
)
|
|
|
|
|
|
class InvalidActionError(GameServiceError):
|
|
"""Raised when an action is invalid."""
|
|
|
|
def __init__(self, game_id: str, player_id: str, reason: str) -> None:
|
|
self.game_id = game_id
|
|
self.player_id = player_id
|
|
self.reason = reason
|
|
super().__init__(f"Invalid action in game {game_id}: {reason}")
|
|
|
|
|
|
class PlayerNotInGameError(GameServiceError):
|
|
"""Raised when a player is not a participant in the game."""
|
|
|
|
def __init__(self, game_id: str, player_id: str) -> None:
|
|
self.game_id = game_id
|
|
self.player_id = player_id
|
|
super().__init__(f"Player {player_id} is not in game {game_id}")
|
|
|
|
|
|
class GameAlreadyEndedError(GameServiceError):
|
|
"""Raised when trying to act on a game that has already ended."""
|
|
|
|
def __init__(self, game_id: str) -> None:
|
|
self.game_id = game_id
|
|
super().__init__(f"Game {game_id} has already ended")
|
|
|
|
|
|
class GameCreationError(GameServiceError):
|
|
"""Raised when game creation fails."""
|
|
|
|
def __init__(self, reason: str) -> None:
|
|
self.reason = reason
|
|
super().__init__(f"Failed to create game: {reason}")
|
|
|
|
|
|
class ForcedActionRequiredError(GameServiceError):
|
|
"""Raised when a forced action is required but a different action was attempted."""
|
|
|
|
def __init__(
|
|
self,
|
|
game_id: str,
|
|
player_id: str,
|
|
required_action_type: str,
|
|
attempted_action_type: str,
|
|
) -> None:
|
|
self.game_id = game_id
|
|
self.player_id = player_id
|
|
self.required_action_type = required_action_type
|
|
self.attempted_action_type = attempted_action_type
|
|
super().__init__(
|
|
f"Forced action required: {required_action_type}, "
|
|
f"but {attempted_action_type} was attempted"
|
|
)
|
|
|
|
|
|
class CannotSpectateOwnGameError(GameServiceError):
|
|
"""Raised when a player tries to spectate a game they are participating in."""
|
|
|
|
def __init__(self, game_id: str, player_id: str) -> None:
|
|
self.game_id = game_id
|
|
self.player_id = player_id
|
|
super().__init__(f"Cannot spectate your own game: {game_id}")
|
|
|
|
|
|
# =============================================================================
|
|
# Result Types
|
|
# =============================================================================
|
|
|
|
|
|
@dataclass
|
|
class PendingForcedAction:
|
|
"""Information about a pending forced action for the client.
|
|
|
|
Attributes:
|
|
player_id: The player who must take the action.
|
|
action_type: The type of action required.
|
|
reason: Human-readable explanation.
|
|
params: Additional parameters for the action.
|
|
"""
|
|
|
|
player_id: str
|
|
action_type: str
|
|
reason: str
|
|
params: dict[str, Any] = field(default_factory=dict)
|
|
|
|
|
|
@dataclass
|
|
class GameActionResult:
|
|
"""Result of executing a game action.
|
|
|
|
Attributes:
|
|
success: Whether the action succeeded.
|
|
game_id: The game ID.
|
|
action_type: The type of action executed.
|
|
message: Description of what happened.
|
|
state_changes: Dict of state changes for client updates.
|
|
game_over: Whether the game ended as a result.
|
|
winner_id: Winner's player ID if game ended with a winner.
|
|
end_reason: Reason the game ended, if applicable.
|
|
turn_changed: Whether the turn changed as a result of this action.
|
|
current_player_id: The current player after action execution.
|
|
pending_forced_action: If set, the next action must be this forced action.
|
|
turn_timeout_seconds: Seconds remaining on turn timer (None if disabled).
|
|
turn_deadline: Unix timestamp when current turn expires (None if disabled).
|
|
"""
|
|
|
|
success: bool
|
|
game_id: str
|
|
action_type: str
|
|
message: str = ""
|
|
state_changes: dict[str, Any] = field(default_factory=dict)
|
|
game_over: bool = False
|
|
winner_id: str | None = None
|
|
end_reason: GameEndReason | None = None
|
|
turn_changed: bool = False
|
|
current_player_id: str | None = None
|
|
pending_forced_action: PendingForcedAction | None = None
|
|
turn_timeout_seconds: int | None = None
|
|
turn_deadline: float | None = None
|
|
|
|
|
|
@dataclass
|
|
class GameJoinResult:
|
|
"""Result of joining or rejoining a game.
|
|
|
|
Used when a player connects/reconnects to a game session. Contains
|
|
everything needed to render the game UI and know what action is required.
|
|
|
|
Attributes:
|
|
success: Whether the join succeeded.
|
|
game_id: The game ID.
|
|
player_id: The joining player's ID.
|
|
visible_state: The game state visible to the player.
|
|
is_your_turn: Whether it's this player's turn (or forced action required).
|
|
game_over: Whether the game has already ended.
|
|
pending_forced_action: If set, this action must be taken before any other.
|
|
message: Additional information or error message.
|
|
turn_timeout_seconds: Seconds remaining on turn timer (None if disabled).
|
|
turn_deadline: Unix timestamp when current turn expires (None if disabled).
|
|
"""
|
|
|
|
success: bool
|
|
game_id: str
|
|
player_id: str
|
|
visible_state: VisibleGameState | None = None
|
|
is_your_turn: bool = False
|
|
game_over: bool = False
|
|
pending_forced_action: PendingForcedAction | None = None
|
|
message: str = ""
|
|
turn_timeout_seconds: int | None = None
|
|
turn_deadline: float | None = None
|
|
|
|
|
|
@dataclass
|
|
class GameCreateResult:
|
|
"""Result of creating a new game.
|
|
|
|
Attributes:
|
|
success: Whether the game was created successfully.
|
|
game_id: The created game's ID (None if failed).
|
|
player1_view: Initial state visible to player 1.
|
|
player2_view: Initial state visible to player 2.
|
|
starting_player_id: Which player goes first.
|
|
message: Description or error message.
|
|
"""
|
|
|
|
success: bool
|
|
game_id: str | None = None
|
|
player1_view: VisibleGameState | None = None
|
|
player2_view: VisibleGameState | None = None
|
|
starting_player_id: str | None = None
|
|
message: str = ""
|
|
|
|
|
|
@dataclass
|
|
class GameEndResult:
|
|
"""Result of ending a game.
|
|
|
|
Contains all information needed to notify clients and record
|
|
the game in history.
|
|
|
|
Attributes:
|
|
success: Whether the game was ended successfully.
|
|
game_id: The game that ended.
|
|
winner_id: The winner's player ID (None for draws).
|
|
loser_id: The loser's player ID (None for draws).
|
|
end_reason: Why the game ended.
|
|
turn_count: Total number of turns played.
|
|
duration_seconds: Game duration in seconds.
|
|
player1_final_view: Final state visible to player 1.
|
|
player2_final_view: Final state visible to player 2.
|
|
history_id: The GameHistory record ID.
|
|
message: Description or error message.
|
|
"""
|
|
|
|
success: bool
|
|
game_id: str
|
|
winner_id: str | None = None
|
|
loser_id: str | None = None
|
|
end_reason: GameEndReason | None = None
|
|
turn_count: int = 0
|
|
duration_seconds: int = 0
|
|
player1_final_view: VisibleGameState | None = None
|
|
player2_final_view: VisibleGameState | None = None
|
|
history_id: str | None = None
|
|
message: str = ""
|
|
|
|
|
|
@dataclass
|
|
class SpectateResult:
|
|
"""Result of spectating a game.
|
|
|
|
Attributes:
|
|
success: Whether spectating succeeded.
|
|
game_id: The game being spectated.
|
|
visible_state: Spectator-filtered game state (no hands visible).
|
|
game_over: Whether the game has already ended.
|
|
message: Additional information or error message.
|
|
"""
|
|
|
|
success: bool
|
|
game_id: str
|
|
visible_state: VisibleGameState | None = None
|
|
game_over: bool = False
|
|
message: str = ""
|
|
|
|
|
|
# =============================================================================
|
|
# GameService
|
|
# =============================================================================
|
|
|
|
|
|
class GameService:
|
|
"""Service for orchestrating game lifecycle operations.
|
|
|
|
This service coordinates between the WebSocket layer and the core
|
|
GameEngine, handling persistence and state management.
|
|
|
|
IMPORTANT: This service is STATELESS regarding game rules.
|
|
- Rules are stored in each GameState (set at creation time)
|
|
- The GameEngine is instantiated per-operation with the game's rules
|
|
- No RulesConfig is stored in this service
|
|
|
|
Attributes:
|
|
_state_manager: GameStateManager for persistence.
|
|
_card_service: CardService for card definitions.
|
|
_timeout_service: TurnTimeoutService for turn timer management.
|
|
_engine_factory: Factory for creating GameEngine for action execution.
|
|
_creation_engine_factory: Factory for creating GameEngine for game creation.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
state_manager: GameStateManager | None = None,
|
|
card_service: CardService | None = None,
|
|
timeout_service: TurnTimeoutService | None = None,
|
|
engine_factory: EngineFactory | None = None,
|
|
creation_engine_factory: CreationEngineFactory | None = None,
|
|
) -> None:
|
|
"""Initialize the GameService.
|
|
|
|
Note: No GameEngine or RulesConfig here - those are per-game,
|
|
not per-service. The engine is created as needed using the
|
|
rules stored in each game's state.
|
|
|
|
Args:
|
|
state_manager: GameStateManager instance. Uses global if not provided.
|
|
card_service: CardService instance. Uses global if not provided.
|
|
timeout_service: TurnTimeoutService instance. Uses global if not provided.
|
|
engine_factory: Optional factory for creating GameEngine for action
|
|
execution. Takes GameState, returns GameEngine. If not provided,
|
|
uses the default _default_engine_factory method.
|
|
creation_engine_factory: Optional factory for creating GameEngine for
|
|
game creation. Takes RulesConfig, returns GameEngine. If not
|
|
provided, uses the default _default_creation_engine_factory method.
|
|
"""
|
|
self._state_manager = state_manager or game_state_manager
|
|
self._card_service = card_service or get_card_service()
|
|
self._timeout_service = timeout_service or turn_timeout_service
|
|
self._engine_factory = engine_factory or self._default_engine_factory
|
|
self._creation_engine_factory = (
|
|
creation_engine_factory or self._default_creation_engine_factory
|
|
)
|
|
|
|
def _default_engine_factory(self, game: GameState) -> GameEngine:
|
|
"""Default factory for creating a GameEngine from game state.
|
|
|
|
The engine is created on-demand using the rules stored in the
|
|
game state. This ensures each game uses its own configuration.
|
|
|
|
For deterministic replay support, we derive a unique seed per action
|
|
by combining the game's base seed with the action count. This ensures:
|
|
- Same game + same action sequence = identical RNG results
|
|
- Each action gets a unique but reproducible random sequence
|
|
|
|
Args:
|
|
game: The game state containing the rules to use.
|
|
|
|
Returns:
|
|
A GameEngine configured with the game's rules and RNG.
|
|
"""
|
|
if game.rng_seed is not None:
|
|
# Derive unique seed per action for deterministic replay
|
|
# Action count ensures each action gets different but reproducible RNG
|
|
action_count = len(game.action_log)
|
|
action_seed = game.rng_seed + action_count
|
|
rng = create_rng(seed=action_seed)
|
|
else:
|
|
# No seed - use cryptographically secure RNG
|
|
rng = create_rng()
|
|
|
|
return GameEngine(rules=game.rules, rng=rng)
|
|
|
|
def _default_creation_engine_factory(self, rules: RulesConfig) -> GameEngine:
|
|
"""Default factory for creating a GameEngine for game creation.
|
|
|
|
Used when creating a new game. The engine is created with the
|
|
provided rules and a fresh RNG.
|
|
|
|
Args:
|
|
rules: The RulesConfig for the new game.
|
|
|
|
Returns:
|
|
A GameEngine configured with the rules and fresh RNG.
|
|
"""
|
|
return GameEngine(rules=rules, rng=create_rng())
|
|
|
|
# =========================================================================
|
|
# Game State Access
|
|
# =========================================================================
|
|
|
|
async def get_game_state(self, game_id: str) -> GameState:
|
|
"""Get the full game state.
|
|
|
|
Args:
|
|
game_id: The game ID.
|
|
|
|
Returns:
|
|
The GameState.
|
|
|
|
Raises:
|
|
GameNotFoundError: If game doesn't exist.
|
|
"""
|
|
state = await self._state_manager.load_state(game_id)
|
|
if state is None:
|
|
raise GameNotFoundError(game_id)
|
|
return state
|
|
|
|
async def get_player_view(
|
|
self,
|
|
game_id: str,
|
|
player_id: str,
|
|
) -> VisibleGameState:
|
|
"""Get the game state filtered for a specific player's view.
|
|
|
|
This applies visibility rules to hide opponent's hidden information
|
|
(hand, deck, prizes).
|
|
|
|
Args:
|
|
game_id: The game ID.
|
|
player_id: The player to get the view for.
|
|
|
|
Returns:
|
|
VisibleGameState with appropriate filtering.
|
|
|
|
Raises:
|
|
GameNotFoundError: If game doesn't exist.
|
|
PlayerNotInGameError: If player is not in the game.
|
|
"""
|
|
state = await self.get_game_state(game_id)
|
|
|
|
if player_id not in state.players:
|
|
raise PlayerNotInGameError(game_id, player_id)
|
|
|
|
return get_visible_state(state, player_id)
|
|
|
|
async def is_player_turn(self, game_id: str, player_id: str) -> bool:
|
|
"""Check if it's the specified player's turn.
|
|
|
|
Args:
|
|
game_id: The game ID.
|
|
player_id: The player to check.
|
|
|
|
Returns:
|
|
True if it's the player's turn.
|
|
|
|
Raises:
|
|
GameNotFoundError: If game doesn't exist.
|
|
"""
|
|
state = await self.get_game_state(game_id)
|
|
return state.current_player_id == player_id
|
|
|
|
async def game_exists(self, game_id: str) -> bool:
|
|
"""Check if a game exists.
|
|
|
|
Args:
|
|
game_id: The game ID.
|
|
|
|
Returns:
|
|
True if the game exists in cache or database.
|
|
"""
|
|
return await self._state_manager.cache_exists(game_id)
|
|
|
|
# =========================================================================
|
|
# Game Lifecycle
|
|
# =========================================================================
|
|
|
|
async def join_game(
|
|
self,
|
|
game_id: str,
|
|
player_id: str,
|
|
last_event_id: str | None = None,
|
|
) -> GameJoinResult:
|
|
"""Join or rejoin a game session.
|
|
|
|
Loads the game state and returns the player's visible view.
|
|
Used when a player connects or reconnects to a game.
|
|
|
|
This method handles both initial joins and reconnections after
|
|
disconnect. On reconnect, the full current state is returned
|
|
including any pending forced actions that the player must complete.
|
|
|
|
Args:
|
|
game_id: The game to join.
|
|
player_id: The joining player's ID.
|
|
last_event_id: Last event ID for reconnection replay (future use).
|
|
Will be used to replay missed events after reconnect.
|
|
|
|
Returns:
|
|
GameJoinResult with the visible state and any pending actions.
|
|
"""
|
|
try:
|
|
state = await self.get_game_state(game_id)
|
|
except GameNotFoundError:
|
|
return GameJoinResult(
|
|
success=False,
|
|
game_id=game_id,
|
|
player_id=player_id,
|
|
message="Game not found",
|
|
)
|
|
|
|
if player_id not in state.players:
|
|
return GameJoinResult(
|
|
success=False,
|
|
game_id=game_id,
|
|
player_id=player_id,
|
|
message="You are not a participant in this game",
|
|
)
|
|
|
|
visible = get_visible_state(state, player_id)
|
|
|
|
# Check if game already ended
|
|
if state.winner_id is not None or state.end_reason is not None:
|
|
logger.info(f"Player {player_id} joined ended game {game_id}")
|
|
return GameJoinResult(
|
|
success=True,
|
|
game_id=game_id,
|
|
player_id=player_id,
|
|
visible_state=visible,
|
|
is_your_turn=False,
|
|
game_over=True,
|
|
message="Game has ended",
|
|
)
|
|
|
|
# Check for pending forced action
|
|
forced = state.get_current_forced_action()
|
|
pending_forced: PendingForcedAction | None = None
|
|
if forced is not None:
|
|
pending_forced = PendingForcedAction(
|
|
player_id=forced.player_id,
|
|
action_type=forced.action_type,
|
|
reason=forced.reason,
|
|
params=forced.params or {},
|
|
)
|
|
|
|
# Determine if it's this player's turn
|
|
# It's their turn if: normal turn OR they have a forced action
|
|
is_turn = state.current_player_id == player_id
|
|
if forced is not None and forced.player_id == player_id:
|
|
is_turn = True
|
|
|
|
# Handle turn timer for reconnection
|
|
turn_timeout_seconds: int | None = None
|
|
turn_deadline: float | None = None
|
|
|
|
if state.rules.win_conditions.turn_timer_enabled:
|
|
# Check if there's an active timer
|
|
timeout_info = await self._timeout_service.get_timeout_info(game_id)
|
|
|
|
if timeout_info is not None:
|
|
# Timer exists - extend if this is the current player reconnecting
|
|
if is_turn and timeout_info.player_id == player_id:
|
|
grace_seconds = state.rules.win_conditions.turn_timer_grace_seconds
|
|
extended_info = await self._timeout_service.extend_timer(game_id, grace_seconds)
|
|
if extended_info is not None:
|
|
timeout_info = extended_info
|
|
logger.debug(
|
|
f"Extended turn timer on reconnect: game={game_id}, "
|
|
f"player={player_id}, grace={grace_seconds}s"
|
|
)
|
|
|
|
turn_timeout_seconds = timeout_info.remaining_seconds
|
|
turn_deadline = timeout_info.deadline
|
|
|
|
logger.info(f"Player {player_id} joined game {game_id}")
|
|
|
|
return GameJoinResult(
|
|
success=True,
|
|
game_id=game_id,
|
|
player_id=player_id,
|
|
visible_state=visible,
|
|
is_your_turn=is_turn,
|
|
game_over=False,
|
|
pending_forced_action=pending_forced,
|
|
turn_timeout_seconds=turn_timeout_seconds,
|
|
turn_deadline=turn_deadline,
|
|
)
|
|
|
|
async def spectate_game(
|
|
self,
|
|
game_id: str,
|
|
user_id: str,
|
|
) -> SpectateResult:
|
|
"""Get spectator view of a game.
|
|
|
|
Returns a visibility-filtered game state suitable for spectators.
|
|
Spectators cannot see any player's hand, deck, or prizes.
|
|
|
|
Args:
|
|
game_id: The game to spectate.
|
|
user_id: The user wanting to spectate.
|
|
|
|
Returns:
|
|
SpectateResult with the spectator-visible state.
|
|
|
|
Raises:
|
|
GameNotFoundError: If game doesn't exist.
|
|
CannotSpectateOwnGameError: If user is a participant in the game.
|
|
"""
|
|
state = await self.get_game_state(game_id)
|
|
|
|
# Players cannot spectate their own game
|
|
if user_id in state.players:
|
|
raise CannotSpectateOwnGameError(game_id, user_id)
|
|
|
|
visible = get_spectator_state(state)
|
|
|
|
# Check if game already ended
|
|
game_over = state.winner_id is not None or state.end_reason is not None
|
|
|
|
logger.info(f"User {user_id} spectating game {game_id}")
|
|
|
|
return SpectateResult(
|
|
success=True,
|
|
game_id=game_id,
|
|
visible_state=visible,
|
|
game_over=game_over,
|
|
message="Spectating game" if not game_over else "Game has ended",
|
|
)
|
|
|
|
async def execute_action(
|
|
self,
|
|
game_id: str,
|
|
player_id: str,
|
|
action: Action,
|
|
) -> GameActionResult:
|
|
"""Execute a player action in the game.
|
|
|
|
Validates the action, executes it through GameEngine, and
|
|
persists the updated state. The GameEngine is created using
|
|
the rules stored in the game state.
|
|
|
|
Handles forced actions: if there's a pending forced action (e.g.,
|
|
select new active after KO), only that action type from the
|
|
specified player is allowed.
|
|
|
|
Persists to database at turn boundaries for durability.
|
|
|
|
Args:
|
|
game_id: The game ID.
|
|
player_id: The acting player's ID.
|
|
action: The action to execute.
|
|
|
|
Returns:
|
|
GameActionResult with success status, state changes, and
|
|
any pending forced actions.
|
|
|
|
Raises:
|
|
GameNotFoundError: If game doesn't exist.
|
|
PlayerNotInGameError: If player is not in the game.
|
|
GameAlreadyEndedError: If game has already ended.
|
|
NotPlayerTurnError: If it's not the player's turn.
|
|
ForcedActionRequiredError: If a forced action is required but
|
|
a different action was attempted.
|
|
InvalidActionError: If the action is invalid.
|
|
"""
|
|
# Load game state
|
|
state = await self.get_game_state(game_id)
|
|
|
|
# Validate player is in game
|
|
if player_id not in state.players:
|
|
raise PlayerNotInGameError(game_id, player_id)
|
|
|
|
# Check game hasn't ended
|
|
if state.winner_id is not None or state.end_reason is not None:
|
|
raise GameAlreadyEndedError(game_id)
|
|
|
|
# Check for forced actions (except resignation, which is always allowed)
|
|
forced = state.get_current_forced_action()
|
|
if forced is not None and not isinstance(action, ResignAction):
|
|
# Only the specified player can act during a forced action
|
|
if player_id != forced.player_id:
|
|
raise NotPlayerTurnError(game_id, player_id, forced.player_id)
|
|
# Only the specified action type is allowed
|
|
if action.type != forced.action_type:
|
|
raise ForcedActionRequiredError(game_id, player_id, forced.action_type, action.type)
|
|
elif not isinstance(action, ResignAction) and state.current_player_id != player_id:
|
|
# Normal turn check (no forced action pending)
|
|
raise NotPlayerTurnError(game_id, player_id, state.current_player_id)
|
|
|
|
# Track turn state before action for boundary detection
|
|
turn_before = state.turn_number
|
|
player_before = state.current_player_id
|
|
phase_before = state.phase
|
|
|
|
# Create engine with this game's rules via factory
|
|
engine = self._engine_factory(state)
|
|
|
|
# Execute the action
|
|
result: ActionResult = await engine.execute_action(state, player_id, action)
|
|
|
|
if not result.success:
|
|
raise InvalidActionError(game_id, player_id, result.message)
|
|
|
|
# Detect turn change
|
|
turn_changed = state.turn_number != turn_before or state.current_player_id != player_before
|
|
|
|
# Save state to cache (fast path)
|
|
await self._state_manager.save_to_cache(state)
|
|
|
|
# Persist to DB at turn boundaries for durability
|
|
if turn_changed:
|
|
await self._state_manager.persist_to_db(state)
|
|
logger.debug(f"Turn boundary: persisted game {game_id} to DB")
|
|
|
|
# Build response
|
|
action_result = GameActionResult(
|
|
success=True,
|
|
game_id=game_id,
|
|
action_type=action.type,
|
|
message=result.message,
|
|
state_changes={
|
|
"changes": result.state_changes,
|
|
},
|
|
turn_changed=turn_changed,
|
|
current_player_id=state.current_player_id,
|
|
)
|
|
|
|
# Include pending forced action if any
|
|
next_forced = state.get_current_forced_action()
|
|
if next_forced is not None:
|
|
action_result.pending_forced_action = PendingForcedAction(
|
|
player_id=next_forced.player_id,
|
|
action_type=next_forced.action_type,
|
|
reason=next_forced.reason,
|
|
params=next_forced.params,
|
|
)
|
|
|
|
# Check for game over
|
|
if result.win_result is not None:
|
|
action_result.game_over = True
|
|
action_result.winner_id = result.win_result.winner_id
|
|
action_result.end_reason = result.win_result.end_reason
|
|
|
|
# Cancel turn timer on game over
|
|
await self._timeout_service.cancel_timer(game_id)
|
|
|
|
# Persist final state to DB
|
|
await self._state_manager.persist_to_db(state)
|
|
|
|
logger.info(
|
|
f"Game {game_id} ended: winner={result.win_result.winner_id}, "
|
|
f"reason={result.win_result.end_reason}"
|
|
)
|
|
elif state.rules.win_conditions.turn_timer_enabled:
|
|
# Determine if we should start the turn timer:
|
|
# 1. Turn changed (player switched turns)
|
|
# 2. SETUP phase just ended (first real turn began)
|
|
setup_ended = phase_before == TurnPhase.SETUP and state.phase != TurnPhase.SETUP
|
|
should_start_timer = turn_changed or setup_ended
|
|
|
|
if should_start_timer:
|
|
timeout_info = await self._timeout_service.start_turn_timer(
|
|
game_id=game_id,
|
|
player_id=state.current_player_id,
|
|
timeout_seconds=state.rules.win_conditions.turn_timer_seconds,
|
|
warning_thresholds=state.rules.win_conditions.turn_timer_warning_thresholds,
|
|
)
|
|
action_result.turn_timeout_seconds = timeout_info.remaining_seconds
|
|
action_result.turn_deadline = timeout_info.deadline
|
|
logger.debug(
|
|
f"Started turn timer: game={game_id}, player={state.current_player_id}, "
|
|
f"timeout={timeout_info.timeout_seconds}s, "
|
|
f"reason={'setup_ended' if setup_ended else 'turn_changed'}"
|
|
)
|
|
|
|
logger.debug(f"Action executed: game={game_id}, player={player_id}, type={action.type}")
|
|
|
|
return action_result
|
|
|
|
async def resign_game(
|
|
self,
|
|
game_id: str,
|
|
player_id: str,
|
|
) -> GameActionResult:
|
|
"""Resign from a game.
|
|
|
|
Convenience method that executes a ResignAction.
|
|
|
|
Args:
|
|
game_id: The game ID.
|
|
player_id: The resigning player's ID.
|
|
|
|
Returns:
|
|
GameActionResult indicating game over.
|
|
"""
|
|
return await self.execute_action(
|
|
game_id=game_id,
|
|
player_id=player_id,
|
|
action=ResignAction(),
|
|
)
|
|
|
|
async def handle_timeout(
|
|
self,
|
|
game_id: str,
|
|
timed_out_player_id: str,
|
|
) -> GameEndResult:
|
|
"""Handle a turn timeout.
|
|
|
|
Called by the background timeout polling task when a player's
|
|
turn timer expires. Declares the timed-out player as the loser.
|
|
|
|
Future enhancement: Could implement auto-pass for first timeout,
|
|
loss only after N consecutive timeouts.
|
|
|
|
Args:
|
|
game_id: The game ID.
|
|
timed_out_player_id: The player who timed out.
|
|
|
|
Returns:
|
|
GameEndResult with timeout as the end reason.
|
|
|
|
Raises:
|
|
GameNotFoundError: If game doesn't exist.
|
|
"""
|
|
state = await self.get_game_state(game_id)
|
|
|
|
# Determine winner (the opponent)
|
|
player_ids = list(state.players.keys())
|
|
winner_id: str | None = None
|
|
for pid in player_ids:
|
|
if pid != timed_out_player_id:
|
|
winner_id = pid
|
|
break
|
|
|
|
logger.info(
|
|
f"Turn timeout: game={game_id}, timed_out_player={timed_out_player_id}, "
|
|
f"winner={winner_id}"
|
|
)
|
|
|
|
return await self.end_game(
|
|
game_id=game_id,
|
|
winner_id=winner_id,
|
|
end_reason=GameEndReason.TIMEOUT,
|
|
)
|
|
|
|
async def end_game(
|
|
self,
|
|
game_id: str,
|
|
winner_id: str | None,
|
|
end_reason: GameEndReason,
|
|
) -> GameEndResult:
|
|
"""End a game and record it in history.
|
|
|
|
This method handles the complete game ending process:
|
|
1. Updates game state with winner and end reason
|
|
2. Creates a GameHistory record with replay data
|
|
3. Deletes the game from ActiveGame table and Redis cache
|
|
4. Returns final state for client notification
|
|
|
|
Called by:
|
|
- execute_action when a win condition is detected
|
|
- Timeout system when a player times out
|
|
- Disconnect handler when a player abandons
|
|
|
|
Args:
|
|
game_id: The game ID.
|
|
winner_id: The winner's player ID, or None for a draw.
|
|
end_reason: Why the game ended.
|
|
|
|
Returns:
|
|
GameEndResult with final state and history record ID.
|
|
|
|
Raises:
|
|
GameNotFoundError: If game doesn't exist.
|
|
"""
|
|
state = await self.get_game_state(game_id)
|
|
|
|
# Cancel any active turn timer
|
|
await self._timeout_service.cancel_timer(game_id)
|
|
|
|
# Set winner and end reason on game state
|
|
state.winner_id = winner_id
|
|
state.end_reason = end_reason
|
|
|
|
# Get player IDs from state
|
|
player_ids = list(state.players.keys())
|
|
player1_id = player_ids[0] if len(player_ids) > 0 else None
|
|
player2_id = player_ids[1] if len(player_ids) > 1 else None
|
|
|
|
# Determine loser (opposite of winner)
|
|
loser_id: str | None = None
|
|
if winner_id is not None and len(player_ids) == 2:
|
|
loser_id = player_ids[1] if player_ids[0] == winner_id else player_ids[0]
|
|
|
|
# Get player views for final state
|
|
player1_view = get_visible_state(state, player1_id) if player1_id else None
|
|
player2_view = get_visible_state(state, player2_id) if player2_id else None
|
|
|
|
# Map core end reason to DB end reason
|
|
db_end_reason = _map_end_reason(end_reason)
|
|
|
|
# Determine if NPC won (for campaign games)
|
|
winner_is_npc = False
|
|
db_winner_id: UUID | None = None
|
|
if winner_id is not None:
|
|
# Check if winner is a player UUID or NPC
|
|
try:
|
|
db_winner_id = UUID(winner_id)
|
|
except ValueError:
|
|
# Winner ID is not a UUID, must be NPC
|
|
winner_is_npc = True
|
|
db_winner_id = None
|
|
|
|
# Build replay data from action log
|
|
replay_data = {
|
|
"version": 1,
|
|
"game_id": game_id,
|
|
"action_log": state.action_log,
|
|
"final_turn": state.turn_number,
|
|
"rules": state.rules.model_dump(mode="json"),
|
|
}
|
|
|
|
# Archive to history (creates GameHistory, deletes ActiveGame and cache)
|
|
archive_result = await self._state_manager.archive_to_history(
|
|
game_id=game_id,
|
|
state=state,
|
|
winner_id=db_winner_id,
|
|
winner_is_npc=winner_is_npc,
|
|
end_reason=db_end_reason,
|
|
replay_data=replay_data,
|
|
)
|
|
|
|
logger.info(
|
|
f"Game {game_id} ended: winner={winner_id}, reason={end_reason}, "
|
|
f"turns={archive_result.turn_count}, duration={archive_result.duration_seconds}s"
|
|
)
|
|
|
|
return GameEndResult(
|
|
success=True,
|
|
game_id=game_id,
|
|
winner_id=winner_id,
|
|
loser_id=loser_id,
|
|
end_reason=end_reason,
|
|
turn_count=archive_result.turn_count,
|
|
duration_seconds=archive_result.duration_seconds,
|
|
player1_final_view=player1_view,
|
|
player2_final_view=player2_view,
|
|
history_id=archive_result.history_id,
|
|
message=f"Game ended: {end_reason.value}",
|
|
)
|
|
|
|
# =========================================================================
|
|
# Game Creation
|
|
# =========================================================================
|
|
|
|
def _cards_to_instances(
|
|
self,
|
|
cards: list,
|
|
player_id: str,
|
|
prefix: str = "card",
|
|
) -> list[CardInstance]:
|
|
"""Convert CardDefinitions to CardInstances with unique IDs.
|
|
|
|
Args:
|
|
cards: List of CardDefinition objects.
|
|
player_id: The owning player's ID (used in instance ID).
|
|
prefix: Prefix for instance IDs (e.g., "main" or "energy").
|
|
|
|
Returns:
|
|
List of CardInstance objects with unique instance_ids.
|
|
"""
|
|
instances = []
|
|
for i, card_def in enumerate(cards):
|
|
instance = CardInstance(
|
|
instance_id=f"{player_id}-{prefix}-{i}-{uuid_module.uuid4().hex[:8]}",
|
|
definition_id=card_def.id,
|
|
)
|
|
instances.append(instance)
|
|
return instances
|
|
|
|
def _build_card_registry(
|
|
self,
|
|
*card_lists: list,
|
|
) -> dict[str, Any]:
|
|
"""Build a card registry from card definition lists.
|
|
|
|
Only includes cards that are actually in the decks, not all cards.
|
|
|
|
Args:
|
|
card_lists: Variable number of CardDefinition lists.
|
|
|
|
Returns:
|
|
Dict mapping definition_id -> CardDefinition.
|
|
"""
|
|
registry: dict[str, Any] = {}
|
|
for cards in card_lists:
|
|
for card_def in cards:
|
|
if card_def.id not in registry:
|
|
registry[card_def.id] = card_def
|
|
return registry
|
|
|
|
async def create_game(
|
|
self,
|
|
player1_id: UUID,
|
|
player2_id: UUID,
|
|
deck1_id: UUID,
|
|
deck2_id: UUID,
|
|
rules_config: RulesConfig,
|
|
deck_service: DeckService,
|
|
game_type: GameType = GameType.FREEPLAY,
|
|
game_id: str | None = None,
|
|
) -> GameCreateResult:
|
|
"""Create a new game between two players.
|
|
|
|
Loads decks via DeckService, converts cards to instances, initializes
|
|
game state via GameEngine, and persists to both Redis and Postgres.
|
|
|
|
IMPORTANT: rules_config is required and comes from the frontend
|
|
request, not from server-side defaults.
|
|
|
|
Args:
|
|
player1_id: First player's UUID.
|
|
player2_id: Second player's UUID.
|
|
deck1_id: First player's deck UUID.
|
|
deck2_id: Second player's deck UUID.
|
|
rules_config: Game rules from the frontend request.
|
|
deck_service: DeckService instance for loading decks (from API layer).
|
|
game_type: Type of game (freeplay, ranked, etc.). Defaults to FREEPLAY.
|
|
game_id: Optional game ID. Auto-generated if not provided.
|
|
|
|
Returns:
|
|
GameCreateResult with game_id, initial player views, and starting player.
|
|
|
|
Raises:
|
|
GameCreationError: If deck loading or game creation fails.
|
|
"""
|
|
p1_str = str(player1_id)
|
|
p2_str = str(player2_id)
|
|
|
|
# Load decks via DeckService
|
|
try:
|
|
# get_deck_for_game returns list[CardDefinition] expanded for quantities
|
|
deck1_cards = await deck_service.get_deck_for_game(player1_id, deck1_id)
|
|
deck2_cards = await deck_service.get_deck_for_game(player2_id, deck2_id)
|
|
|
|
# get_deck returns DeckEntry with energy_cards dict
|
|
deck1_entry = await deck_service.get_deck(player1_id, deck1_id)
|
|
deck2_entry = await deck_service.get_deck(player2_id, deck2_id)
|
|
except Exception as e:
|
|
logger.error(f"Failed to load decks: {e}")
|
|
raise GameCreationError(f"Failed to load decks: {e}") from e
|
|
|
|
# Expand energy cards to CardDefinitions
|
|
energy1_cards = []
|
|
for energy_type, qty in deck1_entry.energy_cards.items():
|
|
# Energy card IDs follow pattern: energy-basic-{type}
|
|
energy_id = f"energy-basic-{energy_type}"
|
|
energy_def = self._card_service.get_card(energy_id)
|
|
if energy_def is None:
|
|
raise GameCreationError(
|
|
f"Energy card definition not found: {energy_id}. "
|
|
"Check that card data is loaded correctly."
|
|
)
|
|
energy1_cards.extend([energy_def] * qty)
|
|
|
|
energy2_cards = []
|
|
for energy_type, qty in deck2_entry.energy_cards.items():
|
|
energy_id = f"energy-basic-{energy_type}"
|
|
energy_def = self._card_service.get_card(energy_id)
|
|
if energy_def is None:
|
|
raise GameCreationError(
|
|
f"Energy card definition not found: {energy_id}. "
|
|
"Check that card data is loaded correctly."
|
|
)
|
|
energy2_cards.extend([energy_def] * qty)
|
|
|
|
# Convert CardDefinitions to CardInstances with unique IDs
|
|
deck1_instances = self._cards_to_instances(deck1_cards, p1_str, "main")
|
|
deck2_instances = self._cards_to_instances(deck2_cards, p2_str, "main")
|
|
energy1_instances = self._cards_to_instances(energy1_cards, p1_str, "energy")
|
|
energy2_instances = self._cards_to_instances(energy2_cards, p2_str, "energy")
|
|
|
|
# Build decks dict for GameEngine
|
|
decks = {
|
|
p1_str: deck1_instances,
|
|
p2_str: deck2_instances,
|
|
}
|
|
energy_decks = {
|
|
p1_str: energy1_instances,
|
|
p2_str: energy2_instances,
|
|
}
|
|
|
|
# Build card registry from only the cards in play
|
|
card_registry = self._build_card_registry(
|
|
deck1_cards, deck2_cards, energy1_cards, energy2_cards
|
|
)
|
|
|
|
# Create engine with the provided rules via factory (allows testing)
|
|
engine = self._creation_engine_factory(rules_config)
|
|
|
|
# Create the game via GameEngine
|
|
result: GameCreationResult = engine.create_game(
|
|
player_ids=[p1_str, p2_str],
|
|
decks=decks,
|
|
card_registry=card_registry,
|
|
energy_decks=energy_decks if energy1_instances or energy2_instances else None,
|
|
game_id=game_id,
|
|
)
|
|
|
|
if not result.success or result.game is None:
|
|
logger.error(f"GameEngine failed to create game: {result.message}")
|
|
raise GameCreationError(result.message)
|
|
|
|
game = result.game
|
|
|
|
# Persist to both cache and database
|
|
try:
|
|
await self._state_manager.save_to_cache(game)
|
|
await self._state_manager.persist_to_db(
|
|
game,
|
|
game_type=game_type,
|
|
player1_id=player1_id,
|
|
player2_id=player2_id,
|
|
rules_config=rules_config.model_dump() if rules_config else None,
|
|
)
|
|
except Exception as e:
|
|
logger.error(f"Failed to persist game state: {e}")
|
|
raise GameCreationError(f"Failed to persist game: {e}") from e
|
|
|
|
# NOTE: Turn timer is NOT started here during SETUP phase.
|
|
# Timer starts when SETUP completes (both players select basic pokemon)
|
|
# and the first real turn begins. See execute_action() for timer start logic.
|
|
|
|
# Get player-visible views
|
|
player1_view = get_visible_state(game, p1_str)
|
|
player2_view = get_visible_state(game, p2_str)
|
|
|
|
logger.info(
|
|
f"Created game {game.game_id}: {p1_str} vs {p2_str}, "
|
|
f"starting player: {game.current_player_id}"
|
|
)
|
|
|
|
return GameCreateResult(
|
|
success=True,
|
|
game_id=game.game_id,
|
|
player1_view=player1_view,
|
|
player2_view=player2_view,
|
|
starting_player_id=game.current_player_id,
|
|
message="Game created successfully",
|
|
)
|
|
|
|
|
|
# Global singleton instance
|
|
# Note: This is safe because GameService is stateless regarding game rules.
|
|
# Each game's rules are stored in its GameState, not in this service.
|
|
game_service = GameService()
|