Initial scopone scientifico backend

Multiplayer scopone scientifico backend on the kaya framework:

- OIDC login (kaya-oidc), session-backed WebSocket auth
- Pure rules engine (forced captures, scopa, primiera scoring) with
  full-match simulation tests
- Live game state in Redis (JSON + TTL, join codes, per-game locks,
  pub/sub state push); in-memory fallback for tests
- WebSocket /ws/games/{id} for real-time play; REST lobby endpoints
  (create/join/snapshot) with hidden-hand views
- Finished matches persisted to Postgres (Tortoise + aerich) for match
  history and leaderboard endpoints
- Docker Compose stack: postgres, redis, mock-oauth2-server, db-migrate, app
- 45 tests passing; mypy clean

src/scopa/__init__.py

@@ -0,0 +1 @@
+"""Scopone scientifico backend built on the kaya framework."""

src/scopa/aerich_config.py

@@ -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",
+        }
+    },
+}

src/scopa/app.py

@@ -0,0 +1,75 @@
+"""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,
+        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.
+from .routes import games, health, stats  # noqa: E402,F401
+from . import ws  # noqa: E402,F401

src/scopa/auth.py

@@ -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

src/scopa/config.py

@@ -0,0 +1,54 @@
+"""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
+    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
+
+    @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"),
+            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")),
+        )
+
+
+settings: Settings = Settings.from_env()

src/scopa/game/__init__.py

@@ -0,0 +1 @@
+"""Scopone scientifico domain package."""

src/scopa/game/engine.py

@@ -0,0 +1,372 @@
+"""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,
+    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)
+
+    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")
+        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
+
+    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,
+    }
+    if viewer is not None and state.phase == PHASE_PLAYING and viewer.seat == state.turn:
+        payload["your_turn"] = True
+    return payload

src/scopa/game/errors.py

@@ -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."""

src/scopa/game/state.py

@@ -0,0 +1,187 @@
+"""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 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
+
+    # -- 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,
+        }
+
+    @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"),
+        )
+
+    # -- 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

src/scopa/http.py

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

src/scopa/models.py

@@ -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"),)

src/scopa/openapi.py

@@ -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.",
+    },
+]

src/scopa/pagination.py

@@ -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)

src/scopa/routes/__init__.py

@@ -0,0 +1 @@
+"""HTTP route modules."""

src/scopa/routes/games.py

@@ -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))

src/scopa/routes/health.py

@@ -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",)},
+    )

src/scopa/routes/stats.py

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

src/scopa/stats.py

@@ -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

src/scopa/store.py

@@ -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

src/scopa/tortoise_mixin.py

@@ -0,0 +1,104 @@
+"""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
+        await self._bind()
+        return None
+
+    async def _ensure_ws_context(self, ws: WebSocket):
+        await self._bind()
+        return None

src/scopa/ws.py

@@ -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)