"""Game lobby endpoints. A game starts as a lobby: the creator is seated first and shares the six-character ``join_code``. When the fourth player joins, the engine deals the first hand and the match begins. Live play then happens over the ``/ws/games/{id}`` websocket (see :mod:`tavolo.ws`); these endpoints cover creation, joining and snapshotting state. """ from __future__ import annotations import secrets import uuid from logging import getLogger from typing import Any, Dict, Optional from kaya.core import HttpContext from kaya.openapi import operation from .. import auth from ..app import app, game_store, oidc_mixin from ..auth import require_auth from ..config import settings from ..game import engine from ..game.errors import GameError from ..game.state import DEFAULT_TARGET_SCORE, PHASE_LOBBY, GameState from ..games import GAME_TYPES, get_game_type from ..http import JsonRequestError, read_json, read_json_optional, send_error, send_json log = getLogger(__name__) # Join codes avoid characters that are easy to confuse when read aloud. _CODE_ALPHABET = "ABCDEFGHJKLMNPQRSTUVWXYZ23456789" _CODE_LENGTH = 6 _MAX_CODE_ATTEMPTS = 20 def _now_code() -> str: return "".join(secrets.choice(_CODE_ALPHABET) for _ in range(_CODE_LENGTH)) async def _unique_code() -> str: for _ in range(_MAX_CODE_ATTEMPTS): code = _now_code() if await game_store.find_by_code(code) is None: return code raise RuntimeError("could not allocate a unique join code") def _lobby_payload(state: GameState) -> Dict[str, Any]: return { "id": state.id, "join_code": state.join_code, "game_type": state.game_type, "target_score": state.target_score, "napola": state.napola, "phase": state.phase, "players": [ {"sub": p.sub, "name": p.name, "seat": p.seat, "team": "A" if p.seat % 2 == 0 else "B"} for p in state.players ], "seats_open": 4 - len(state.players), } @app.GET("/api/game-types") @operation(summary="List available games", description="Every card game the platform can host, for the " "match-creation dropdown.", tags=["games"], responses={200: {"description": "The available game types"}}) async def list_game_types(ctx: HttpContext) -> None: await send_json(ctx, 200, { "results": [ {"id": g.id, "name": g.name, "description": g.description} for g in GAME_TYPES.values() ] }) @app.POST("/api/games") @operation(summary="Create a game", description="Creates a lobby game and seats the caller in seat 0. " "Share the returned join_code with three other players.", tags=["games"], request_body={ "required": False, "content": { "application/json": { "schema": { "type": "object", "properties": { "game_type": { "type": "string", "default": "scopone_scientifico", "description": "One of the ids from GET /api/game-types", }, "target_score": {"type": "integer", "minimum": 1, "maximum": 100}, "napola": { "type": "boolean", "default": True, "description": "Score the napola rule; a full " "denari sweep wins the match", }, }, } } }, }, responses={ 201: {"description": "The created lobby"}, 400: {"description": "Invalid game_type, target_score or body"}, 401: {"description": "Authentication required"}, }) @require_auth async def create_game(ctx: HttpContext) -> None: body: dict = {} try: body = await read_json_optional(ctx) except JsonRequestError as exc: await send_error(ctx, 400, str(exc)) return target_score: Any = body.get("target_score", DEFAULT_TARGET_SCORE) if isinstance(target_score, bool) or not isinstance(target_score, int): await send_error(ctx, 400, "target_score must be an integer") return game_type: Any = body.get("game_type", "scopone_scientifico") if not isinstance(game_type, str) or get_game_type(game_type) is None: await send_error(ctx, 400, f"unknown game_type: {game_type!r}") return napola: Any = body.get("napola", True) if not isinstance(napola, bool): await send_error(ctx, 400, "napola must be a boolean") return user = oidc_mixin.get_user(ctx) assert user is not None # enforced by @require_auth game_id = str(uuid.uuid4()) join_code = await _unique_code() try: state = engine.create_game( game_id=game_id, join_code=join_code, creator_sub=user.sub, creator_name=auth.display_name(user), target_score=target_score, hand_ack_timeout=settings.hand_ack_timeout_seconds, turn_timeout=settings.turn_timeout_seconds, game_type=game_type, napola=napola, ) except GameError as exc: await send_error(ctx, 400, str(exc)) return await game_store.save(state) log.info( "game %s created by %s (%s, target score %d)", game_id, user.sub, game_type, target_score, ) await send_json(ctx, 201, _lobby_payload(state)) @app.POST("/api/games/join") @operation(summary="Join a game by code", description="Seats the caller in the next free chair. Joining as the " "fourth player starts the match.", tags=["games"], request_body={ "required": True, "content": { "application/json": { "schema": { "type": "object", "properties": {"code": {"type": "string"}}, "required": ["code"], } } }, }, responses={ 200: {"description": "Seated; game state (may be playing)"}, 400: {"description": "Missing code"}, 401: {"description": "Authentication required"}, 404: {"description": "Unknown join code"}, 409: {"description": "Already joined or lobby full"}, }) @require_auth async def join_game(ctx: HttpContext) -> None: try: body = await read_json(ctx) except JsonRequestError as exc: await send_error(ctx, 400, str(exc)) return code = body.get("code") if not isinstance(code, str) or not code: await send_error(ctx, 400, "code is required") return user = oidc_mixin.get_user(ctx) assert user is not None existing = await game_store.find_by_code(code) if existing is None: log.debug("join rejected for %s: unknown code %r", user.sub, code) await send_error(ctx, 404, "unknown join code") return async with game_store.lock(existing.id): state = await game_store.load(existing.id) if state is None: await send_error(ctx, 404, "unknown join code") return try: engine.join_game(state, user.sub, auth.display_name(user)) except GameError as exc: log.debug("join rejected for %s in game %s: %s", user.sub, state.id, exc) await send_error(ctx, 409, str(exc)) return await game_store.save(state) await game_store.publish(state.id) seat = next(p.seat for p in state.players if p.sub == user.sub) if state.phase == PHASE_LOBBY: log.info("%s joined game %s (seat %d, %d/4 players)", user.sub, state.id, seat, len(state.players)) else: log.info("%s joined game %s (seat %d); match started", user.sub, state.id, seat) await send_json(ctx, 200, engine.state_for_player(state, user.sub)) return await send_json(ctx, 200, _lobby_payload(state)) @app.GET("/api/games/${game_id}") @operation(summary="Get a game snapshot", description="Only seated players may read a game; other players' " "hands are hidden.", tags=["games"], responses={ 200: {"description": "The personalized game state"}, 401: {"description": "Authentication required"}, 403: {"description": "Not a player in this game"}, 404: {"description": "Game not found"}, }) @require_auth async def get_game(ctx: HttpContext, game_id: str) -> None: state = await game_store.load(game_id) if state is None: await send_error(ctx, 404, "game not found") return user = oidc_mixin.get_user(ctx) assert user is not None if not state.seated(user.sub): await send_error(ctx, 403, "forbidden") return await send_json(ctx, 200, engine.state_for_player(state, user.sub))