tavolo/server/src/tavolo/store.py

"""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 ``tavolo: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:`tavolo.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 logging import getLogger
from typing import AsyncContextManager, AsyncIterator, Dict, Optional, Set

from redis.asyncio import Redis

from .game.state import GameState

log = getLogger(__name__)

GAME_KEY_PREFIX = "tavolo:game:"
CODE_KEY_PREFIX = "tavolo:code:"
CHANNEL_PREFIX = "tavolo: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:
            log.debug("redis load %s: miss", game_id)
            return None
        if isinstance(raw, bytes):
            raw = raw.decode("utf-8")
        log.debug("redis load %s: hit", game_id)
        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()
        log.debug("redis save %s (phase %s, ttl %ds)", state.id, state.phase, self._ttl)

    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")
        log.debug("redis publish %s", game_id)


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