modbison/backend/app/services/game_servers.py

"""Game-server orchestration.

On ``POST /api/worlds/{id}/join`` the backend ensures a headless Godot
dedicated-server process is running for that world and returns its address.
One process per world, keyed by ``world_id``. A background reaper kills a
world's server once its status file reports zero players for
``GAME_SERVER_IDLE_TIMEOUT_SECONDS``.

This lives in-process (a module-level singleton) rather than as a separate
service, so the backend MUST run single-worker -- which it does
(``uvicorn app.main:app`` with no ``--workers``). The Godot binary and
project are on the host (see ``settings.GODOT_BIN`` / ``GAME_PROJECT_DIR``);
the backend spawns them as subprocesses.

Design notes / limits (PC3):
 - Port allocation is a linear scan of the configured range.
 - No crash-restart: if a server process dies, the next join respawns it.
   With world content now durable in Postgres (world_persistence_impl.md
   D2), this is genuinely safe rather than merely tolerated: the sim
   process holds no unsaved world content, so a reap/crash/kill can never
   lose world data.

Per-spawn auth token (world_persistence_impl.md §7.3): each spawned server
gets a random ``secrets.token_urlsafe(32)`` on its command line, which it
uses as a Bearer token against ``/api/internal`` to pull its world's
content read-only before binding its ENet port. The token grants access to
exactly one world and dies with its process (the reaper's
``del self._servers[wid]`` invalidates it; see ``validate_token``).
"""
from __future__ import annotations

import asyncio
import hmac
import json
import secrets
import socket
import subprocess
import tempfile
import time
from dataclasses import dataclass, field
from pathlib import Path

from app.config import settings


@dataclass
class _WorldServer:
    world_id: str
    port: int
    proc: subprocess.Popen
    status_path: Path
    token: str
    started_at: float = field(default_factory=time.time)
    last_join: float = field(default_factory=time.time)

    def status(self) -> dict:
        try:
            return json.loads(self.status_path.read_text())
        except (OSError, ValueError):
            return {}

    def player_count(self) -> int:
        return int(self.status().get("players", 0))

    def is_ready(self) -> bool:
        return self.status().get("ready") is True

    def alive(self) -> bool:
        return self.proc.poll() is None

    def stop(self) -> None:
        if self.alive():
            self.proc.terminate()
            try:
                self.proc.wait(timeout=10)
            except subprocess.TimeoutExpired:
                self.proc.kill()
        self.status_path.unlink(missing_ok=True)


class GameServerSupervisor:
    def __init__(self) -> None:
        self._servers: dict[str, _WorldServer] = {}
        self._lock = asyncio.Lock()
        self._reaper: asyncio.Task | None = None
        self._runtime_dir = Path(tempfile.gettempdir()) / "agp-game-servers"
        self._runtime_dir.mkdir(parents=True, exist_ok=True)

    # ── lifecycle ────────────────────────────────────────────────────────
    def start(self) -> None:
        if not settings.GAME_SERVERS_ENABLED:
            return
        if self._reaper is None or self._reaper.done():
            self._reaper = asyncio.create_task(self._reap_loop())

    async def shutdown(self) -> None:
        if self._reaper:
            self._reaper.cancel()
        async with self._lock:
            for srv in list(self._servers.values()):
                srv.stop()
            self._servers.clear()

    # ── join ─────────────────────────────────────────────────────────────
    async def ensure_server(self, world_id: str) -> dict:
        """Return ``{"host", "port", "world_id"}`` for the world's dedicated
        server, spawning it if needed. Raises RuntimeError if orchestration
        is disabled or no port/engine is available."""
        if not settings.GAME_SERVERS_ENABLED:
            raise RuntimeError("game servers are not enabled on this backend")

        async with self._lock:
            srv = self._servers.get(world_id)
            if srv is not None and not srv.alive():
                srv.stop()
                srv = None
                del self._servers[world_id]

            if srv is None:
                srv = self._spawn(world_id)
                self._servers[world_id] = srv
                try:
                    await self._await_ready(srv)
                except TimeoutError:
                    # Real failure now, not a silent "hope it's fine" --
                    # see _await_ready's docstring. Don't leave a possibly
                    # half-started process registered.
                    srv.stop()
                    del self._servers[world_id]
                    raise RuntimeError(
                        f"dedicated server for world {world_id} did not become ready in time"
                    )

            srv.last_join = time.time()
            return {"host": settings.GAME_SERVER_HOST, "port": srv.port, "world_id": world_id}

    def validate_token(self, world_id: str, token: str) -> bool:
        """`require_game_server` dependency support (world_persistence_impl.md
        §7.3): true iff *token* is the live token for *world_id*'s server.
        A dict lookup, as the doc specifies, plus a constant-time compare
        since it's a bearer credential."""
        srv = self._servers.get(str(world_id))
        if srv is None or not srv.alive():
            return False
        return hmac.compare_digest(srv.token, token)

    # ── internals ────────────────────────────────────────────────────────
    def _spawn(self, world_id: str) -> _WorldServer:
        port = self._free_port()
        token = secrets.token_urlsafe(32)
        status_path = self._runtime_dir / f"world_{world_id}_{port}.json"
        status_path.unlink(missing_ok=True)
        cmd = [
            settings.GODOT_BIN, "--headless",
            "--path", settings.GAME_PROJECT_DIR,
            "res://scenes/server.tscn", "--",
            "--port", str(port),
            "--world", str(world_id),
            "--status-file", str(status_path),
            "--api-url", settings.GAME_SERVER_INTERNAL_API_URL,
            "--server-token", token,
        ]
        proc = subprocess.Popen(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
        return _WorldServer(
            world_id=world_id, port=port, proc=proc, status_path=status_path, token=token
        )

    def _free_port(self) -> int:
        used = {s.port for s in self._servers.values()}
        for p in range(settings.GAME_SERVER_PORT_MIN, settings.GAME_SERVER_PORT_MAX + 1):
            if p in used:
                continue
            with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s:
                try:
                    s.bind(("0.0.0.0", p))
                    return p
                except OSError:
                    continue
        raise RuntimeError("no free game-server port in the configured range")

    async def _await_ready(self, srv: _WorldServer) -> None:
        """Wait for a ``"ready": true`` field in the status file
        (world_persistence_impl.md §7.4).

        This replaces the old ``_await_port``, which had two real bugs:
        it treated a bound UDP port as "ready", but with content loading
        the port binds *before* the server has fetched its world's rooms
        and chunks (world_persistence_impl.md §7.2 step 4) -- so a client
        could get handed an address for a server still mid-fetch. And it
        *failed open*: past its 8s deadline it simply returned, with no
        error, so ``ensure_server`` reported success for a server that may
        never have started at all. This raises ``TimeoutError`` instead,
        which ``ensure_server`` turns into a real 503 for the joining
        client, and also bails out early (same exception) if the process
        has already exited.
        """
        deadline = time.time() + settings.GAME_SERVER_READY_TIMEOUT_SECONDS
        while time.time() < deadline:
            if srv.is_ready():
                return
            if not srv.alive():
                raise TimeoutError(
                    f"server process for world {srv.world_id} exited before becoming ready"
                )
            await asyncio.sleep(0.25)
        raise TimeoutError(
            f"server for world {srv.world_id} did not report ready within "
            f"{settings.GAME_SERVER_READY_TIMEOUT_SECONDS}s"
        )

    async def _reap_loop(self) -> None:
        try:
            while True:
                await asyncio.sleep(30)
                async with self._lock:
                    now = time.time()
                    for wid, srv in list(self._servers.items()):
                        if not srv.alive():
                            srv.stop()
                            del self._servers[wid]
                            continue
                        idle_for = now - srv.last_join
                        if idle_for > settings.GAME_SERVER_IDLE_TIMEOUT_SECONDS and srv.player_count() == 0:
                            srv.stop()
                            del self._servers[wid]
        except asyncio.CancelledError:
            pass

    def debug_state(self) -> list[dict]:
        return [
            {
                "world_id": s.world_id, "port": s.port, "alive": s.alive(),
                "players": s.player_count(), "uptime_s": round(time.time() - s.started_at),
            }
            for s in self._servers.values()
        ]


supervisor = GameServerSupervisor()