"""Moonchess server: serves the page and runs live games over WebSockets. Any string works as a game code. Joining a code with no game creates one, and a game is deleted as soon as nobody is connected to it. Seating: each room keeps a list of slots. Slots 0 and 1 are the players (X and O), everything after is spectators. A new connection takes the earliest empty slot, so when a player drops, whoever connects next -- including a spectator who refreshes fast enough -- inherits their seat. That is intentional. """ import argparse import asyncio import logging import time from pathlib import Path from aiohttp import WSMsgType, web import practice from game import Game STATIC_DIR = Path(__file__).parent / "static" STALE_AFTER = 2.0 # seconds without a message before a seat shows as disconnected HEARTBEAT_TIMEOUT = 4.0 # seconds without a message before a client is dropped MAX_ROOMS = 10_000 MAX_CODE_LENGTH = 32 CLOSE_INVALID_CODE = 4400 CLOSE_SERVER_FULL = 4503 MAX_MESSAGE_SIZE = 4096 # bytes; real messages are tiny, larger ones close the connection log = logging.getLogger("moonchess") class Client: def __init__(self, ws, client_id): self.ws = ws self.client_id = client_id self.last_seen = time.monotonic() class Room: def __init__(self, token): self.token = token self.game = Game() self.slots = [] # Client or None self.last_seats = None # seat_status() as of the last broadcast def join(self, client): # A reconnect from the same page load reclaims its own not-yet-reaped slot. for i, c in enumerate(self.slots): if c is not None and client.client_id and c.client_id == client.client_id: self.slots[i] = client asyncio.ensure_future(c.ws.close()) return for i, c in enumerate(self.slots): if c is None: self.slots[i] = client return self.slots.append(client) def leave(self, client): if client not in self.slots: return False i = self.slots.index(client) self.slots[i] = None if i < 2: self.game.rematch_votes.discard(i) while len(self.slots) > 2 and self.slots[-1] is None: self.slots.pop() return True def connected(self): return [c for c in self.slots if c is not None] def seat_of(self, client): i = self.slots.index(client) return i if i < 2 else None def players_present(self): return [len(self.slots) > i and self.slots[i] is not None for i in (0, 1)] def seat_status(self): """'connected', 'disconnected' (silent but not yet dropped) or 'empty' per seat.""" now = time.monotonic() status = [] for i in (0, 1): c = self.slots[i] if len(self.slots) > i else None if c is None: status.append("empty") elif now - c.last_seen > STALE_AFTER: status.append("disconnected") else: status.append("connected") return status async def broadcast(self): self.last_seats = self.seat_status() base = { "type": "state", "token": self.token, "first": self.game.first, "players": self.last_seats, "spectators": sum(1 for c in self.slots[2:] if c is not None), **self.game.to_dict(), } for c in self.connected(): try: await c.ws.send_json({**base, "seat": self.seat_of(c)}) except (ConnectionError, RuntimeError): pass # the reaper or the handler will clean it up rooms: dict[str, Room] = {} def normalize_code(raw): """Case-insensitive, with surrounding and repeated whitespace ignored.""" return " ".join(raw.split()).upper() async def index(request): return web.FileResponse(STATIC_DIR / "index.html") async def practice_page(request): return web.FileResponse(STATIC_DIR / "practice.html") async def game_socket(request): code = normalize_code(request.query.get("game", "")) ws = web.WebSocketResponse(max_msg_size=MAX_MESSAGE_SIZE) await ws.prepare(request) if not 0 < len(code) <= MAX_CODE_LENGTH: await ws.close(code=CLOSE_INVALID_CODE, message=f"Codes must be 1-{MAX_CODE_LENGTH} characters".encode()) return ws room = rooms.get(code) if room is None: if len(rooms) >= MAX_ROOMS: await ws.close(code=CLOSE_SERVER_FULL, message=b"Server is full, try again later") return ws room = rooms[code] = Room(code) log.info("created room %r (%d rooms)", code, len(rooms)) client = Client(ws, request.query.get("client", "")[:64]) room.join(client) await room.broadcast() try: async for msg in ws: if msg.type != WSMsgType.TEXT: continue client.last_seen = time.monotonic() try: data = msg.json() except ValueError: continue if not isinstance(data, dict): continue kind = data.get("type") if kind == "ping": await ws.send_json({"type": "pong"}) continue if client not in room.slots: break # reaped or replaced while this message was in flight seat = room.seat_of(client) if seat is None: continue # spectators can't act if kind == "move": if not all(room.players_present()): error = "Waiting for an opponent" else: error = room.game.play(seat, data.get("cell")) if error: await ws.send_json({"type": "error", "message": error}) else: await room.broadcast() elif kind == "rematch": room.game.vote_rematch(seat) await room.broadcast() finally: if room.leave(client) and not delete_if_empty(room): await room.broadcast() return ws def delete_if_empty(room): if room.connected(): return False if rooms.get(room.token) is room: del rooms[room.token] log.info("deleted empty room %r (%d rooms)", room.token, len(rooms)) return True async def reaper(app): """Drop silent clients, announce seats going quiet, and delete abandoned rooms.""" while True: await asyncio.sleep(1) now = time.monotonic() for room in list(rooms.values()): changed = False for c in room.connected(): if now - c.last_seen > HEARTBEAT_TIMEOUT: room.leave(c) asyncio.ensure_future(c.ws.close()) changed = True if delete_if_empty(room): continue if changed or room.seat_status() != room.last_seats: await room.broadcast() async def start_reaper(app): task = asyncio.create_task(reaper(app)) yield task.cancel() @web.middleware async def no_cache(request, handler): # Keep Cloudflare and browsers from serving stale copies after an update. response = await handler(request) if not isinstance(response, web.WebSocketResponse): response.headers["Cache-Control"] = "no-cache" return response def make_app(): app = web.Application(middlewares=[no_cache]) app.router.add_get("/", index) app.router.add_get("/ws", game_socket) app.router.add_get("/practice", practice_page) app.router.add_post("/api/play", practice.play) app.router.add_static("/static", STATIC_DIR) app.cleanup_ctx.append(start_reaper) return app if __name__ == "__main__": parser = argparse.ArgumentParser(description="Moonchess server") parser.add_argument("--host", default="127.0.0.1") parser.add_argument("--port", type=int, default=8080) args = parser.parse_args() logging.basicConfig(level=logging.INFO, format="%(asctime)s %(message)s") web.run_app(make_app(), host=args.host, port=args.port)