woggioni/tavolo
2
0
Fork 0
Code Actions Packages Releases Activity

Add Sycamore/WASM frontend and restructure into server/ + web/

Browse Source 
Repo is now a monorepo:

- server/: the kaya backend, unchanged in behaviour, plus:
  - GET /api/me for SPA session detection
  - last_move recorded on every play and broadcast in the game state, so
    clients can show who played which card the moment they play it
  - legal_moves per hand card for the player on turn (rules stay
    server-side)
  - static catch-all route serving the compiled SPA with index.html
    fallback; Tortoise context now bound only for /api/* requests
  - configurable OIDC post-login/logout redirects for dev against trunk
- web/: Sycamore 0.9 + WASM frontend (trunk): login via the OIDC flow,
  lobby (create match / join by code), live game page over websocket with
  card images (CC0 woodcut napoletane deck), capture picker, move banner,
  game-over overlay, match history and leaderboard pages
- server/Dockerfile gains a rust+trunk stage building web/dist; the single
  app image serves the SPA; compose builds from the repo root with
  overridable ports/OIDC env

Verified end-to-end against the compose stack: four OIDC logins, game
creation, three joins by code, websocket play with broadcasts to all
players and out-of-turn rejection. 51 backend tests, mypy and cargo tests
all green.
This commit is contained in:
woggioni
2026-09-16 03:14:07 +00:00
parent aa7ac056d3
commit 3583c411c3
104 changed files with 151484 additions and 236 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/src/scopa/__init__.py
+1
View File
@@ -0,0 +1 @@
"""Scopone scientifico backend built on the kaya framework."""
server/src/scopa/aerich_config.py
+23
View File
@@ -0,0 +1,23 @@
"""Tortoise ORM configuration consumed by the aerich CLI.
Kept separate from :mod:`scopa.app` so ``aerich`` can import it without
assembling the whole application (mixins, routes). The database URL comes
from the same :class:`~scopa.config.Settings` the app uses, so the CLI
and the app always point at the same database.
``aerich.models`` is required alongside the app models: it provides the
table aerich uses to track applied migrations.
"""
from __future__ import annotations
from .config import settings
TORTOISE_ORM = {
"connections": {"default": settings.database_url},
"apps": {
"models": {
"models": ["scopa.models", "aerich.models"],
"default_connection": "default",
}
},
}
server/src/scopa/app.py
+80
View File
@@ -0,0 +1,80 @@
"""Application entry point.
Assembles the :class:`~kaya.core.KayaApp` with four mixins:
- :class:`~kaya.session.SessionMixin` (sessions persisted in Redis via
:class:`~kaya.session.redis.RedisSessionStore` when ``REDIS_URL`` is set,
otherwise an in-memory store — e.g. for tests)
- :class:`~kaya.oidc.OIDCMixin` (OIDC login)
- :class:`~scopa.tortoise_mixin.TortoiseMixin` (Postgres match statistics;
skipped for ``/api/health`` and the OpenAPI documentation endpoints)
- :class:`~kaya.openapi.OpenAPIMixin` (serves the OpenAPI document at
``/api/openapi.json`` and a Swagger UI at ``/api/docs``)
Live games are kept in :data:`game_store` (Redis when configured, in-memory
otherwise). Routes and the websocket handlers are registered by importing
their modules at the bottom; imports must happen after ``app`` is built.
"""
from __future__ import annotations
from importlib.metadata import version as _pkg_version
from kaya.core import KayaApp
from kaya.oidc import OIDCConfig, OIDCMixin
from kaya.openapi import OpenAPIMixin
from kaya.session import InMemorySessionStore, SessionMixin, SessionStore
from kaya.session.redis import RedisSessionStore
from redis.asyncio import Redis
from .config import settings
from .store import GameStore, InMemoryGameStore, RedisGameStore
from .tortoise_mixin import TortoiseMixin
session_store: SessionStore
if settings.redis_url is not None:
# Lazy client: no connection is opened until a session is actually
# loaded/saved, so importing this module never requires a live Redis.
session_store = RedisSessionStore(Redis.from_url(settings.redis_url))
game_store: GameStore = RedisGameStore(
Redis.from_url(settings.redis_url, decode_responses=False),
ttl_seconds=settings.game_ttl_seconds,
)
else:
session_store = InMemorySessionStore()
game_store = InMemoryGameStore()
session_mixin = SessionMixin(session_store)
oidc_mixin = OIDCMixin(
OIDCConfig(
issuer=settings.oidc_issuer,
client_id=settings.oidc_client_id,
client_secret=settings.oidc_client_secret,
redirect_uri=settings.oidc_redirect_uri,
post_login_redirect=settings.oidc_post_login_redirect,
post_logout_redirect=settings.oidc_post_logout_redirect,
fetch_userinfo=True,
),
session=session_mixin,
)
openapi_mixin = OpenAPIMixin(
title="scopa",
version=_pkg_version("scopa"),
description="Scopone scientifico multiplayer API",
spec_path="/api/openapi.json",
docs_path="/api/docs",
)
tortoise_mixin = TortoiseMixin(
database_url=settings.database_url,
models_modules=["scopa.models"],
skip_paths=frozenset({"/api/health", "/api/docs", "/api/openapi.json"}),
)
app = KayaApp(mixins=[session_mixin, oidc_mixin, tortoise_mixin, openapi_mixin])
# Register routes by importing modules. Order does not matter; each module
# pulls ``app`` from here and decorates its handlers at import time. The
# static SPA catch-all is registered last and only matches paths no other
# route claimed.
from .routes import games, health, me, stats # noqa: E402,F401
from . import ws # noqa: E402,F401
from .routes import static # noqa: E402,F401
server/src/scopa/auth.py
+59
View File
@@ -0,0 +1,59 @@
"""Authentication helpers on top of the kaya-oidc mixin.
Scopa has no application roles: every authenticated user may create and
join games. Authorization beyond login is game membership, checked against
the live game state in Redis.
"""
from __future__ import annotations
from typing import Any, Callable, Mapping, Optional
from kaya.core import HttpContext, WebSocket
from kaya.oidc import OIDCUser
from .app import oidc_mixin
def get_ws_user(ws: WebSocket) -> Optional[OIDCUser]:
"""Return the authenticated user of a WebSocket connection, if any.
The session mixin injects ``session`` into the websocket wrapper; the
OIDC mixin stores the userinfo there at login. Patched in tests.
"""
session = getattr(ws, "session", None)
if session is None:
return None
claims = session.get("oidc_user")
if not isinstance(claims, Mapping):
return None
return OIDCUser(claims)
def display_name(user: OIDCUser) -> str:
"""Best-effort human-readable name for a user."""
for key in ("name", "preferred_username", "email"):
value = user.get(key)
if isinstance(value, str) and value:
return value
return user.sub
def require_auth(handler: Callable[..., Any]) -> Callable[..., Any]:
"""Decorator: gate a handler on being authenticated (any OIDC user).
Responds ``401`` with a JSON error envelope when unauthenticated —
unlike kaya's built-in ``OIDCMixin.require_auth`` which redirects to
the login page (wrong for a JSON API).
"""
async def guarded(ctx: HttpContext, *args: Any, **kwargs: Any) -> None:
if oidc_mixin.get_user(ctx) is None:
await ctx.send_bytes(
401,
b'{"error":"unauthenticated"}',
{"content-type": ("application/json",)},
)
return
await handler(ctx, *args, **kwargs)
return guarded
server/src/scopa/config.py
+65
View File
@@ -0,0 +1,65 @@
"""Environment-driven configuration for the scopa application.
Mirrors kaya's own pattern: read ``os.environ`` directly into a plain
dataclass. No pydantic-settings, no settings module.
"""
from __future__ import annotations
import os
from dataclasses import dataclass
from typing import Optional
def _env(name: str, default: Optional[str] = None) -> str:
value = os.environ.get(name)
if value is None or value == "":
if default is None:
raise RuntimeError(f"Missing required environment variable: {name}")
return default
return value
@dataclass(frozen=True)
class Settings:
database_url: str
oidc_issuer: str
oidc_client_id: str
oidc_client_secret: Optional[str]
oidc_redirect_uri: str
# Where the browser is sent after login/logout. In production the SPA is
# served by this app ("/"); in development point these at the trunk dev
# server (e.g. "http://localhost:8000/").
oidc_post_login_redirect: str
oidc_post_logout_redirect: str
app_host: str
app_port: int
redis_url: Optional[str]
# How long a live game (and its join-code index) survives in Redis
# without activity, in seconds. Defaults to 24h.
game_ttl_seconds: int
# Directory holding the compiled frontend (trunk's dist output),
# served for every path that is not under /api or /auth.
static_dir: str
@staticmethod
def from_env() -> "Settings":
return Settings(
database_url=_env("DATABASE_URL", "postgres://scopa:scopa@localhost:5432/scopa"),
oidc_issuer=_env("OIDC_ISSUER", "http://localhost:8180/scopa"),
oidc_client_id=_env("OIDC_CLIENT_ID", "scopa"),
oidc_client_secret=os.environ.get("OIDC_CLIENT_SECRET"),
oidc_redirect_uri=_env("OIDC_REDIRECT_URI", "http://localhost:8080/auth/callback"),
oidc_post_login_redirect=_env("OIDC_POST_LOGIN_REDIRECT", "/"),
oidc_post_logout_redirect=_env("OIDC_POST_LOGOUT_REDIRECT", "/"),
app_host=_env("APP_HOST", "0.0.0.0"),
app_port=int(_env("APP_PORT", "8080")),
# When unset, sessions and live games fall back to in-memory
# stores (tests, ephemeral dev). Set to e.g.
# redis://localhost:6379/0 to persist both in Redis.
redis_url=os.environ.get("REDIS_URL"),
game_ttl_seconds=int(_env("GAME_TTL_SECONDS", "86400")),
static_dir=_env("STATIC_DIR", "web/dist"),
)
settings: Settings = Settings.from_env()
server/src/scopa/game/__init__.py
+1
View File
@@ -0,0 +1 @@
"""Scopone scientifico domain package."""
server/src/scopa/game/engine.py
+396
View File
@@ -0,0 +1,396 @@
"""Pure rules engine for scopone scientifico.
Every function here is deterministic and I/O-free: it mutates (or reads)
:class:`~scopa.game.state.GameState` and raises
:class:`~scopa.game.errors.GameError` subclasses on rule violations. This
makes the whole rule set unit-testable without Redis, Postgres or HTTP.
Rules implemented
-----------------
* 40-card Italian deck (4 suits x ranks 1-10), ten cards per player, empty
table at the start of every hand.
* A card captures either a **single card of equal rank** or a **combination
of cards whose ranks sum to its own**. When an equal-ranked card is on the
table that capture is mandatory; the player may not take an alternative
combination instead.
* Emptying the table with a capture is a **scopa** (+1), except on the very
last play of a hand.
* At the end of a hand the remaining table cards go to the player who made
the last capture.
* Hand points: ``carte`` (most captured cards), ``denara`` (most diamond
cards), ``settebello`` (the 7 of diamonds), ``primiera`` (best
seven/five/four/three card of each suit, all four suits required), plus
one point per ``scopa``. Ties on carte/denara/primiera award nothing.
* The match ends when a team reaches the target score with a clear lead; a
tie at or above the target is broken by playing another hand.
"""
from __future__ import annotations
import random
from datetime import datetime, timezone
from itertools import combinations
from typing import Dict, List, Optional, Sequence, Tuple
from .errors import (
AlreadyJoined,
CardNotInHand,
GameFinished,
GameNotStarted,
IllegalMove,
LobbyFull,
NotYourTurn,
)
from .state import (
DEFAULT_TARGET_SCORE,
PHASE_FINISHED,
PHASE_LOBBY,
PHASE_PLAYING,
SUITS,
TEAM_NAMES,
Card,
GameState,
Move,
PlayerState,
parse_card,
)
# Number of cards dealt to each player at the start of a hand.
HAND_SIZE = 10
PLAYERS = 4
# Primiera card values: sevens are best, then sixes, then aces, then the
# remaining ranks in descending order. All of 8/9/10 are worth 10.
PRIMIERA_VALUES: Dict[int, int] = {
7: 21,
6: 18,
1: 16,
5: 15,
4: 14,
3: 13,
2: 12,
8: 10,
9: 10,
10: 10,
}
_rng = random.SystemRandom()
def full_deck() -> List[Card]:
"""Return the 40 cards of the Italian deck in canonical order."""
return [Card(rank=rank, suit=suit) for suit in SUITS for rank in range(1, 11)]
def shuffled_deck(rng: Optional[random.Random] = None) -> List[Card]:
"""Return a shuffled deck. Pass ``rng`` for deterministic tests."""
deck = full_deck()
(rng or _rng).shuffle(deck)
return deck
def legal_captures(table: Sequence[Card], card: Card) -> List[List[Card]]:
"""Return every legal capture (a list of card sets) for ``card``.
If an equal-ranked card is on the table, only those single-card
captures are returned (the rule forbids taking a combination instead).
Otherwise every subset of the table whose ranks sum to ``card.rank`` is
returned.
"""
equal = [c for c in table if c.rank == card.rank]
if equal:
return [[c] for c in equal]
candidates = [c for c in table if c.rank <= card.rank]
captures: List[List[Card]] = []
# A sum-equal capture needs at least two cards (single non-equal cards
# cannot sum to the played card).
for size in range(2, len(candidates) + 1):
for combo in combinations(candidates, size):
if sum(c.rank for c in combo) == card.rank:
captures.append(list(combo))
return captures
def create_game(
game_id: str,
join_code: str,
creator_sub: str,
creator_name: str,
target_score: int = DEFAULT_TARGET_SCORE,
) -> GameState:
"""Create a lobby game with the creator seated first."""
if target_score < 1 or target_score > 100:
raise IllegalMove("target_score must be between 1 and 100")
return GameState(
id=game_id,
join_code=join_code,
creator_sub=creator_sub,
target_score=target_score,
phase=PHASE_LOBBY,
players=[PlayerState(sub=creator_sub, name=creator_name, seat=0)],
created_at=datetime.now(timezone.utc).isoformat(),
)
def join_game(state: GameState, sub: str, name: str) -> None:
"""Seat ``sub`` in the next free chair, starting the match when full."""
if state.phase != PHASE_LOBBY:
raise GameNotStarted("game has already started")
if state.seated(sub):
raise AlreadyJoined("already joined this game")
if len(state.players) >= PLAYERS:
raise LobbyFull("game is full")
seat = len(state.players)
state.players.append(PlayerState(sub=sub, name=name, seat=seat))
if len(state.players) == PLAYERS:
start_game(state)
def start_game(state: GameState) -> None:
"""Deal the first hand and switch the game to playing."""
if len(state.players) != PLAYERS:
raise GameNotStarted("need exactly four players to start")
state.phase = PHASE_PLAYING
_deal_hand(state)
def _deal_hand(state: GameState) -> None:
deck = shuffled_deck()
for player in state.players:
player.hand = []
player.captured = []
player.scope = 0
state.table = []
state.last_taker = None
# Dealer rotates each hand; the first card is played by the player to
# the dealer's left.
state.turn = (state.dealer + 1) % PLAYERS
for offset in range(HAND_SIZE):
for seat in range(PLAYERS):
player = _player_at(state, (state.dealer + 1 + seat) % PLAYERS)
player.hand.append(deck.pop())
def _player_at(state: GameState, seat: int) -> PlayerState:
for player in state.players:
if player.seat == seat:
return player
raise IllegalMove(f"no player in seat {seat}")
def play(
state: GameState,
sub: str,
card_code: str,
capture_codes: Optional[Sequence[str]] = None,
) -> None:
"""Apply one move by the player identified by ``sub``.
``capture_codes`` selects which table cards to capture; it must be a
legal capture (see :func:`legal_captures`) when one exists and empty
otherwise. Raises a :class:`~scopa.game.errors.GameError` subclass on
any violation.
"""
if state.phase == PHASE_FINISHED:
raise GameFinished("the match is over")
if state.phase != PHASE_PLAYING:
raise GameNotStarted("the game has not started yet")
player = state.player_for(sub)
if player is None or player.seat != state.turn:
raise NotYourTurn("it is not your turn")
card = parse_card(card_code)
if card not in player.hand:
raise CardNotInHand(f"card {card.code} is not in your hand")
# Remove the card now so the scopa check below can tell whether this
# was the last play of the hand.
player.hand.remove(card)
requested = [parse_card(c) for c in (capture_codes or [])]
options = legal_captures(state.table, card)
taken: List[Card] = []
scopa = False
if not options:
if requested:
raise IllegalMove("no capture is possible with that card")
state.table.append(card)
else:
chosen = _match_option(options, requested)
if chosen is None:
raise IllegalMove("the requested capture is not legal")
taken = chosen
for captured in chosen:
state.table.remove(captured)
player.captured.append(captured)
player.captured.append(card)
state.last_taker = player.seat
# A scopa scores only if cards remain to be played this hand.
hands_empty = all(not p.hand for p in state.players)
if not state.table and not hands_empty:
player.scope += 1
scopa = True
state.last_move = Move(
seat=player.seat,
name=player.name,
card=card.code,
captured=[c.code for c in taken],
scopa=scopa,
)
if all(not p.hand for p in state.players):
_end_hand(state)
else:
state.turn = (state.turn + 1) % PLAYERS
def _match_option(
options: Sequence[Sequence[Card]], requested: Sequence[Card]
) -> Optional[List[Card]]:
"""Return the option matching ``requested`` exactly, if any."""
wanted = sorted(c.code for c in requested)
if not wanted:
return None
for option in options:
if sorted(c.code for c in option) == wanted:
return list(option)
return None
def _end_hand(state: GameState) -> None:
"""Sweep the table, score the hand and either deal again or finish."""
if state.table and state.last_taker is not None:
taker = _player_at(state, state.last_taker)
taker.captured.extend(state.table)
state.table = []
points, details = hand_points(state)
for team in (0, 1):
state.scores[team] += points[team]
details["hand"] = state.hand_number
details["team_a_points"] = points[0]
details["team_b_points"] = points[1]
state.hand_scores.append(details)
a, b = state.scores
reached = max(a, b) >= state.target_score
if reached and a != b:
state.phase = PHASE_FINISHED
state.winner = 0 if a > b else 1
state.finished_at = datetime.now(timezone.utc).isoformat()
return
state.hand_number += 1
state.dealer = (state.dealer + 1) % PLAYERS
_deal_hand(state)
def primiera_score(captured: Sequence[Card]) -> int:
"""Return the primiera value of a capture pile (0 if a suit is absent)."""
best: Dict[str, int] = {}
for card in captured:
value = PRIMIERA_VALUES[card.rank]
if card.suit not in best or value > best[card.suit]:
best[card.suit] = value
if len(best) < len(SUITS):
return 0
return sum(best.values())
def hand_points(state: GameState) -> Tuple[List[int], Dict[str, object]]:
"""Compute the hand points for both teams (index 0 = team A)."""
piles: List[List[Card]] = [[], []]
scope: List[int] = [0, 0]
for player in state.players:
piles[player.team].extend(player.captured)
scope[player.team] += player.scope
points = [0, 0]
# Carte: most captured cards. Ties award nothing.
cards = [len(piles[0]), len(piles[1])]
if cards[0] != cards[1]:
points[0 if cards[0] > cards[1] else 1] += 1
# Denara: most diamond cards. Ties award nothing.
coins = [
sum(1 for c in piles[t] if c.suit == "D") for t in (0, 1)
]
if coins[0] != coins[1]:
points[0 if coins[0] > coins[1] else 1] += 1
# Settebello: the 7 of diamonds.
settebello = [
any(c.rank == 7 and c.suit == "D" for c in piles[t]) for t in (0, 1)
]
if settebello[0] != settebello[1]:
points[0 if settebello[0] else 1] += 1
# Primiera: highest value, only if the team holds all four suits.
primiera = [primiera_score(piles[t]) for t in (0, 1)]
if primiera[0] != primiera[1]:
points[0 if primiera[0] > primiera[1] else 1] += 1
# Scope: one point each.
points[0] += scope[0]
points[1] += scope[1]
details: Dict[str, object] = {
"cards": {"A": cards[0], "B": cards[1]},
"denara": {"A": coins[0], "B": coins[1]},
"settebello": {"A": settebello[0], "B": settebello[1]},
"primiera": {"A": primiera[0], "B": primiera[1]},
"scope": {"A": scope[0], "B": scope[1]},
}
return points, details
def state_for_player(state: GameState, sub: str) -> Dict[str, object]:
"""Serialize ``state`` hiding other players' hands.
Hands are reduced to a count, except for the requesting player's own
hand. Raises :class:`~scopa.game.errors.GameNotFound`-style access via
the caller; this function assumes ``sub`` may or may not be seated and
simply omits the hand for non-seated viewers.
"""
viewer = state.player_for(sub)
players: List[Dict[str, object]] = []
for player in state.players:
view: Dict[str, object] = {
"sub": player.sub,
"name": player.name,
"seat": player.seat,
"team": TEAM_NAMES[player.team],
"cards_left": len(player.hand),
"captured_count": len(player.captured),
"scope": player.scope,
}
if viewer is not None and viewer.seat == player.seat:
view["hand"] = [c.code for c in player.hand]
players.append(view)
payload: Dict[str, object] = {
"id": state.id,
"join_code": state.join_code,
"phase": state.phase,
"target_score": state.target_score,
"hand_number": state.hand_number,
"dealer": state.dealer,
"turn": state.turn,
"scores": {"A": state.scores[0], "B": state.scores[1]},
"winner": None if state.winner is None else TEAM_NAMES[state.winner],
"table": [c.code for c in state.table],
"players": players,
"last_hand": state.hand_scores[-1] if state.hand_scores else None,
"last_move": state.last_move.to_json() if state.last_move else None,
}
if viewer is not None and state.phase == PHASE_PLAYING and viewer.seat == state.turn:
payload["your_turn"] = True
# Only the player on turn receives their legal captures, so all the
# rule logic stays server-side.
legal_moves: Dict[str, List[List[str]]] = {}
for hand_card in viewer.hand:
options = legal_captures(state.table, hand_card)
if options:
legal_moves[hand_card.code] = [
[c.code for c in option] for option in options
]
payload["legal_moves"] = legal_moves
return payload
server/src/scopa/game/errors.py
+42
View File
@@ -0,0 +1,42 @@
"""Typed errors raised by the scopone engine.
Route/WebSocket handlers translate these into 4xx responses or ``error``
WebSocket messages; the engine itself stays transport-agnostic.
"""
from __future__ import annotations
class GameError(Exception):
"""Base class for every rule/validation failure."""
class IllegalMove(GameError):
"""The requested play violates the rules of scopone scientifico."""
class NotYourTurn(GameError):
"""A player attempted to play out of turn."""
class CardNotInHand(GameError):
"""The played card is not held by the player."""
class GameNotStarted(GameError):
"""An action was attempted before the game left the lobby."""
class GameFinished(GameError):
"""An action was attempted after the match ended."""
class LobbyFull(GameError):
"""A game already has four players."""
class AlreadyJoined(GameError):
"""A player tried to join a game they are already seated in."""
class GameNotFound(GameError):
"""No live game exists for the given id or join code."""
server/src/scopa/game/state.py
+222
View File
@@ -0,0 +1,222 @@
"""In-memory representation of a scopone scientifico game.
The whole mutable game lives in :class:`GameState`, which is serialized to
and from plain JSON for storage in Redis (see :mod:`scopa.store`). Keeping
the representation JSON-native means the store needs no custom codecs and
the state is inspectable with ``redis-cli``.
Deck convention: a 40-card Italian deck. Suits are ``D`` (denari),
``C`` (coppe), ``S`` (spade) and ``B`` (bastoni); ranks are ``1``..``10``.
A card is rendered as ``RRSUIT`` (e.g. ``07D`` is the settebello).
"""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional
SUITS = ("D", "C", "S", "B")
RANKS = tuple(range(1, 11))
# Teams are derived from the seat: seats 0 and 2 form team A (index 0),
# seats 1 and 3 form team B (index 1). Team pairs always sit opposite each
# other, as in real scopone scientifico.
TEAM_A = 0
TEAM_B = 1
TEAM_NAMES = ("A", "B")
PHASE_LOBBY = "lobby"
PHASE_PLAYING = "playing"
PHASE_FINISHED = "finished"
DEFAULT_TARGET_SCORE = 11
def team_of(seat: int) -> int:
return seat % 2
@dataclass(frozen=True)
class Card:
rank: int
suit: str
def __post_init__(self) -> None:
if self.suit not in SUITS:
raise ValueError(f"invalid suit: {self.suit!r}")
if self.rank not in RANKS:
raise ValueError(f"invalid rank: {self.rank!r}")
@property
def code(self) -> str:
return f"{self.rank:02d}{self.suit}"
@staticmethod
def parse(code: str) -> "Card":
code = str(code).upper()
if len(code) != 3 or not code[:2].isdigit():
raise ValueError(f"invalid card code: {code!r}")
return Card(rank=int(code[:2]), suit=code[2])
def to_json(self) -> str:
return self.code
@staticmethod
def from_json(value: Any) -> "Card":
return Card.parse(str(value))
def parse_card(code: Any) -> Card:
"""Parse a card code, raising :class:`ValueError` on malformed input."""
try:
return Card.parse(str(code))
except ValueError:
raise
@dataclass
class Move:
"""Record of a single play, broadcast so every client can show who
played which card and what it captured."""
seat: int
name: str
card: str
captured: List[str] = field(default_factory=list)
scopa: bool = False
def to_json(self) -> Dict[str, Any]:
return {
"seat": self.seat,
"name": self.name,
"card": self.card,
"captured": list(self.captured),
"scopa": self.scopa,
}
@staticmethod
def from_json(data: Dict[str, Any]) -> "Move":
return Move(
seat=int(data["seat"]),
name=str(data["name"]),
card=str(data["card"]),
captured=[str(c) for c in data.get("captured", [])],
scopa=bool(data.get("scopa", False)),
)
@dataclass
class PlayerState:
sub: str
name: str
seat: int
hand: List[Card] = field(default_factory=list)
captured: List[Card] = field(default_factory=list)
scope: int = 0
@property
def team(self) -> int:
return team_of(self.seat)
def to_json(self) -> Dict[str, Any]:
return {
"sub": self.sub,
"name": self.name,
"seat": self.seat,
"hand": [c.to_json() for c in self.hand],
"captured": [c.to_json() for c in self.captured],
"scope": self.scope,
}
@staticmethod
def from_json(data: Dict[str, Any]) -> "PlayerState":
return PlayerState(
sub=str(data["sub"]),
name=str(data["name"]),
seat=int(data["seat"]),
hand=[Card.from_json(c) for c in data.get("hand", [])],
captured=[Card.from_json(c) for c in data.get("captured", [])],
scope=int(data.get("scope", 0)),
)
@dataclass
class GameState:
id: str
join_code: str
creator_sub: str
target_score: int = DEFAULT_TARGET_SCORE
phase: str = PHASE_LOBBY
players: List[PlayerState] = field(default_factory=list)
table: List[Card] = field(default_factory=list)
dealer: int = 0
turn: int = 0
hand_number: int = 1
scores: List[int] = field(default_factory=lambda: [0, 0])
winner: Optional[int] = None
last_taker: Optional[int] = None
# Per-hand points awarded, for a compact audit trail in the API.
hand_scores: List[Dict[str, Any]] = field(default_factory=list)
stats_saved: bool = False
# ISO-8601 timestamps, used when the match result is written to Postgres.
created_at: Optional[str] = None
finished_at: Optional[str] = None
# The most recent play in the current hand, for move announcements.
last_move: Optional[Move] = None
# -- serialization ----------------------------------------------------
def to_json(self) -> Dict[str, Any]:
return {
"id": self.id,
"join_code": self.join_code,
"creator_sub": self.creator_sub,
"target_score": self.target_score,
"phase": self.phase,
"players": [p.to_json() for p in self.players],
"table": [c.to_json() for c in self.table],
"dealer": self.dealer,
"turn": self.turn,
"hand_number": self.hand_number,
"scores": list(self.scores),
"winner": self.winner,
"last_taker": self.last_taker,
"hand_scores": list(self.hand_scores),
"stats_saved": self.stats_saved,
"created_at": self.created_at,
"finished_at": self.finished_at,
"last_move": self.last_move.to_json() if self.last_move else None,
}
@staticmethod
def from_json(data: Dict[str, Any]) -> "GameState":
return GameState(
id=str(data["id"]),
join_code=str(data["join_code"]),
creator_sub=str(data.get("creator_sub", "")),
target_score=int(data.get("target_score", DEFAULT_TARGET_SCORE)),
phase=str(data.get("phase", PHASE_LOBBY)),
players=[PlayerState.from_json(p) for p in data.get("players", [])],
table=[Card.from_json(c) for c in data.get("table", [])],
dealer=int(data.get("dealer", 0)),
turn=int(data.get("turn", 0)),
hand_number=int(data.get("hand_number", 1)),
scores=[int(x) for x in data.get("scores", [0, 0])],
winner=data.get("winner"),
last_taker=data.get("last_taker"),
hand_scores=list(data.get("hand_scores", [])),
stats_saved=bool(data.get("stats_saved", False)),
created_at=data.get("created_at"),
finished_at=data.get("finished_at"),
last_move=Move.from_json(data["last_move"]) if data.get("last_move") else None,
)
# -- helpers ----------------------------------------------------------
def player_for(self, sub: str) -> Optional[PlayerState]:
for player in self.players:
if player.sub == sub:
return player
return None
def seated(self, sub: str) -> bool:
return self.player_for(sub) is not None
server/src/scopa/http.py
+73
View File
@@ -0,0 +1,73 @@
"""JSON helpers for kaya HTTP handlers.
Kaya has no built-in request/response JSON helpers: the request body is an
async byte stream on ``ctx.request_body`` and responses are sent with
``ctx.send_*``. These wrappers handle the boilerplate of draining the body,
parsing JSON, and sending JSON responses.
"""
from __future__ import annotations
import json
from typing import Any, List, Mapping
from kaya.core import HttpContext
JSON_HEADERS = {"content-type": ("application/json",)}
class JsonRequestError(ValueError):
"""Raised by :func:`read_json` when the request body is not valid JSON
or is not a JSON object."""
def extract_query_params(query_string: str) -> Mapping[str, List[str]]:
"""Parse a raw query string into a mapping of param name to list of
values.
Wraps :func:`urllib.parse.parse_qs` so callers don't repeat the
incantation; always returns a mapping (never None).
"""
from urllib.parse import parse_qs
return parse_qs(query_string, keep_blank_values=True)
async def read_json(ctx: HttpContext) -> dict:
"""Drain and parse the request body as JSON.
Returns the parsed ``dict`` on success. Raises :class:`JsonRequestError`
with a short human-readable reason on failure.
"""
body = b""
async for chunk in ctx.request_body:
body += chunk
if not body:
raise JsonRequestError("empty body")
try:
parsed = json.loads(body.decode("utf-8"))
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
raise JsonRequestError("invalid JSON") from exc
if not isinstance(parsed, dict):
raise JsonRequestError("JSON body must be an object")
return parsed
async def read_json_optional(ctx: HttpContext) -> dict:
"""Like :func:`read_json` but treats an empty body as ``{}``."""
try:
return await read_json(ctx)
except JsonRequestError as exc:
if str(exc) == "empty body":
return {}
raise
async def send_json(ctx: HttpContext, status: int, payload: Any) -> None:
"""Send ``payload`` as a JSON response."""
body = json.dumps(payload).encode("utf-8")
await ctx.send_bytes(status, body, JSON_HEADERS)
async def send_error(ctx: HttpContext, status: int, message: str) -> None:
"""Send a JSON error envelope."""
await send_json(ctx, status, {"error": message})
server/src/scopa/models.py
+53
View File
@@ -0,0 +1,53 @@
"""Tortoise ORM models: match statistics persisted in Postgres.
Live game state lives in Redis (see :mod:`scopa.store`); only completed
matches are written here. The two tables answer the question "every match
a player took part in, with the final score":
* :class:`Match` — one row per finished match with both teams' scores.
* :class:`MatchPlayer` — one row per participant, linking an OIDC
``sub`` to a seat/team and whether they won.
"""
from __future__ import annotations
from tortoise import fields
from tortoise.models import Model
class Match(Model):
"""A completed scopone scientifico match."""
id = fields.UUIDField(pk=True)
team_a_score = fields.SmallIntField()
team_b_score = fields.SmallIntField()
# "A" or "B".
winner_team = fields.CharField(max_length=1)
target_score = fields.SmallIntField()
hands_played = fields.SmallIntField()
started_at = fields.DatetimeField()
finished_at = fields.DatetimeField()
players: fields.ReverseRelation["MatchPlayer"]
class Meta:
table = "match"
ordering = ["-finished_at"]
class MatchPlayer(Model):
"""Participation of one user in one match."""
id = fields.UUIDField(pk=True)
match: fields.ForeignKeyRelation[Match] = fields.ForeignKeyField(
"models.Match", related_name="players", on_delete=fields.CASCADE
)
# OIDC subject of the player; no local users table.
user_sub = fields.CharField(max_length=255, db_index=True)
display_name = fields.CharField(max_length=200)
seat = fields.SmallIntField()
team = fields.CharField(max_length=1)
won = fields.BooleanField()
class Meta:
table = "match_player"
unique_together = (("match", "user_sub"),)
server/src/scopa/openapi.py
+21
View File
@@ -0,0 +1,21 @@
"""Shared OpenAPI fragments for the ``@operation`` decorators in ``routes/``."""
from __future__ import annotations
from typing import Any, Dict, List
PAGINATION_PARAMETERS: List[Dict[str, Any]] = [
{
"name": "limit",
"in": "query",
"required": False,
"schema": {"type": "integer", "minimum": 1, "maximum": 100, "default": 20},
"description": "Maximum number of results per page (clamped to [1, 100]).",
},
{
"name": "cursor",
"in": "query",
"required": False,
"schema": {"type": "string"},
"description": "Opaque pagination cursor from a previous response's next_cursor.",
},
]
server/src/scopa/pagination.py
+128
View File
@@ -0,0 +1,128 @@
"""Cursor-based pagination for listing endpoints.
Uses keyset pagination (not OFFSET/LIMIT): each page ends with an opaque
cursor encoding the sort key tuple of the last item on that page; the next
request passes that cursor and the query continues from the point it left
off. This is stable under concurrent inserts and cheaper than OFFSET for
large result sets.
Cursor is a base64-encoded JSON object mapping the sort-field names to the
values of the last item on the previous page. It's opaque to callers and
must be treated as a black box.
"""
from __future__ import annotations
import base64
import json
from dataclasses import dataclass
from typing import Any, Dict, List, Optional, Tuple
from tortoise.queryset import QuerySet
from .http import extract_query_params
DEFAULT_LIMIT = 20
MAX_LIMIT = 100
MIN_LIMIT = 1
CURSOR_PARAM = "cursor"
LIMIT_PARAM = "limit"
class CursorDecodeError(ValueError):
"""Raised when the ``cursor`` query param cannot be decoded."""
def encode_cursor(values: Dict[str, Any]) -> str:
# ``default=str`` handles datetimes (ISO) so keyset cursors can carry
# datetime-typed sort fields (e.g. finished_at for match history).
raw = json.dumps(values, separators=(",", ":"), default=str).encode("utf-8")
return base64.urlsafe_b64encode(raw).rstrip(b"=").decode("ascii")
def decode_cursor(s: Optional[str]) -> Optional[Dict[str, Any]]:
if s is None or s == "":
return None
try:
# Tolerate missing padding.
padded = s + "=" * (-len(s) % 4)
raw = base64.urlsafe_b64decode(padded.encode("ascii"))
obj = json.loads(raw.decode("utf-8"))
except (ValueError, UnicodeDecodeError) as exc:
raise CursorDecodeError("invalid cursor") from exc
if not isinstance(obj, dict):
raise CursorDecodeError("invalid cursor")
return obj
@dataclass(frozen=True)
class Cursor:
limit: int
after: Optional[Dict[str, Any]]
def parse_cursor_params(query_string: str) -> Cursor:
params = extract_query_params(query_string)
limit_raw = params.get(LIMIT_PARAM)
if limit_raw:
try:
limit = int(limit_raw[0])
except ValueError as exc:
raise CursorDecodeError("invalid limit") from exc
else:
limit = DEFAULT_LIMIT
limit = max(MIN_LIMIT, min(MAX_LIMIT, limit))
after = decode_cursor(params.get(CURSOR_PARAM, [None])[0])
return Cursor(limit=limit, after=after)
# A sort field: (model field name, "ASC" or "DESC"). The tuple is the full
# keyset; the cursor encodes exactly these fields.
Sort = List[Tuple[str, str]]
def _keyset_where(sort: Sort, after: Dict[str, Any]):
"""Build a Tortoise ``Q`` filter from a cursor."""
from tortoise.queryset import Q # local to keep import edge narrow
clauses = []
for i, (field, direction) in enumerate(sort):
key = f"{field}__{'gt' if direction == 'ASC' else 'lt'}"
value = after.get(field)
if value is None:
return Q()
clause = Q(**{key: value})
for j in range(i):
prev_field, _ = sort[j]
prev_value = after.get(prev_field)
if prev_value is None:
return Q()
clause = clause & Q(**{prev_field: prev_value})
clauses.append(clause)
result = clauses[0]
for clause in clauses[1:]:
result = result | clause
return result
async def paginate(
queryset: QuerySet,
sort: Sort,
cursor: Cursor,
) -> Tuple[List[Any], Optional[str]]:
"""Return one page of ``queryset`` plus the opaque cursor to continue."""
order_by: List[str] = []
for field, direction in sort:
order_by.append(field if direction == "ASC" else f"-{field}")
qs = queryset.order_by(*order_by)
if cursor.after:
qs = qs.filter(_keyset_where(sort, cursor.after))
rows = await qs.limit(cursor.limit + 1)
if len(rows) <= cursor.limit:
return rows, None
page = rows[: cursor.limit]
last = page[-1]
key: Dict[str, Any] = {}
for field, _ in sort:
key[field] = getattr(last, field)
return page, encode_cursor(key)
server/src/scopa/routes/__init__.py
+1
View File
@@ -0,0 +1 @@
"""HTTP route modules."""
server/src/scopa/routes/games.py
+197
View File
@@ -0,0 +1,197 @@
"""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:`scopa.ws`); these endpoints cover
creation, joining and snapshotting state.
"""
from __future__ import annotations
import secrets
import uuid
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 ..game import engine
from ..game.errors import GameError
from ..game.state import DEFAULT_TARGET_SCORE, PHASE_LOBBY, GameState
from ..http import JsonRequestError, read_json, read_json_optional, send_error, send_json
# 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,
"target_score": state.target_score,
"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.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": {
"target_score": {"type": "integer", "minimum": 1, "maximum": 100},
},
}
}
},
},
responses={
201: {"description": "The created lobby"},
400: {"description": "Invalid 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
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,
)
except GameError as exc:
await send_error(ctx, 400, str(exc))
return
await game_store.save(state)
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:
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:
await send_error(ctx, 409, str(exc))
return
await game_store.save(state)
await game_store.publish(state.id)
if state.phase == PHASE_LOBBY:
await send_json(ctx, 200, _lobby_payload(state))
else:
await send_json(ctx, 200, engine.state_for_player(state, user.sub))
@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))
server/src/scopa/routes/health.py
+19
View File
@@ -0,0 +1,19 @@
"""Liveness probe."""
from __future__ import annotations
from kaya.core import HttpContext
from kaya.openapi import operation
from ..app import app
@app.GET("/api/health")
@operation(summary="Health check",
tags=["health"],
responses={200: {"description": "The service is up"}})
async def health(ctx: HttpContext) -> None:
await ctx.send_bytes(
200,
b'{"status":"ok"}',
{"content-type": ("application/json",)},
)
server/src/scopa/routes/me.py
+25
View File
@@ -0,0 +1,25 @@
"""Whoami endpoint: lets the single-page app detect the login state."""
from __future__ import annotations
from kaya.core import HttpContext
from kaya.openapi import operation
from ..app import app, oidc_mixin
from ..auth import display_name, require_auth
from ..http import send_json
@app.GET("/api/me")
@operation(summary="Current user",
description="Returns the authenticated user's identity from the "
"session; 401 when not logged in.",
tags=["auth"],
responses={
200: {"description": "The current user"},
401: {"description": "Not logged in"},
})
@require_auth
async def me(ctx: HttpContext) -> None:
user = oidc_mixin.get_user(ctx)
assert user is not None # enforced by @require_auth
await send_json(ctx, 200, {"sub": user.sub, "name": display_name(user)})
server/src/scopa/routes/static.py
+79
View File
@@ -0,0 +1,79 @@
"""Static hosting for the compiled single-page application.
In production the kaya backend itself serves the WASM frontend built into
``STATIC_DIR`` (the ``web/dist`` output of ``trunk build --release``; see
the Docker image). A glob catch-all (``/*``) handles every path that did
not match an API or auth route: real files are served with their content
type, anything else falls back to ``index.html`` so client-side routes
(``/game/<id>`` etc.) work on direct loads and refreshes.
kaya-openapi deliberately skips glob routes, so this handler never appears
in the API specification.
"""
from __future__ import annotations
import asyncio
from pathlib import Path
from typing import Mapping
from kaya.core import HttpContext
from ..app import app
from ..config import settings
# Explicit content types: wasm-pack/trunk outputs (.wasm, .js) are not
# consistently covered by the system mime database in slim containers.
_CONTENT_TYPES: Mapping[str, str] = {
".html": "text/html; charset=utf-8",
".js": "text/javascript; charset=utf-8",
".mjs": "text/javascript; charset=utf-8",
".wasm": "application/wasm",
".css": "text/css; charset=utf-8",
".svg": "image/svg+xml",
".png": "image/png",
".ico": "image/x-icon",
".json": "application/json",
".webmanifest": "application/manifest+json",
".woff2": "font/woff2",
}
def _static_root() -> Path:
return Path(settings.static_dir).resolve()
async def _send_path(ctx: HttpContext, target: Path) -> None:
"""Serve a static file, or 404 when it does not exist.
Uses ``send_bytes`` rather than kaya's ``send_file`` (unimplemented by
the ASGI adapter); frontend artifacts are small enough to buffer.
"""
if not target.is_file():
await ctx.send_empty(404)
return
content_type = _CONTENT_TYPES.get(target.suffix.lower(), "application/octet-stream")
body = await asyncio.to_thread(target.read_bytes)
await ctx.send_bytes(200, body, {"content-type": (content_type,)})
@app.GET("/")
async def index(ctx: HttpContext) -> None:
"""Serve the SPA shell at the site root (the glob below cannot match
an empty path)."""
await _send_path(ctx, _static_root() / "index.html")
@app.GET("/*", recursive=True)
async def spa(ctx: HttpContext, _matched: object = None) -> None:
root = _static_root()
relative = ctx.path.lstrip("/")
target = (root / relative).resolve() if relative else root
# Path-traversal guard: the resolved target must stay inside the dist
# directory.
if root != target and root not in target.parents:
await ctx.send_empty(404)
return
if not target.is_file():
# SPA fallback: unknown paths render the app shell.
target = root / "index.html"
await _send_path(ctx, target)
server/src/scopa/routes/stats.py
+110
View File
@@ -0,0 +1,110 @@
"""Player statistics endpoints, served from Postgres.
Every finished match is persisted by :func:`scopa.stats.save_match_result`.
These endpoints expose a player's own match history and a global
leaderboard aggregated from the same two tables.
"""
from __future__ import annotations
from typing import Any, Dict, List
from kaya.core import HttpContext
from kaya.openapi import operation
from ..app import app, oidc_mixin
from ..auth import require_auth
from ..http import send_error, send_json
from ..models import Match, MatchPlayer
from ..openapi import PAGINATION_PARAMETERS
from ..pagination import CursorDecodeError, paginate, parse_cursor_params
async def _serialize_match(match: Match, viewer: str) -> Dict[str, Any]:
participants = await MatchPlayer.filter(match_id=match.id).order_by("seat")
return {
"id": str(match.id),
"team_a_score": match.team_a_score,
"team_b_score": match.team_b_score,
"winner_team": match.winner_team,
"target_score": match.target_score,
"hands_played": match.hands_played,
"started_at": match.started_at.isoformat(),
"finished_at": match.finished_at.isoformat(),
"you_won": any(p.user_sub == viewer and p.won for p in participants),
"players": [
{
"user_sub": p.user_sub,
"display_name": p.display_name,
"seat": p.seat,
"team": p.team,
"won": p.won,
}
for p in participants
],
}
@app.GET("/api/me/matches")
@operation(summary="List my matches",
description="Cursor-paginated history of finished matches the caller "
"played, newest first, with the final score.",
tags=["stats"],
parameters=PAGINATION_PARAMETERS,
responses={
200: {"description": "A page of matches"},
400: {"description": "Invalid pagination cursor"},
401: {"description": "Authentication required"},
})
@require_auth
async def my_matches(ctx: HttpContext) -> None:
try:
cursor = parse_cursor_params(ctx.query_string)
except CursorDecodeError as exc:
await send_error(ctx, 400, str(exc))
return
user = oidc_mixin.get_user(ctx)
assert user is not None
queryset = Match.filter(players__user_sub=user.sub).distinct()
matches, next_cursor = await paginate(
queryset, [("finished_at", "DESC"), ("id", "ASC")], cursor
)
results = [await _serialize_match(m, user.sub) for m in matches]
await send_json(ctx, 200, {"results": results, "next_cursor": next_cursor})
@app.GET("/api/leaderboard")
@operation(summary="Global leaderboard",
description="Aggregated wins, matches played and team points for every "
"player with at least one finished match. Sorted by wins.",
tags=["stats"],
responses={200: {"description": "The leaderboard"}})
async def leaderboard(ctx: HttpContext) -> None:
rows = await MatchPlayer.all().prefetch_related("match")
aggregate: Dict[str, Dict[str, Any]] = {}
for row in rows:
entry = aggregate.setdefault(
row.user_sub,
{
"user_sub": row.user_sub,
"display_name": row.display_name,
"matches": 0,
"wins": 0,
"points": 0,
},
)
entry["matches"] += 1
entry["wins"] += 1 if row.won else 0
match = row.match
if match is not None:
entry["points"] += (
match.team_a_score if row.team == "A" else match.team_b_score
)
# Keep the most recent display name seen.
entry["display_name"] = row.display_name
ranking: List[Dict[str, Any]] = sorted(
aggregate.values(),
key=lambda e: (e["wins"], e["points"], -e["matches"]),
reverse=True,
)
await send_json(ctx, 200, {"results": ranking})
server/src/scopa/stats.py
+57
View File
@@ -0,0 +1,57 @@
"""Copy finished match results from Redis into Postgres.
Called once when a game reaches the finished phase (guarded by the
``stats_saved`` flag on the state). The write is transactional so a match
never appears with only some of its players.
"""
from __future__ import annotations
import uuid
from datetime import datetime, timezone
from typing import Optional
from tortoise.transactions import in_transaction
from .game.state import PHASE_FINISHED, TEAM_NAMES, GameState
def _parse_timestamp(value: Optional[str]) -> datetime:
if value:
try:
return datetime.fromisoformat(value)
except ValueError:
pass
return datetime.now(timezone.utc)
async def save_match_result(state: GameState) -> None:
"""Persist ``state`` to Postgres if it is finished and not yet saved."""
if state.stats_saved or state.phase != PHASE_FINISHED or state.winner is None:
return
from .models import Match, MatchPlayer
started_at = _parse_timestamp(state.created_at)
finished_at = _parse_timestamp(state.finished_at)
async with in_transaction():
match = await Match.create(
id=uuid.uuid4(),
team_a_score=state.scores[0],
team_b_score=state.scores[1],
winner_team=TEAM_NAMES[state.winner],
target_score=state.target_score,
hands_played=state.hand_number,
started_at=started_at,
finished_at=finished_at,
)
for player in state.players:
await MatchPlayer.create(
id=uuid.uuid4(),
match=match,
user_sub=player.sub,
display_name=player.name,
seat=player.seat,
team=TEAM_NAMES[player.team],
won=player.team == state.winner,
)
state.stats_saved = True
server/src/scopa/store.py
+187
View File
@@ -0,0 +1,187 @@
"""Persistence for live games.
Game state is small, mutable and short-lived, which makes Redis a natural
fit: the whole match is a single JSON value under ``scopa:game:<id>`` with
a sliding TTL, and a join-code index maps the short code a player shares to
that id. Completed matches are copied to Postgres (see
:mod:`scopa.models`); Redis keeps serving the finished state until it
expires.
Two implementations satisfy the same interface:
* :class:`RedisGameStore` — production, used when ``REDIS_URL`` is set.
* :class:`InMemoryGameStore` — tests and ephemeral dev, used otherwise.
Concurrency is handled with a per-game lock so two simultaneous plays
cannot interleave. State changes are broadcast on a per-game pub/sub
channel as a simple "something changed" signal; every open websocket
reloads the state and renders the personalized view. Publishing only a
signal (never the state) means updated state reaches connections on every
worker without leaking hidden hands into the channel.
"""
from __future__ import annotations
import asyncio
import contextlib
import json
from abc import ABC, abstractmethod
from typing import AsyncContextManager, AsyncIterator, Dict, Optional, Set
from redis.asyncio import Redis
from .game.state import GameState
GAME_KEY_PREFIX = "scopa:game:"
CODE_KEY_PREFIX = "scopa:code:"
CHANNEL_PREFIX = "scopa:game:"
# Sentinel pushed into in-memory subscriber queues to signal a change.
_BUMP = b"update"
class GameStore(ABC):
"""Abstract persistence + notification layer for live games."""
@abstractmethod
async def load(self, game_id: str) -> Optional[GameState]:
"""Return the live state for ``game_id`` or ``None``."""
@abstractmethod
async def save(self, state: GameState) -> None:
"""Persist ``state``, refreshing its TTL and code index."""
@abstractmethod
async def find_by_code(self, code: str) -> Optional[GameState]:
"""Return the live state for a join ``code`` or ``None``."""
@abstractmethod
def lock(self, game_id: str) -> AsyncContextManager[None]:
"""Async context manager serializing mutations of one game."""
@abstractmethod
def subscribe(self, game_id: str) -> AsyncContextManager[AsyncIterator[None]]:
"""Async context manager yielding an async iterator of change signals."""
@abstractmethod
async def publish(self, game_id: str) -> None:
"""Signal that the state of ``game_id`` changed."""
def _channel(game_id: str) -> str:
return f"{CHANNEL_PREFIX}{game_id}:events"
class RedisGameStore(GameStore):
def __init__(self, redis: Redis, ttl_seconds: int = 86400) -> None:
self._redis = redis
self._ttl = ttl_seconds
def lock(self, game_id: str):
# Lock and state use distinct key names; the lock expires on its own
# if a worker dies mid-mutation.
return self._redis.lock(f"{GAME_KEY_PREFIX}{game_id}:lock",
timeout=10, blocking_timeout=10)
async def load(self, game_id: str) -> Optional[GameState]:
raw = await self._redis.get(f"{GAME_KEY_PREFIX}{game_id}")
if raw is None:
return None
if isinstance(raw, bytes):
raw = raw.decode("utf-8")
return GameState.from_json(json.loads(raw))
async def save(self, state: GameState) -> None:
payload = json.dumps(state.to_json())
async with self._redis.pipeline(transaction=True) as pipe:
pipe.set(f"{GAME_KEY_PREFIX}{state.id}", payload, ex=self._ttl)
pipe.set(f"{CODE_KEY_PREFIX}{state.join_code}", state.id, ex=self._ttl)
await pipe.execute()
async def find_by_code(self, code: str) -> Optional[GameState]:
game_id = await self._redis.get(f"{CODE_KEY_PREFIX}{code.upper()}")
if game_id is None:
return None
if isinstance(game_id, bytes):
game_id = game_id.decode("utf-8")
return await self.load(str(game_id))
@contextlib.asynccontextmanager
async def subscribe(self, game_id: str) -> AsyncIterator[AsyncIterator[None]]:
pubsub = self._redis.pubsub()
await pubsub.subscribe(_channel(game_id))
try:
yield _redis_events(pubsub)
finally:
with contextlib.suppress(Exception):
await pubsub.unsubscribe(_channel(game_id))
await pubsub.aclose()
async def publish(self, game_id: str) -> None:
await self._redis.publish(_channel(game_id), "update")
async def _redis_events(pubsub) -> AsyncIterator[None]:
async for message in pubsub.listen():
if message.get("type") == "message":
yield None
class InMemoryGameStore(GameStore):
"""Process-local store used by tests and when Redis is not configured."""
def __init__(self) -> None:
self._games: Dict[str, GameState] = {}
self._codes: Dict[str, str] = {}
self._locks: Dict[str, asyncio.Lock] = {}
self._subscribers: Dict[str, Set[asyncio.Queue]] = {}
def _lock_for(self, game_id: str) -> asyncio.Lock:
lock = self._locks.get(game_id)
if lock is None:
lock = asyncio.Lock()
self._locks[game_id] = lock
return lock
@contextlib.asynccontextmanager
async def lock(self, game_id: str) -> AsyncIterator[None]:
async with self._lock_for(game_id):
yield
async def load(self, game_id: str) -> Optional[GameState]:
state = self._games.get(game_id)
return GameState.from_json(state.to_json()) if state else None
async def save(self, state: GameState) -> None:
self._games[state.id] = GameState.from_json(state.to_json())
self._codes[state.join_code] = state.id
async def find_by_code(self, code: str) -> Optional[GameState]:
game_id = self._codes.get(code.upper())
if game_id is None:
return None
return await self.load(game_id)
@contextlib.asynccontextmanager
async def subscribe(self, game_id: str) -> AsyncIterator[AsyncIterator[None]]:
queue: asyncio.Queue = asyncio.Queue()
self._subscribers.setdefault(game_id, set()).add(queue)
try:
yield _queue_events(queue)
finally:
subscribers = self._subscribers.get(game_id)
if subscribers is not None:
subscribers.discard(queue)
if not subscribers:
self._subscribers.pop(game_id, None)
async def publish(self, game_id: str) -> None:
for queue in list(self._subscribers.get(game_id, ())):
queue.put_nowait(_BUMP)
async def _queue_events(queue: asyncio.Queue) -> AsyncIterator[None]:
while True:
await queue.get()
yield None
server/src/scopa/tortoise_mixin.py
+109
View File
@@ -0,0 +1,109 @@
"""A :class:`~kaya.core.KayaMixin` that drives the TortoiseORM lifecycle.
kaya calls ``KayaMixin.setup`` / ``shutdown`` synchronously from inside a
running event loop. Tortoise 1.1.7 binds database connections to a
:class:`~tortoise.context.TortoiseContext` looked up via a contextvar, and
kaya dispatches each HTTP request (and WebSocket connection) as a separate
``loop.create_task``, so a context set by an early request does not
automatically reach later requests.
This mixin therefore:
1. Lazily builds a :class:`TortoiseContext` for the active event loop
(rebuilding it if the running loop changes, which happens in tests that
use a fresh ``asyncio.run`` per test).
2. Per HTTP request, binds that context to the current task via the
``_current_context`` contextvar so the handler — running in the same
task as the ``before_request`` hook — sees an active context.
3. Binds the same context at the start of every WebSocket connection
(``before_websocket`` hook), because the match-result write happens at
the end of a WebSocket match. The long-lived connection task keeps the
context for its whole lifetime.
It deliberately avoids the global-fallback singleton
(``_enable_global_fallback``), which Tortoise only allows to be set once
per process and would therefore break across event loops.
Schema management is split by backend: in-memory sqlite databases (the
test suite) get ``generate_schemas`` on every fresh context; Postgres
schemas are owned by aerich migrations and must be applied externally
(``aerich upgrade``, run by the ``db-migrate`` compose service) before
the app serves requests.
"""
from __future__ import annotations
from asyncio import AbstractEventLoop, get_running_loop
from logging import getLogger
from typing import AbstractSet, Optional, Sequence
from kaya.core import HttpContext, KayaApp, KayaMixin, WebSocket
from tortoise.context import TortoiseContext, _current_context
log = getLogger(__name__)
class TortoiseMixin(KayaMixin):
"""Initialize and tear down a per-loop :class:`TortoiseContext`."""
def __init__(self,
database_url: str,
models_modules: Sequence[str],
skip_paths: AbstractSet[str] = frozenset({"/api/health"})) -> None:
self._database_url = database_url
self._models_modules = list(models_modules)
self._skip_paths = skip_paths
self._ctx: "Optional[TortoiseContext]" = None
self._init_loop: "Optional[AbstractEventLoop]" = None
def apply(self, app: KayaApp) -> None:
app.add_before_request_hook(self._ensure_context)
app.add_before_websocket_hook(self._ensure_ws_context)
def setup(self, loop: AbstractEventLoop) -> None:
pass
def shutdown(self, loop: AbstractEventLoop) -> None:
if self._init_loop is loop and self._ctx is not None:
loop.create_task(self._ctx.close_connections())
self._ctx = None
self._init_loop = None
async def _build_context(self) -> TortoiseContext:
ctx = TortoiseContext()
with ctx:
await ctx.init(
db_url=self._database_url,
modules={"models": self._models_modules},
)
# Schema creation is only done for sqlite (in-memory test
# databases). Postgres schemas are managed by aerich migrations
# (applied by the db-migrate compose service / `aerich upgrade`).
if self._database_url.startswith("sqlite"):
await ctx.generate_schemas()
return ctx
async def _bind(self) -> None:
loop = get_running_loop()
if self._init_loop is not loop:
if self._ctx is not None:
# A previous test loop went away; drop its context.
self._ctx = None
self._ctx = await self._build_context()
self._init_loop = loop
assert self._ctx is not None
_current_context.set(self._ctx)
async def _ensure_context(self, ctx: HttpContext):
if ctx.path in self._skip_paths:
return None
# Only API endpoints touch the database: auth callbacks, websocket
# handshakes (handled by _ensure_ws_context) and the static SPA
# catch-all must not pay for a Tortoise context.
if not ctx.path.startswith("/api/"):
return None
await self._bind()
return None
async def _ensure_ws_context(self, ws: WebSocket):
await self._bind()
return None
server/src/scopa/ws.py
+175
View File
@@ -0,0 +1,175 @@
"""WebSocket endpoint for live play.
Clients connect to ``/ws/games/{game_id}`` using their session cookie (the
OIDC login stores the user in the session, which the session mixin loads
onto the websocket). Only seated players are accepted.
Protocol
--------
Server -> client messages are JSON objects with a ``type``:
* ``state`` — the personalized game view (own hand visible, others hidden).
* ``game_over`` — sent once when the match ends, with the final scores.
* ``error`` — a rejected action or malformed message.
Client -> server messages are JSON objects::
{"action": "play", "card": "07D", "capture": ["02D", "05C"]}
{"action": "play", "card": "07D"}
{"action": "state"}
``capture`` lists the table cards to take and must be a legal capture when
one exists (see :func:`scopa.game.engine.legal_captures`); it is omitted
when the played card cannot capture.
Mutations run under the per-game lock; after a successful move the new
state is saved to Redis and a change signal is published. Every connected
websocket is subscribed to that signal and re-renders the state, so all
players see the move immediately (and consistently across workers).
"""
from __future__ import annotations
import asyncio
import json
from contextlib import suppress
from typing import Any, Awaitable, Callable, Dict, Optional
from kaya.core import WebSocket
from . import auth
from .app import app, game_store
from .game import engine
from .game.errors import GameError
from .game.state import PHASE_FINISHED, GameState
from .stats import save_match_result
Send = Callable[[Dict[str, Any]], Awaitable[None]]
def _error(message: str, code: str = "invalid") -> Dict[str, Any]:
return {"type": "error", "code": code, "message": message}
def _state_message(state: GameState, sub: str) -> Dict[str, Any]:
return {"type": "state", "game": engine.state_for_player(state, sub)}
@app.websocket("/ws/games/${game_id}")
async def game_socket(ws: WebSocket, game_id: str) -> None:
user = auth.get_ws_user(ws)
if user is None:
await ws.close(4401)
return
state = await game_store.load(game_id)
if state is None:
await ws.close(4404)
return
if not state.seated(user.sub):
await ws.close(4403)
return
await ws.accept()
send_lock = asyncio.Lock()
async def send(payload: Dict[str, Any]) -> None:
async with send_lock:
await ws.send_text(json.dumps(payload))
await send(_state_message(state, user.sub))
async with game_store.subscribe(game_id) as events:
forward = asyncio.create_task(
_forward(events, game_id, user.sub, send)
)
try:
async for message in ws:
if message.kind == "close":
break
if message.kind != "text" or not isinstance(message.data, str):
await send(_error("expected a text frame with a JSON object"))
continue
await _handle_message(send, game_id, user.sub, message.data)
finally:
forward.cancel()
with suppress(asyncio.CancelledError):
await forward
async def _forward(
events,
game_id: str,
sub: str,
send: Send,
) -> None:
async for _ in events:
state = await game_store.load(game_id)
if state is None:
return
await send(_state_message(state, sub))
if state.phase == PHASE_FINISHED:
await send(
{
"type": "game_over",
"scores": {"A": state.scores[0], "B": state.scores[1]},
"winner": "A" if state.winner == 0 else "B",
}
)
return
async def _handle_message(send: Send, game_id: str, sub: str, raw: str) -> None:
try:
data = json.loads(raw)
except (ValueError, TypeError):
await send(_error("invalid JSON"))
return
if not isinstance(data, dict):
await send(_error("message must be a JSON object"))
return
action = data.get("action")
if action == "play":
await _handle_play(send, game_id, sub, data)
elif action in ("state", "sync"):
state = await game_store.load(game_id)
if state is not None:
await send(_state_message(state, sub))
else:
await send(_error(f"unknown action: {action!r}"))
async def _handle_play(
send: Send, game_id: str, sub: str, data: Dict[str, Any]
) -> None:
card = data.get("card")
capture = data.get("capture")
if not isinstance(card, str):
await send(_error("'card' must be a card code string"))
return
if capture is not None and (
not isinstance(capture, list)
or any(not isinstance(item, str) for item in capture)
):
await send(_error("'capture' must be a list of card codes"))
return
async with game_store.lock(game_id):
state = await game_store.load(game_id)
if state is None:
await send(_error("game not found", code="not_found"))
return
try:
engine.play(state, sub, card, capture)
except GameError as exc:
await send(_error(str(exc), code="illegal_move"))
return
except ValueError:
await send(_error("invalid card code", code="illegal_move"))
return
if state.phase == PHASE_FINISHED:
await save_match_result(state)
await game_store.save(state)
await game_store.publish(game_id)