tavolo/server/src/tavolo/ws.py

"""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": "ack"}
    {"action": "state"}

``capture`` lists the table cards to take and must be a legal capture when
one exists (see :func:`tavolo.game.engine.legal_captures`); it is omitted
when the played card cannot capture. ``ack`` acknowledges the hand-end
scoring summary; the next hand is dealt when all four players have
acknowledged or the timeout fires.

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

If a player does not move before the per-game ``turn_timeout``, the server
plays a random card (with a random legal capture when one is required) for
them, so a disconnected or idle player cannot stall the match. The timer is
re-armed by every client connection and state broadcast, and fires
immediately when a reconnect finds the deadline already past.
"""
from __future__ import annotations

import asyncio
import json
from contextlib import suppress
from datetime import datetime, timezone
from logging import getLogger
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, PHASE_HAND_END, PHASE_PLAYING, GameState
from .stats import save_match_result

log = getLogger(__name__)

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:
        log.debug("websocket %s rejected: no authenticated user", game_id)
        await ws.close(4401)
        return

    state = await game_store.load(game_id)
    if state is None:
        log.debug("websocket rejected: unknown game %s", game_id)
        await ws.close(4404)
        return
    if not state.seated(user.sub):
        log.debug("websocket %s rejected: %s is not seated", game_id, user.sub)
        await ws.close(4403)
        return

    await ws.accept()
    log.info("%s connected to game %s", user.sub, game_id)

    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))
    schedule_turn_timer(game_id, state)

    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
            log.debug("%s disconnected from game %s", user.sub, game_id)


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
        schedule_turn_timer(game_id, state)
        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):
        log.debug("game %s: malformed message from %s (not JSON)", game_id, sub)
        await send(_error("invalid JSON"))
        return
    if not isinstance(data, dict):
        log.debug("game %s: malformed message from %s (not an object)", game_id, sub)
        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 == "ack":
        await _handle_ack(send, game_id, sub)
    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}"))


# --- hand-end acknowledgement ------------------------------------------------

# Running auto-continue timers, keyed by (game_id, hand_number), so a hand's
# timeout is scheduled only once even when several clients are connected.
_hand_end_timers: Dict[tuple, asyncio.Task] = {}


async def _handle_ack(send: Send, game_id: str, sub: str) -> None:
    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.acknowledge_hand(state, sub)
        except GameError as exc:
            await send(_error(str(exc), code="illegal_move"))
            return
        log.debug("game %s: %s acknowledged hand %d", game_id, sub, state.hand_number)
        await game_store.save(state)
        await game_store.publish(game_id)


def schedule_hand_end_timer(game_id: str, hand_number: int, timeout: int) -> None:
    """Deal the next hand after the acknowledgement timeout, even if not
    everyone has clicked. Fizzles if the hand already advanced."""
    key = (game_id, hand_number)
    if key in _hand_end_timers:
        return

    async def _auto_advance() -> None:
        try:
            await asyncio.sleep(timeout)
            async with game_store.lock(game_id):
                state = await game_store.load(game_id)
                if (
                    state is None
                    or state.phase != engine.PHASE_HAND_END
                    or state.hand_number != hand_number
                ):
                    return
                for player in state.players:
                    engine.acknowledge_hand(state, player.sub)
                await game_store.save(state)
                await game_store.publish(game_id)
                log.info(
                    "game %s: hand %d auto-advanced after the acknowledgement timeout",
                    game_id,
                    hand_number,
                )
        finally:
            _hand_end_timers.pop(key, None)

    _hand_end_timers[key] = asyncio.create_task(_auto_advance())


# --- auto-play on turn timeout ------------------------------------------------

# Running turn timers, keyed by (game_id, hand_number, turn, deadline), so a
# turn's timeout is scheduled only once even when several clients are
# connected. Including the deadline means a re-arm after a reconnect cannot
# duplicate a timer for a turn that was already auto-played.
_turn_timers: Dict[tuple, asyncio.Task] = {}


def schedule_turn_timer(game_id: str, state: GameState) -> None:
    """Auto-play a random legal card if the player on turn misses the
    deadline. Fizzles if the turn already advanced."""
    if state.phase != PHASE_PLAYING or not state.turn_deadline:
        return
    key = (game_id, state.hand_number, state.turn, state.turn_deadline)
    if key in _turn_timers:
        return

    hand_number = state.hand_number
    turn = state.turn
    deadline_raw = state.turn_deadline
    try:
        deadline = datetime.fromisoformat(deadline_raw)
    except ValueError:
        return

    async def _auto_play() -> None:
        try:
            delay = (deadline - datetime.now(timezone.utc)).total_seconds()
            await asyncio.sleep(max(delay, 0))
            async with game_store.lock(game_id):
                state = await game_store.load(game_id)
                if (
                    state is None
                    or state.phase != PHASE_PLAYING
                    or state.hand_number != hand_number
                    or state.turn != turn
                    or state.turn_deadline != deadline_raw
                ):
                    # The turn moved on (or the game ended) without this
                    # timer firing: make sure the current turn is armed.
                    if state is not None:
                        schedule_turn_timer(game_id, state)
                    return
                try:
                    engine.auto_play(state)
                except GameError:
                    return
                log.info(
                    "game %s: auto-played for %s (turn timeout, hand %d)",
                    game_id,
                    state.players[turn].sub if turn < len(state.players) else "?",
                    hand_number,
                )
                await _after_play(state, game_id)
        finally:
            _turn_timers.pop(key, None)

    _turn_timers[key] = asyncio.create_task(_auto_play())


async def _after_play(state: GameState, game_id: str) -> None:
    """Persist a successful move and notify every connected player.

    Callers must hold the per-game lock. Handles the two terminal
    transitions: the match result is written to Postgres once, and a
    hand-end summary schedules the auto-continue timeout.
    """
    if state.phase == PHASE_FINISHED:
        await save_match_result(state)
        log.info(
            "game %s finished: team %s wins %d-%d",
            game_id,
            "A" if state.winner == 0 else "B",
            state.scores[0],
            state.scores[1],
        )
    elif state.phase == PHASE_HAND_END:
        schedule_hand_end_timer(game_id, state.hand_number, state.hand_ack_timeout)
    await game_store.save(state)
    await game_store.publish(game_id)


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:
            log.debug("game %s: illegal move by %s: %s", game_id, sub, exc)
            await send(_error(str(exc), code="illegal_move"))
            return
        except ValueError:
            log.debug("game %s: invalid card code from %s: %r", game_id, sub, card)
            await send(_error("invalid card code", code="illegal_move"))
            return

        log.debug("game %s: %s played %s (capture: %s)", game_id, sub, card, capture or "-")
        await _after_play(state, game_id)