tactical-shooter/scripts/map_packaging/map_registry_server.py

#!/usr/bin/env python3
"""
map_registry_server.py — Map Registry HTTP Server

Serves packaged .pck map files and a JSON map listing for the Godot
client's MapDownloader to consume.

Endpoints
---------

GET /maps
    Returns a JSON list of available maps with metadata.

    Response:
    {
        "maps": [
            {
                "name": "de_dust2",
                "size": 4194304,
                "version": 1,
                "description": "Classic bomb-defusal map",
                "scene": "res://scenes/maps/de_dust2.tscn",
                "checksum_sha256": "a1b2c3..."
            }
        ],
        "server_name": "Tactical Shooter Map Registry",
        "map_count": 1
    }

GET /maps/<name>.pck
    Download a packaged map file. Content-Type: application/octet-stream.

GET /maps/<name>.json
    Download metadata for a specific map.

    Response:
    {
        "name": "de_dust2",
        "size": 4194304,
        "version": 1,
        "description": "Classic bomb-defusal map",
        "scene": "res://scenes/maps/de_dust2.tscn",
        "checksum_sha256": "a1b2c3...",
        "packed_at": "2026-06-30 12:00:00"
    }

Usage
-----

    # Start the server on the default port (8090):
    python3 map_registry_server.py

    # Custom port and map directory:
    python3 map_registry_server.py --port 8080 --maps-dir /data/maps

    # Docker / container friendly:
    MAP_REGISTRY_PORT=8080 MAP_REGISTRY_MAPS=/data/maps python3 map_registry_server.py

The server scans `maps_dir` (default: ./packed_maps/) for *.pck files and
their matching *.json metadata files on startup and on SIGHUP.

Integration
-----------

    In your server's ServerConfig or config, add a registry section:

        [map_registry]
        enabled=true
        url="http://maps.example.com:8090"

    Or set env: MAP_REGISTRY_URL on the game server.

    The game client's MapDownloader singleton will connect to this server
    to fetch the map list and download .pck files.
"""

import argparse
import hashlib
import json
import logging
import os
import signal
import sys
import time
from http.server import HTTPServer, BaseHTTPRequestHandler
from pathlib import Path
from typing import Dict, Optional

logger = logging.getLogger("MapRegistry")

# ---------------------------------------------------------------------------
# Configuration
# ---------------------------------------------------------------------------

DEFAULT_PORT = 8090
DEFAULT_MAPS_DIR = "packed_maps"
SERVER_NAME = "Tactical Shooter Map Registry"
VERSION = "1.0.0"


# ---------------------------------------------------------------------------
# Map Registry
# ---------------------------------------------------------------------------

class MapRegistry:
    """Scans a directory for .pck files and their metadata."""

    def __init__(self, maps_dir: str):
        self.maps_dir = Path(maps_dir)
        self.maps_dir.mkdir(parents=True, exist_ok=True)
        self._cache: Dict[str, dict] = {}
        self._last_scan: float = 0
        self._scan_interval: float = 5.0  # seconds between rescans
        self._scan()

    def _scan(self) -> None:
        """Scan the maps directory for .pck and .json files."""
        now = time.time()
        if now - self._last_scan < self._scan_interval:
            return
        self._last_scan = now

        self._cache.clear()
        if not self.maps_dir.exists():
            logger.warning("Maps directory does not exist: %s", self.maps_dir)
            return

        for pck_file in sorted(self.maps_dir.glob("*.pck")):
            map_name = pck_file.stem
            json_file = pck_file.with_suffix(".json")

            entry = {
                "name": map_name,
                "size": pck_file.stat().st_size,
                "path": str(pck_file.relative_to(self.maps_dir)),
                "version": 1,
                "description": "",
                "scene": f"res://scenes/maps/{map_name}.tscn",
                "checksum_sha256": "",
                "packed_at": "",
            }

            # Load metadata from sidecar JSON if it exists
            if json_file.exists():
                try:
                    with open(json_file, "r") as f:
                        meta = json.load(f)
                    entry["version"] = meta.get("version", 1)
                    entry["description"] = meta.get("description", "")
                    entry["packed_at"] = meta.get("packed_at", "")
                    entry["source_scene"] = meta.get("source_scene", "")
                except (json.JSONDecodeError, OSError) as e:
                    logger.warning("Failed to parse metadata for %s: %s", map_name, e)

            # Compute SHA-256 checksum (on first scan, cached in memory)
            ck = self._compute_checksum(pck_file)
            if ck:
                entry["checksum_sha256"] = ck

            self._cache[map_name] = entry
            logger.debug("Registered map: %s (%d bytes)", map_name, entry["size"])

        logger.info("Scanned %s: %d maps registered", self.maps_dir, len(self._cache))

    def _compute_checksum(self, path: Path) -> str:
        """Compute SHA-256 checksum of a file."""
        try:
            h = hashlib.sha256()
            with open(path, "rb") as f:
                while True:
                    chunk = f.read(65536)
                    if not chunk:
                        break
                    h.update(chunk)
            return h.hexdigest()
        except OSError as e:
            logger.warning("Checksum failed for %s: %s", path.name, e)
            return ""

    def get_map_list(self) -> dict:
        """Return the full map listing as a dict."""
        self._scan()
        return {
            "maps": list(self._cache.values()),
            "server_name": SERVER_NAME,
            "server_version": VERSION,
            "map_count": len(self._cache),
        }

    def get_map_info(self, name: str) -> Optional[dict]:
        """Return metadata for a single map."""
        self._scan()
        return self._cache.get(name)

    def get_map_file(self, name: str) -> Optional[Path]:
        """Return the filesystem path to a .pck file, or None."""
        self._scan()
        entry = self._cache.get(name)
        if entry is None:
            return None
        pck_path = self.maps_dir / entry["path"]
        return pck_path if pck_path.exists() else None


# ---------------------------------------------------------------------------
# HTTP Handler
# ---------------------------------------------------------------------------

class MapRegistryHandler(BaseHTTPRequestHandler):
    """HTTP request handler for the map registry API."""

    # Shared across all instances (set by the server)
    registry: MapRegistry = None  # type: ignore

    def log_message(self, format: str, *args) -> None:
        logger.info("%s - %s", self.client_address[0], format % args)

    def _send_json(self, data: dict, status: int = 200) -> None:
        body = json.dumps(data, indent=2).encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.send_header("Access-Control-Allow-Origin", "*")
        self.end_headers()
        self.wfile.write(body)

    def _send_error(self, status: int, message: str) -> None:
        self._send_json({"error": message}, status)

    def _send_file(self, path: Path, content_type: str = "application/octet-stream") -> None:
        try:
            file_size = path.stat().st_size
            self.send_response(200)
            self.send_header("Content-Type", content_type)
            self.send_header("Content-Length", str(file_size))
            self.send_header("Content-Disposition", f'attachment; filename="{path.name}"')
            self.send_header("Access-Control-Allow-Origin", "*")
            self.end_headers()

            with open(path, "rb") as f:
                while True:
                    chunk = f.read(65536)
                    if not chunk:
                        break
                    self.wfile.write(chunk)
        except OSError as e:
            logger.error("File send failed for %s: %s", path.name, e)
            self._send_error(500, "Internal server error")

    def do_GET(self) -> None:
        path = self.path.rstrip("/")

        # GET /maps — list all available maps
        if path == "/maps":
            self._send_json(self.registry.get_map_list())
            return

        # GET /maps/<name>.pck — download a map file
        if path.startswith("/maps/") and path.endswith(".pck"):
            map_name = path[len("/maps/"):-len(".pck")]
            map_file = self.registry.get_map_file(map_name)
            if map_file:
                self._send_file(map_file)
            else:
                self._send_error(404, f"Map '{map_name}' not found")
            return

        # GET /maps/<name>.json — get metadata for a single map
        if path.startswith("/maps/") and path.endswith(".json"):
            map_name = path[len("/maps/"):-len(".json")]
            info = self.registry.get_map_info(map_name)
            if info:
                self._send_json(info)
            else:
                self._send_error(404, f"Map '{map_name}' not found")
            return

        # GET / — server info
        if path == "/" or path == "":
            self._send_json({
                "service": SERVER_NAME,
                "version": VERSION,
                "endpoints": {
                    "list_maps": "/maps",
                    "download_map": "/maps/<name>.pck",
                    "map_metadata": "/maps/<name>.json",
                },
            })
            return

        self._send_error(404, f"Not found: {path}")

    def do_OPTIONS(self) -> None:
        """CORS preflight."""
        self.send_response(204)
        self.send_header("Access-Control-Allow-Origin", "*")
        self.send_header("Access-Control-Allow-Methods", "GET, OPTIONS")
        self.send_header("Access-Control-Allow-Headers", "Content-Type")
        self.end_headers()


# ---------------------------------------------------------------------------
# Server
# ---------------------------------------------------------------------------

def create_server(port: int, maps_dir: str) -> HTTPServer:
    """Create and configure the HTTP server."""
    registry = MapRegistry(maps_dir)
    MapRegistryHandler.registry = registry

    server = HTTPServer(("0.0.0.0", port), MapRegistryHandler)
    server.timeout = 0.5  # allow signal handling

    logger.info("Map Registry Server v%s", VERSION)
    logger.info("  Listen:     http://0.0.0.0:%d", port)
    logger.info("  Maps dir:   %s", registry.maps_dir.resolve())
    logger.info("  Maps found: %d", len(registry._cache))
    logger.info("  Endpoints:")
    logger.info("    List:      GET /maps")
    logger.info("    Download:  GET /maps/<name>.pck")
    logger.info("    Metadata:  GET /maps/<name>.json")

    return server


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Tactical Shooter Map Registry Server",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=__doc__,
    )
    parser.add_argument(
        "--port", "-p",
        type=int,
        default=int(os.environ.get("MAP_REGISTRY_PORT", DEFAULT_PORT)),
        help="HTTP port (default: %d)" % DEFAULT_PORT,
    )
    parser.add_argument(
        "--maps-dir", "-d",
        type=str,
        default=os.environ.get("MAP_REGISTRY_MAPS", DEFAULT_MAPS_DIR),
        help="Directory containing .pck files (default: %s)" % DEFAULT_MAPS_DIR,
    )
    parser.add_argument(
        "--verbose", "-v",
        action="store_true",
        help="Enable debug logging",
    )
    return parser.parse_args()


def main():
    args = parse_args()

    logging.basicConfig(
        level=logging.DEBUG if args.verbose else logging.INFO,
        format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
        datefmt="%Y-%m-%d %H:%M:%S",
    )

    server = create_server(args.port, args.maps_dir)

    # Handle SIGTERM/SIGINT for graceful shutdown
    shutdown_requested = False

    def handle_signal(sig, frame):
        nonlocal shutdown_requested
        if not shutdown_requested:
            logger.info("Shutdown requested (signal %d)", sig)
            shutdown_requested = True
            server.shutdown()

    signal.signal(signal.SIGTERM, handle_signal)
    signal.signal(signal.SIGINT, handle_signal)

    # SIGHUP rescans the map directory
    if hasattr(signal, "SIGHUP"):
        def handle_hup(sig, frame):
            logger.info("SIGHUP received — rescanning maps")
            server.RequestHandlerClass.registry._scan()
        signal.signal(signal.SIGHUP, handle_hup)

    try:
        server.serve_forever()
    except KeyboardInterrupt:
        pass
    finally:
        server.server_close()
        logger.info("Server stopped")


if __name__ == "__main__":
    main()