Files
bridzik/api/__init__.py
tim 2c2f07c2ec Apply velvet-table redesign, fix game lifecycle and history bugs
Frontend:
- Dark green/gold "velvet table" visual redesign across the whole app
  (Auth, Lobby, GameList, GameTable, History, GameOver, modals), with
  Playfair Display/DM Sans typography and a centralized Tailwind palette.
- Desktop game table fit-scales to fill the window; mobile gets
  overlapping hand/trick layouts and larger touch-friendly cards.
- Standings sidebar now groups completed rounds by series with a
  per-series subtotal row, struck-through tips on missed bids.
- History page rewritten into a scoreboard-style detail view (player
  totals beside names, series grouped 2-up on desktop / stacked on
  mobile) and gained game names, completed/abandoned status, and a
  button to reopen a prematurely-ended game back into the lobby.

Backend:
- Fix started games being deleted from memory (and vanishing from
  everyone's lobby) when all players disconnect; only `end_game` tears
  down a started game now.
- Fix a crash writing a timezone-aware datetime into the naive
  `ended_at` Postgres column.
- Add `reopen_game`/`restore_game` to un-end a prematurely-ended game
  from history and resume it from the lobby.
- Let any seated player end an abandoned game once the host is
  offline, not just the host, so the game isn't stuck forever.
- Expose SERIES_PER_GAME/ROUNDS_PER_SERIES as named constants on the
  engine so the persistence layer derives game-completion rules from
  bridzik.py instead of re-encoding them.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-07-01 00:11:42 +02:00

587 lines
20 KiB
Python

import json
import os
import uuid
from json import JSONEncoder
import socketio
from bridzik import Bridzik, BridzikException, Card
from db.db import init_db
from api import auth as auth_module, history
from api.auth import AuthError
def _env_bool(name: str, default: bool) -> bool:
val = os.environ.get(name)
if val is None:
return default
return val.lower() in ("1", "true", "yes", "on")
# --- configuration (env-driven, dev-friendly defaults) --------------------
_cors = os.environ.get("CORS_ALLOWED_ORIGINS", "*")
CORS_ALLOWED_ORIGINS = "*" if _cors == "*" else [o.strip() for o in _cors.split(",")]
SIO_LOGGER = _env_bool("SOCKETIO_LOGGER", False)
LOBBY = "lobby" # room every connection joins to receive the public game list
sio = socketio.AsyncServer(
async_mode="asgi",
cors_allowed_origins=CORS_ALLOWED_ORIGINS,
logger=SIO_LOGGER,
engineio_logger=SIO_LOGGER,
)
async def _health_app(scope, receive, send):
"""Minimal ASGI handler for non-socket.io HTTP routes (liveness checks)."""
if scope["type"] == "lifespan":
while True:
message = await receive()
if message["type"] == "lifespan.startup":
await init_db()
await _restore_unfinished_games()
await send({"type": "lifespan.startup.complete"})
elif message["type"] == "lifespan.shutdown":
await send({"type": "lifespan.shutdown.complete"})
return
if scope["type"] == "http":
ok = scope.get("path", "") in ("/health", "/healthz")
status = 200 if ok else 404
body = b"ok" if ok else b"not found"
await send({"type": "http.response.start", "status": status,
"headers": [(b"content-type", b"text/plain")]})
await send({"type": "http.response.body", "body": body})
# Run with: uvicorn api:app --host 0.0.0.0 --port 5000
app = socketio.ASGIApp(sio, other_asgi_app=_health_app)
# --- in-memory state ------------------------------------------------------
# Single-process only. For multi-worker deployments this moves to Redis
# (socketio.AsyncRedisManager) plus a shared game store.
games: dict[str, "Game"] = {}
# Maps a live connection (sid) to the seat it controls: {"gid": str, "order": int}.
# This is the source of truth for "who is acting" — never trust a client-supplied
# player number.
sessions: dict[str, dict] = {}
# Maps a live connection (sid) to its authenticated account: {"player_id": int,
# "username": str}. Set on login/confirm or on connect via the auth token.
# Required before a connection may create or join a game.
accounts: dict[str, dict] = {}
class Game:
def __init__(self, gid: str, name: str):
self.gid = gid
self.name = name
self.players: list["Player"] = []
self.started = False
self.bridzik_core: Bridzik | None = None
def start(self):
self.bridzik_core = Bridzik()
self.started = True
def player_by_token(self, token: str) -> "Player | None":
return next((p for p in self.players if p.token == token), None)
def player_by_sid(self, sid: str) -> "Player | None":
return next((p for p in self.players if p.sid == sid), None)
def player_by_order(self, order: int) -> "Player | None":
return next((p for p in self.players if p.order == order), None)
class Player:
def __init__(self, sid: str, name: str, order: int, player_id: int):
self.sid = sid
self.name = name # display name == account username
self.order = order
self.player_id = player_id # persistent account id (db.models.Player.id)
self.token = str(uuid.uuid4()) # secret token used for secure reconnect
self.connected = True
class CardStatusEncoder(JSONEncoder):
"""Serializes the engine status, which may contain Card objects."""
def default(self, obj):
if isinstance(obj, Card):
return {"color": obj.color.name, "value": obj.value.name}
return JSONEncoder.default(self, obj)
def public_games() -> list:
"""Public lobby view — no sids, no reconnect tokens."""
return [
{
"gid": g.gid,
"name": g.name,
"started": g.started,
"players": [
{
"order": p.order,
"name": p.name,
"connected": p.connected,
"player_id": p.player_id,
}
for p in g.players
],
}
for g in games.values()
]
# --- emit helpers ---------------------------------------------------------
async def broadcast_lobby():
await sio.emit("get_games", {"games": public_games()}, room=LOBBY)
async def send_game_status(gid: str):
game = games[gid]
core = game.bridzik_core
last_round = core.series[-1].get_last_round()
status = json.loads(json.dumps(core.get_status(), cls=CardStatusEncoder))
# Use DB-backed standings so the score is correct even after a server restart
# (the engine only knows rounds completed since restart).
status["standings"], status["standings_guesses"] = await history.get_standings(gid)
await sio.emit(
"game_status",
{
"gid": gid,
"completed": core.is_completed(),
# Self-contained roster so the game view doesn't depend on the lobby snapshot.
"players": [
{"order": p.order, "name": p.name, "connected": p.connected}
for p in sorted(game.players, key=lambda p: p.order)
],
"series_number": core.series[-1].series_number,
"round_number": last_round.round_number,
"cards_in_round": 8 - last_round.round_number, # tricks == max bid
"status": status,
},
room=gid,
)
async def send_player_cards(gid: str, order: int, to: str):
core = games[gid].bridzik_core
await sio.emit(
"player_cards",
{"cards": json.loads(json.dumps(core.get_player_cards(int(order)), cls=Card.JSONEncoder))},
to=to,
)
async def send_error(sid: str, message: str):
await sio.emit("error", {"error": message}, to=sid)
async def _mark_player_offline(game: "Game", player: "Player"):
"""Mark player disconnected. An unstarted game with nobody left is cleaned
up; a started game is kept in memory so it stays in the lobby and can be
resumed (it's torn down only by end_game)."""
player.connected = False
if not any(p.connected for p in game.players) and not game.started:
del games[game.gid]
else:
await sio.emit(
"player_connection",
{"order": player.order, "connected": False},
room=game.gid,
)
def _active_game(sid: str) -> "tuple[Game, dict] | None":
"""Resolve the started game and seat for a connection, or None."""
sess = sessions.get(sid)
if sess is None:
return None
game = games.get(sess["gid"])
if game is None or not game.started:
return None
return game, sess
async def _restore_unfinished_games():
"""Po starte servera obnov rozohrate hry z DB do pamate (hraci offline).
Karty su rozdane nanovo (pozicia z `series`/`round`); hraci sa vratia cez
`rejoin_game` podla svojej trvalej identity (per-hra tokeny restart neprezili).
"""
for info in await history.get_unfinished_games():
if info["gid"] in games:
continue
_load_game_into_memory(info)
def _load_game_into_memory(info: dict) -> "Game":
"""Postav in-memory Game z restore-info (gid/name/seats/core), hraci offline,
a vlozi ju do `games`. Pouzite pri starte aj pri obnove hry z historie."""
game = Game(info["gid"], info["name"])
game.bridzik_core = info["core"]
game.started = True
for seat, (pid, uname) in enumerate(info["seats"]):
player = Player(None, uname, seat, pid)
player.connected = False
game.players.append(player)
games[info["gid"]] = game
return game
# --- connection lifecycle -------------------------------------------------
@sio.event
async def connect(sid, environ, auth=None):
await sio.enter_room(sid, LOBBY)
# Auto-login via the session token the client stored after a previous login.
token = auth.get("token") if isinstance(auth, dict) else None
identity = await auth_module.player_by_token(token)
if identity is not None:
accounts[sid] = identity
await sio.emit("login", {"player": identity}, to=sid)
await sio.emit("get_games", {"games": public_games()}, to=sid)
@sio.event
async def disconnect(sid):
accounts.pop(sid, None)
sess = sessions.pop(sid, None)
game = games.get(sess["gid"]) if sess else None
if game is not None:
player = game.player_by_sid(sid)
if player is not None:
await _mark_player_offline(game, player)
await broadcast_lobby()
# --- authentication (TOTP) ------------------------------------------------
@sio.on("register_account")
async def register_account(sid, username):
try:
data = await auth_module.register_account(username)
except AuthError as exc:
return await send_error(sid, str(exc))
# otpauth_uri -> the client renders it as a QR code to scan into the app.
await sio.emit("register_account", data, to=sid)
@sio.on("confirm_account")
async def confirm_account(sid, username, code):
try:
identity = await auth_module.confirm_account(username, code)
except AuthError as exc:
return await send_error(sid, str(exc))
accounts[sid] = {"player_id": identity["player_id"], "username": identity["username"]}
await sio.emit("login", {"player": accounts[sid], "token": identity["token"]}, to=sid)
@sio.on("login")
async def login(sid, username, code):
try:
identity = await auth_module.login(username, code)
except AuthError as exc:
return await send_error(sid, str(exc))
accounts[sid] = {"player_id": identity["player_id"], "username": identity["username"]}
await sio.emit("login", {"player": accounts[sid], "token": identity["token"]}, to=sid)
# --- lobby ----------------------------------------------------------------
@sio.on("create_game")
async def create_game(sid, name):
if sid not in accounts:
return await send_error(sid, "Musíte byť prihlásený.")
gid = str(uuid.uuid4())
games[gid] = Game(gid, name)
await sio.emit("create_game", {"gid": gid}, to=sid)
await broadcast_lobby()
@sio.on("get_games")
async def get_games(sid, *args):
await sio.emit("get_games", {"games": public_games()}, to=sid)
@sio.on("register_player")
async def register_player(sid, gid):
account = accounts.get(sid)
if account is None:
return await send_error(sid, "Musíte byť prihlásený.")
if sid in sessions:
return await send_error(sid, "Uz ste v hre.")
game = games.get(gid)
if game is None:
return await send_error(sid, "Hra neexistuje.")
if game.started:
return await send_error(sid, "Hra uz zacala.")
if len(game.players) >= 4:
return await send_error(sid, "Prekroceny pocet hracov.")
if any(p.player_id == account["player_id"] for p in game.players):
return await send_error(sid, "Uz ste v tejto hre.")
# Lowest free seat (robust if someone left the lobby before start).
used = {p.order for p in game.players}
order = next(o for o in range(4) if o not in used)
player = Player(sid, account["username"], order, account["player_id"])
game.players.append(player)
sessions[sid] = {"gid": gid, "order": order}
await sio.enter_room(sid, gid)
# The token is private to this player and required for a secure reconnect.
await sio.emit(
"register_player",
{"player": {"order": order, "name": player.name}, "token": player.token},
to=sid,
)
await broadcast_lobby()
@sio.on("leave_game")
async def leave_game(sid):
"""Explicit exit (e.g. a 'Back to lobby' button). The socket stays
connected and remains in the lobby room."""
sess = sessions.pop(sid, None)
if sess is None:
return # not in a game; nothing to do
game = games.get(sess["gid"])
if game is not None:
await sio.leave_room(sid, game.gid)
player = game.player_by_sid(sid)
if game.started:
# Game in progress: keep the seat (reconnect via token still works),
# just mark the player offline.
if player is not None:
await _mark_player_offline(game, player)
else:
# Not started yet: free the seat entirely.
if player is not None:
game.players.remove(player)
if not game.players:
del games[game.gid]
await broadcast_lobby()
@sio.on("start_game")
async def start_game(sid, gid):
sess = sessions.get(sid)
if sess is None or sess["gid"] != gid:
return await send_error(sid, "Nie ste v tejto hre.")
if sess["order"] != 0:
return await send_error(sid, "Iba hostitel moze spustit hru.")
game = games.get(gid)
if game is None:
return await send_error(sid, "Hra neexistuje.")
if game.started:
return await send_error(sid, "Hra uz zacala.")
if len(game.players) != 4:
return await send_error(sid, "Nedostatocny pocet hracov.")
game.start()
# Persist the game with its 4 seats (ordered 0..3) so history can attribute guesses.
seated = sorted(game.players, key=lambda p: p.order)
await history.record_game_started(gid, game.name, [p.player_id for p in seated])
await broadcast_lobby()
await send_game_status(gid)
for player in game.players:
await send_player_cards(gid, player.order, player.sid)
@sio.on("end_game")
async def end_game(sid, gid):
"""Any seated player can permanently end a game that won't be finished --
not just the host, so the other players aren't stuck forever if the host
abandons the game. Marks it ended in the DB (so it won't be restored) and
sends everyone back to the lobby."""
sess = sessions.get(sid)
if sess is None or sess["gid"] != gid:
return await send_error(sid, "Nie ste v tejto hre.")
game = games.get(gid)
if game is None:
return await send_error(sid, "Hra neexistuje.")
await history.mark_game_ended(gid)
# Notify the room first (while players are still in it), then tear it down.
await sio.emit("game_ended", {"gid": gid}, room=gid)
for player in game.players:
if player.sid:
sessions.pop(player.sid, None)
await sio.leave_room(player.sid, gid)
del games[gid]
await broadcast_lobby()
@sio.on("reconnect_to_game")
async def reconnect_to_game(sid, gid, token):
# Best-effort background reconnect: fail silently (no error toast). After a
# server restart the old token is gone -> the user rejoins from the lobby.
game = games.get(gid)
if game is None:
return
player = game.player_by_token(token)
if player is None:
return
old_sid = player.sid
if old_sid != sid:
sessions.pop(old_sid, None)
player.sid = sid
player.connected = True
sessions[sid] = {"gid": gid, "order": player.order}
await sio.enter_room(sid, gid)
await sio.emit(
"register_player",
{"player": {"order": player.order, "name": player.name}, "token": player.token},
to=sid,
)
if game.started:
await send_game_status(gid)
await send_player_cards(gid, player.order, sid)
await sio.emit("player_connection", {"order": player.order, "connected": True}, room=gid)
await broadcast_lobby()
@sio.on("rejoin_game")
async def rejoin_game(sid, gid):
"""Re-seat into a game via the logged-in account (used after a server restart,
when per-game reconnect tokens are gone). Identity comes from the session."""
account = accounts.get(sid)
if account is None:
return await send_error(sid, "Musíte byť prihlásený.")
if sid in sessions:
return await send_error(sid, "Uz ste v hre.")
game = games.get(gid)
if game is None:
return await send_error(sid, "Hra neexistuje.")
player = next(
(p for p in game.players if p.player_id == account["player_id"]), None
)
if player is None:
return await send_error(sid, "Nie ste hracom tejto hry.")
old_sid = player.sid
if old_sid and old_sid != sid:
sessions.pop(old_sid, None)
player.sid = sid
player.connected = True
sessions[sid] = {"gid": gid, "order": player.order}
await sio.enter_room(sid, gid)
await sio.emit(
"register_player",
{"player": {"order": player.order, "name": player.name}, "token": player.token},
to=sid,
)
if game.started:
await send_game_status(gid)
await send_player_cards(gid, player.order, sid)
await sio.emit("player_connection", {"order": player.order, "connected": True}, room=gid)
await broadcast_lobby()
@sio.on("restore_game")
async def restore_game(sid, gid):
"""Obnov predcasne ukoncenu hru z historie spat do lobby. Smie ju vyvolat
iba hrac danej hry; v lobby sa potom objavi ako rozohrata a clenovia sa
pripoja cez `rejoin_game`."""
account = accounts.get(sid)
if account is None:
return await send_error(sid, "Musíte byť prihlásený.")
if gid in games:
# Uz je v pamati (lobby) -- staci obnovit zoznam hier u klienta.
await sio.emit("game_restored", {"gid": gid}, to=sid)
return await sio.emit("get_games", {"games": public_games()}, to=sid)
info = await history.reopen_game(gid, account["player_id"])
if info is None:
return await send_error(sid, "Hru sa nepodarilo obnovit.")
_load_game_into_memory(info)
await sio.emit("game_restored", {"gid": gid}, to=sid)
await broadcast_lobby()
# --- in-game actions (seat derived from the connection, never the client) -
@sio.on("game_status")
async def game_status(sid, *args):
resolved = _active_game(sid)
if resolved is None:
return await send_error(sid, "Nie ste v rozohratej hre.")
game, _ = resolved
await send_game_status(game.gid)
@sio.on("player_cards")
async def player_cards(sid, *args):
resolved = _active_game(sid)
if resolved is None:
return await send_error(sid, "Nie ste v rozohratej hre.")
game, sess = resolved
await send_player_cards(game.gid, sess["order"], sid)
@sio.on("add_guess")
async def add_guess(sid, guess):
resolved = _active_game(sid)
if resolved is None:
return await send_error(sid, "Nie ste v rozohratej hre.")
game, sess = resolved
try:
value = int(guess)
except (TypeError, ValueError):
return await send_error(sid, "Neplatny tip.")
try:
game.bridzik_core.add_player_guess(sess["order"], value)
except BridzikException as exc:
return await send_error(sid, str(exc))
await send_game_status(game.gid)
@sio.on("play_card")
async def play_card(sid, card_key):
resolved = _active_game(sid)
if resolved is None:
return await send_error(sid, "Nie ste v rozohratej hre.")
game, sess = resolved
core = game.bridzik_core
hand = core.get_player_cards(sess["order"])
try:
key = int(card_key)
except (TypeError, ValueError):
return await send_error(sid, "Neplatna karta.")
if key not in hand:
return await send_error(sid, "Neplatna karta.")
try:
core.play_card(sess["order"], hand[key])
except BridzikException as exc:
return await send_error(sid, str(exc))
# Persist completed rounds first so the DB-backed standings in game_status are
# up to date (idempotent; also marks the game ended).
await history.record_completed_rounds(game.gid, core)
await send_game_status(game.gid)
for player in game.players:
await send_player_cards(game.gid, player.order, player.sid)
# --- history (read-only) --------------------------------------------------
@sio.on("get_player_history")
async def get_player_history(sid, *args):
account = accounts.get(sid)
if account is None:
return await send_error(sid, "Musíte byť prihlásený.")
rows = await history.get_player_history(account["player_id"])
await sio.emit("get_player_history", {"games": rows}, to=sid)
@sio.on("get_game_detail")
async def get_game_detail(sid, gid):
account = accounts.get(sid)
if account is None:
return await send_error(sid, "Musíte byť prihlásený.")
detail = await history.get_game_detail(gid)
if detail is None:
return await send_error(sid, "Hra neexistuje.")
await sio.emit("get_game_detail", detail, to=sid)