from __future__ import annotations import argparse import json import logging import re import socket import threading import uuid from dataclasses import dataclass from datetime import datetime, timezone from http import HTTPStatus from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from typing import Any from urllib.parse import parse_qs, urlparse ROOT_DIR = Path(__file__).resolve().parent.parent DATA_DIR = ROOT_DIR / "data" USERS_FILE = DATA_DIR / "users.json" GAMES_FILE = DATA_DIR / "games.json" SERVER_PORT = 8765 LOGGER = logging.getLogger("brass_backend") def now_iso() -> str: return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z") def get_local_ip() -> str: sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) try: sock.connect(("8.8.8.8", 80)) return sock.getsockname()[0] except OSError: return "127.0.0.1" finally: sock.close() def load_json(path: Path, default: Any) -> Any: if not path.exists(): return default with path.open("r", encoding="utf-8") as handle: return json.load(handle) def save_json(path: Path, payload: Any) -> None: path.parent.mkdir(parents=True, exist_ok=True) with path.open("w", encoding="utf-8") as handle: json.dump(payload, handle, ensure_ascii=False, indent=2) def ensure_player_metadata_slots(game: dict[str, Any]) -> None: cfg = game.setdefault("GameStartConfiguration", {}) wanted = int(cfg.get("NumberOfPlayersAsInt", 4) or 4) players = game.setdefault("PlayerMetadata", []) while len(players) < wanted: players.append({}) def normalize_revision(value: Any) -> int: try: return int(value) except (TypeError, ValueError): return 0 def make_response(data: Any = None, status: int = 200) -> tuple[int, dict[str, Any]]: return status, {"data": data} def make_error(message: str, status: int = 400) -> tuple[int, dict[str, Any]]: return status, {"error": message} def make_auth_response(access_token: str, status: int = 200) -> tuple[int, dict[str, Any]]: return status, {"access_token": access_token} def make_user_response(user: dict[str, Any], status: int = 200) -> tuple[int, dict[str, Any]]: return status, { "id": user["id"], "email": user["email"], "name": user.get("name") or user.get("nickname") or user["email"], } def make_message_response(message: str = "ok", status: int = 200) -> tuple[int, dict[str, Any]]: return status, {"message": message} def coerce_payload_dict(body: Any) -> dict[str, Any]: if not isinstance(body, dict): return {} nested = body.get("data") if isinstance(nested, dict): merged = dict(nested) for key, value in body.items(): if key != "data" and key not in merged: merged[key] = value return merged return body def nested_get(payload: Any, path: list[str]) -> Any: current = payload for part in path: if isinstance(current, dict): current = current.get(part) else: return None return current def make_raw_response(payload: Any, status: int = 200) -> tuple[int, Any]: return status, payload def ensure_game_metadata_shape(game: dict[str, Any]) -> dict[str, Any]: metadata = game.setdefault("metadata", {}) if "PlayerMetadata" not in metadata and "PlayerMetadata" in game: metadata["PlayerMetadata"] = game["PlayerMetadata"] if "NetworkGameProgressInformation" not in metadata and "NetworkGameProgressInformation" in game: metadata["NetworkGameProgressInformation"] = game["NetworkGameProgressInformation"] if "GameStartConfiguration" not in metadata and "GameStartConfiguration" in game: metadata["GameStartConfiguration"] = game["GameStartConfiguration"] if "events" not in metadata and "events" in game: metadata["events"] = game["events"] return game def infer_nickname_from_game_name(game_name: Any) -> str | None: if not isinstance(game_name, str) or not game_name: return None if "'的游戏" in game_name: return game_name.split("'的游戏", 1)[0].strip() or None if "'s game" in game_name: return game_name.split("'s game", 1)[0].strip() or None return None def metadata_values(metadata: dict[str, Any], key: str) -> list[Any]: value = metadata.get(key) if isinstance(value, list): return value if isinstance(value, dict) and isinstance(value.get("$values"), list): return value["$values"] return [] def normalize_player_entry(entry: Any) -> Any: if entry is None or not isinstance(entry, dict): return entry if isinstance(entry.get("metadata"), dict): normalized = dict(entry["metadata"]) for key in ( "PlayerNetworkId", "NickName", "GameId", "AiType", "PlayerAvatar", "PlayerStartConfiguration", "LastSeenEventRevision", ): if entry.get(key) is not None and normalized.get(key) is None: normalized[key] = entry.get(key) return normalized return entry def normalize_player_metadata_slots(metadata: dict[str, Any]) -> list[Any]: players = metadata.get("PlayerMetadata", []) if isinstance(players, dict) and "$values" in players: players = players["$values"] metadata["PlayerMetadata"] = players if not isinstance(players, list): players = [] metadata["PlayerMetadata"] = players for idx, entry in enumerate(players): players[idx] = normalize_player_entry(entry) max_players = nested_get(metadata, ["GameStartConfiguration", "NumberOfPlayersAsInt"]) try: max_players = int(max_players) except (TypeError, ValueError): max_players = len(players) while len(players) < max_players: players.append(None) if max_players > 0 and len(players) > max_players: del players[max_players:] return players def real_player_count(players: list[Any]) -> int: return sum(1 for entry in players if isinstance(entry, dict)) def sync_network_game_status(metadata: dict[str, Any]) -> list[Any]: players = normalize_player_metadata_slots(metadata) max_players = nested_get(metadata, ["GameStartConfiguration", "NumberOfPlayersAsInt"]) try: max_players = int(max_players) except (TypeError, ValueError): max_players = len(players) progress = metadata.setdefault("NetworkGameProgressInformation", {}) joined = real_player_count(players) if joined >= max_players and max_players > 0: progress["CurrentGameStatus"] = "InProgress" if not progress.get("CurrentPlayerId"): first_player = next((entry for entry in players if isinstance(entry, dict)), None) if first_player: progress["CurrentPlayerId"] = first_player.get("PlayerNetworkId") elif joined > 0: progress["CurrentGameStatus"] = "Open" return players def event_revision(event: dict[str, Any]) -> int: if not isinstance(event, dict): return 0 nested = event.get("Metadata") if isinstance(nested, dict): try: return int(nested.get("Revision") or 0) except (TypeError, ValueError): return 0 try: return int(event.get("Revision") or 0) except (TypeError, ValueError): return 0 def set_event_revision(event: dict[str, Any], revision: int) -> None: metadata = event.setdefault("Metadata", {}) if isinstance(metadata, dict): metadata["Revision"] = revision event["Revision"] = revision def normalize_events_payload(events: Any) -> list[Any]: if isinstance(events, dict) and isinstance(events.get("$values"), list): return list(events["$values"]) if isinstance(events, list): flattened: list[Any] = [] for event in events: if isinstance(event, dict) and isinstance(event.get("$values"), list): flattened.extend(event["$values"]) else: flattened.append(event) return flattened return [events] def _resolve_player_id_from_color(metadata: dict[str, Any], player_color: Any) -> str | None: if player_color is None: return None players = metadata_values(metadata, "PlayerMetadata") for player in players: if not isinstance(player, dict): continue start_cfg = player.get("PlayerStartConfiguration") if isinstance(start_cfg, dict) and start_cfg.get("Color") == player_color: return player.get("PlayerNetworkId") or None if isinstance(player.get("PlayerStartConfiguration"), dict): cfg_color = player.get("PlayerStartConfiguration", {}).get("Color") if cfg_color == player_color: return player.get("PlayerNetworkId") or None if player.get("NickName") and str(player.get("NickName")).lower() == str(player_color).lower(): return player.get("PlayerNetworkId") or None return None def _resolve_next_player_id(metadata: dict[str, Any], current_player_color: Any) -> str | None: if current_player_color is None: return None start_configs = nested_get(metadata, ["GameStartConfiguration", "Players", "$values"]) if not isinstance(start_configs, list): return None ordered_colors = [] for entry in start_configs: if isinstance(entry, dict): color = entry.get("Color") if color is not None: ordered_colors.append(str(color)) if not ordered_colors: return None try: current_index = ordered_colors.index(str(current_player_color)) except ValueError: return None next_index = (current_index + 1) % len(ordered_colors) next_color = ordered_colors[next_index] return _resolve_player_id_from_color(metadata, next_color) def _resolve_next_player(metadata: dict[str, Any], current_player_color: Any) -> tuple[str | None, str | None]: if current_player_color is None: return None, None start_configs = nested_get(metadata, ["GameStartConfiguration", "Players", "$values"]) if not isinstance(start_configs, list): return None, None ordered_colors = [] for entry in start_configs: if isinstance(entry, dict): color = entry.get("Color") if color is not None: ordered_colors.append(str(color)) if not ordered_colors: return None, None try: current_index = ordered_colors.index(str(current_player_color)) except ValueError: return None, None next_index = (current_index + 1) % len(ordered_colors) next_color = ordered_colors[next_index] next_player_id = _resolve_player_id_from_color(metadata, next_color) return next_player_id, next_color def sync_progress_from_events(game: dict[str, Any]) -> None: metadata = game.setdefault("metadata", {}) progress = metadata.setdefault("NetworkGameProgressInformation", {}) game_progress = metadata.setdefault("GameProgressInformation", {}) events = game.get("events", []) if not isinstance(events, list): return current_player_id = None current_player_color = None for event in reversed(events): if not isinstance(event, dict): continue event_type = str(event.get("$type") or "") if "TurnStarted" in event_type and event.get("PlayerColor") is not None: current_player_id = _resolve_player_id_from_color(metadata, event.get("PlayerColor")) current_player_color = str(event.get("PlayerColor")) break if "TurnEnded" in event_type: previous_turn_color = None for candidate in reversed(events[: events.index(event)]): if not isinstance(candidate, dict): continue candidate_type = str(candidate.get("$type") or "") if "TurnStarted" in candidate_type and candidate.get("PlayerColor") is not None: previous_turn_color = str(candidate.get("PlayerColor")) break if previous_turn_color is not None: current_player_id, current_player_color = _resolve_next_player(metadata, previous_turn_color) break if current_player_id: progress["CurrentPlayerId"] = current_player_id progress["CurrentGameStatus"] = "InProgress" progress["CurrentPlayerColor"] = current_player_color or progress.get("CurrentPlayerColor") current_player = game_progress.setdefault("CurrentPlayer", {}) if current_player_color: current_player["Color"] = current_player_color return for event in reversed(events): if not isinstance(event, dict): continue event_type = str(event.get("$type") or "") if "TurnStarted" in event_type and event.get("PlayerColor") is not None: player_color = event.get("PlayerColor") player_id = _resolve_player_id_from_color(metadata, player_color) if player_id: progress["CurrentPlayerId"] = player_id progress["CurrentGameStatus"] = "InProgress" progress["CurrentPlayerColor"] = str(player_color) current_player = game_progress.setdefault("CurrentPlayer", {}) current_player["Color"] = str(player_color) break def game_display_name(game: dict[str, Any]) -> str: metadata = game.get("metadata", {}) return str(nested_get(metadata, ["NetworkGameStartConfiguration", "GameName"]) or "") def game_owner_key(game: dict[str, Any]) -> str: metadata = game.get("metadata", {}) players = normalize_player_metadata_slots(metadata) first_player = next((entry for entry in players if isinstance(entry, dict)), None) if first_player: return str(first_player.get("PlayerNetworkId") or first_player.get("NickName") or "") nickname = infer_nickname_from_game_name(game_display_name(game)) return nickname or "" def game_dedupe_key(game: dict[str, Any]) -> str: return f"{game_owner_key(game)}|{game_display_name(game)}" def is_client_safe_game_metadata(metadata: Any) -> bool: if not isinstance(metadata, dict): return False if not metadata.get("$type"): return False if not isinstance(metadata.get("GameStartConfiguration"), dict): return False if not isinstance(metadata.get("NetworkGameProgressInformation"), dict): return False if not isinstance(metadata.get("NetworkGameStartConfiguration"), dict): return False player_metadata = metadata.get("PlayerMetadata") if isinstance(player_metadata, dict): if not isinstance(player_metadata.get("$values"), list): return False elif not isinstance(player_metadata, list): return False return True def cleanup_games_on_load(raw_games: list[Any], users: list[dict[str, Any]]) -> dict[str, dict[str, Any]]: games_by_id: dict[str, dict[str, Any]] = {} user_names = { user.get("id", ""): user.get("name") or user.get("nickname") or user.get("email") or "" for user in users if isinstance(user, dict) } deduped: dict[str, dict[str, Any]] = {} removed_ids: list[str] = [] for entry in raw_games: if not isinstance(entry, dict) or "id" not in entry: continue game = ensure_game_metadata_shape(dict(entry)) metadata = game.setdefault("metadata", {}) # Normalize PlayerMetadata container shape. normalize_player_metadata_slots(metadata) # Repair empty or anonymous owner slot from game name when possible. players = metadata.get("PlayerMetadata", []) real_players = [p for p in players if isinstance(p, dict)] if not real_players: nickname = infer_nickname_from_game_name(game_display_name(game)) matched_user_id = None for uid, uname in user_names.items(): if uname == nickname: matched_user_id = uid break start_players = nested_get(metadata, ["GameStartConfiguration", "Players", "$values"]) first_start = start_players[0] if isinstance(start_players, list) and start_players else None inferred = { "$type": "BrassApplication.Game.PlayerMetadata, Assembly-CSharp", "NickName": nickname or "Host", "AiType": "Human", "PlayerAvatar": "Arkwright", "PlayerStartConfiguration": first_start, "LastSeenEventRevision": 0, } if matched_user_id: inferred["PlayerNetworkId"] = matched_user_id player_count = nested_get(metadata, ["GameStartConfiguration", "NumberOfPlayersAsInt"]) try: player_count = int(player_count) except (TypeError, ValueError): player_count = 2 players = [inferred] while len(players) < max(player_count, 1): players.append(None) metadata["PlayerMetadata"] = players sync_network_game_status(metadata) game["updated_at"] = game.get("updated_at") or game.get("created_at") or now_iso() games_by_id[game["id"]] = game for game in games_by_id.values(): key = game_dedupe_key(game) or game.get("id", "") existing = deduped.get(key) if existing is None: deduped[key] = game continue existing_players = real_player_count(normalize_player_metadata_slots(existing.get("metadata", {}))) current_players = real_player_count(normalize_player_metadata_slots(game.get("metadata", {}))) existing_updated = str(existing.get("updated_at") or "") current_updated = str(game.get("updated_at") or "") if current_players > existing_players or ( current_players == existing_players and current_updated >= existing_updated ): removed_ids.append(existing["id"]) deduped[key] = game else: removed_ids.append(game["id"]) if removed_ids: LOGGER.warning("[BOOT] removed duplicate/stale game ids=%s", removed_ids) return {game["id"]: game for game in deduped.values()} def setup_logging() -> None: logging.basicConfig( level=logging.INFO, format="%(asctime)s %(levelname)s [%(name)s] %(message)s", ) @dataclass class Storage: lock: threading.Lock users: list[dict[str, Any]] games: dict[str, dict[str, Any]] tokens: dict[str, str] @classmethod def load(cls) -> "Storage": DATA_DIR.mkdir(parents=True, exist_ok=True) users = load_json(USERS_FILE, []) games_list = load_json(GAMES_FILE, []) games = cleanup_games_on_load(games_list, users) return cls(lock=threading.Lock(), users=users, games=games, tokens={}) def save_users(self) -> None: save_json(USERS_FILE, self.users) def save_games(self) -> None: save_json(GAMES_FILE, list(self.games.values())) def find_user_by_email(self, email: str) -> dict[str, Any] | None: email = email.strip().lower() for user in self.users: if user.get("email", "").strip().lower() == email: return user return None def user_by_id(self, user_id: str) -> dict[str, Any] | None: for user in self.users: if user.get("id") == user_id: return user return None class BrassAPI: def __init__(self, storage: Storage) -> None: self.storage = storage def repair_game_metadata_for_listing(self, game: dict[str, Any]) -> dict[str, Any]: game = ensure_game_metadata_shape(dict(game)) metadata = game.setdefault("metadata", {}) players = metadata.get("PlayerMetadata") if isinstance(players, dict) and "$values" in players: players = players["$values"] metadata["PlayerMetadata"] = players if isinstance(players, list) and len(players) == 0: game_name = nested_get(metadata, ["NetworkGameStartConfiguration", "GameName"]) nickname = infer_nickname_from_game_name(game_name) matched_user = None if nickname: for user in self.storage.users: if user.get("name") == nickname or user.get("nickname") == nickname: matched_user = user break start_players = nested_get(metadata, ["GameStartConfiguration", "Players", "$values"]) first_start = start_players[0] if isinstance(start_players, list) and start_players else None inferred_player = { "$type": "BrassApplication.Game.PlayerMetadata, Assembly-CSharp", "NickName": nickname or "Host", "AiType": "Human", "PlayerAvatar": "Arkwright", "PlayerStartConfiguration": first_start, "LastSeenEventRevision": 0, } if matched_user: inferred_player["PlayerNetworkId"] = matched_user["id"] player_count = nested_get(metadata, ["GameStartConfiguration", "NumberOfPlayersAsInt"]) try: player_count = int(player_count) except (TypeError, ValueError): player_count = 2 repaired_players: list[Any] = [inferred_player] while len(repaired_players) < max(player_count, 1): repaired_players.append(None) metadata["PlayerMetadata"] = repaired_players LOGGER.warning( "[GAME] repaired empty PlayerMetadata game_id=%s nickname=%s inferred_user_id=%s", game.get("id"), nickname, matched_user.get("id") if matched_user else None, ) return game def parse_body(self, handler: BaseHTTPRequestHandler) -> Any: length = int(handler.headers.get("Content-Length", "0") or "0") raw = handler.rfile.read(length) if length else b"" if not raw: return {} ctype = handler.headers.get("Content-Type", "") if "application/json" in ctype: return json.loads(raw.decode("utf-8")) if "application/x-www-form-urlencoded" in ctype: form = parse_qs(raw.decode("utf-8"), keep_blank_values=True) return {key: values[-1] if values else "" for key, values in form.items()} try: return json.loads(raw.decode("utf-8")) except json.JSONDecodeError: form = parse_qs(raw.decode("utf-8"), keep_blank_values=True) return {key: values[-1] if values else "" for key, values in form.items()} def get_param(self, handler: BaseHTTPRequestHandler, body: Any, key: str, default: Any = None) -> Any: if isinstance(body, dict) and key in body: return body[key] query = parse_qs(urlparse(handler.path).query) if key in query and query[key]: return query[key][-1] return default def current_user_id(self, handler: BaseHTTPRequestHandler) -> str | None: auth = handler.headers.get("Authorization", "") if auth.startswith("Bearer "): token = auth.removeprefix("Bearer ").strip() if token in self.storage.tokens: user_id = self.storage.tokens[token] LOGGER.info("[AUTH] bearer token resolved user_id=%s path=%s", user_id, handler.path) return user_id if token.startswith("fake-token-"): user_id = token.removeprefix("fake-token-").strip() LOGGER.info("[AUTH] fake token resolved user_id=%s path=%s", user_id, handler.path) return user_id token = handler.headers.get("access_token", "") if token: if token in self.storage.tokens: user_id = self.storage.tokens[token] LOGGER.info("[AUTH] access_token resolved user_id=%s path=%s", user_id, handler.path) return user_id if token.startswith("fake-token-"): user_id = token.removeprefix("fake-token-").strip() LOGGER.info("[AUTH] fake access_token resolved user_id=%s path=%s", user_id, handler.path) return user_id direct = handler.headers.get("X-User-Id", "") if direct: user_id = direct.strip() LOGGER.info("[AUTH] X-User-Id resolved user_id=%s path=%s", user_id, handler.path) return user_id LOGGER.warning("[AUTH] no auth headers path=%s remote=%s", handler.path, handler.client_address[0]) return None def current_user(self, handler: BaseHTTPRequestHandler) -> dict[str, Any] | None: user_id = self.current_user_id(handler) if not user_id: return None user = self.storage.user_by_id(user_id) if user: LOGGER.info("[AUTH] user resolved user_id=%s email=%s", user_id, user.get("email")) else: LOGGER.warning("[AUTH] user not found for user_id=%s path=%s", user_id, handler.path) return user def route(self, handler: BaseHTTPRequestHandler) -> tuple[int, dict[str, Any]]: parsed = urlparse(handler.path) path = parsed.path method = handler.command.upper() body = self.parse_body(handler) if method in {"POST", "PUT", "PATCH", "DELETE"} else {} LOGGER.info("[REQ] %s %s remote=%s", method, path, handler.client_address[0]) if method == "POST" and path in {"/login", "/login/with/e-mail"}: return self.login(body) if method == "POST" and path == "/register": return self.register(body) if method == "GET" and path == "/me": return self.me(handler) if method == "POST" and path == "/logout": return make_message_response("ok") if method == "POST" and path == "/password/email": return make_message_response("ok") if method == "GET" and path == "/1.0.0/games": return self.get_games(handler) if method == "POST" and path == "/1.0.0/games": return self.create_game(handler, body) if method == "POST" and path == "/1.0.0/devices": return make_response({"message": "ok"}) if method == "GET" and path == "/debug/games": return make_response(list(self.storage.games.values())) player_match = re.fullmatch(r"/1\.0\.0/games/([^/]+)/players", path) if player_match: game_id = player_match.group(1) if method == "PUT": return self.join_game(handler, game_id, body) if method == "DELETE": return self.leave_game(handler, game_id, body) events_match = re.fullmatch(r"/1\.0\.0/games/([^/]+)/events", path) if events_match: game_id = events_match.group(1) if method == "GET": return self.get_events(game_id) if method == "POST": return self.save_events(game_id, body) if method == "DELETE": return self.delete_events(game_id) metadata_match = re.fullmatch(r"/1\.0\.0/games/([^/]+)/metadata/lastSeenRevision", path) if metadata_match and method == "POST": return self.update_last_seen(handler, metadata_match.group(1), body) events_meta_match = re.fullmatch(r"/1\.0\.0/games/([^/]+)/events-and-metadata", path) if events_meta_match: if method in {"POST", "PUT"}: return self.save_events_and_metadata(events_meta_match.group(1), body) if method == "GET": return self.get_game_snapshot(events_meta_match.group(1)) return make_error("Not found", status=404) def login(self, body: Any) -> tuple[int, dict[str, Any]]: email = str(body.get("email", "")).strip().lower() if isinstance(body, dict) else "" password = str(body.get("password", "")).strip() if isinstance(body, dict) else "" LOGGER.info("[LOGIN] attempt email=%s", email or "") if not email or not password: LOGGER.warning("[LOGIN] rejected missing credentials email=%s", email or "") return make_error("Missing email or password", status=400) user = self.storage.find_user_by_email(email) if not user: LOGGER.warning("[LOGIN] user not found email=%s", email) return make_error("User not found", status=404) token = str(uuid.uuid4()) self.storage.tokens[token] = user["id"] LOGGER.info("[LOGIN] success email=%s user_id=%s token=%s", email, user["id"], token) return 200, {"access_token": token, "data": {"access_token": token, "user": make_user_response(user)[1]}} def register(self, body: Any) -> tuple[int, dict[str, Any]]: email = str(body.get("email", "")).strip().lower() if isinstance(body, dict) else "" password = str(body.get("password", "")).strip() if isinstance(body, dict) else "" name = str(body.get("name", "")).strip() if isinstance(body, dict) else "" LOGGER.info("[REGISTER] attempt email=%s name=%s", email or "", name or "") if not email or not password: LOGGER.warning("[REGISTER] rejected missing credentials email=%s", email or "") return make_error("Missing email or password", status=400) if not name: LOGGER.warning("[REGISTER] rejected missing name email=%s", email or "") return make_error("Missing name", status=400) if self.storage.find_user_by_email(email): LOGGER.warning("[REGISTER] duplicate email=%s", email) return make_error("User already exists", status=409) user = { "id": str(uuid.uuid4()), "email": email, "nickname": name, "name": name, "password": password, "createdAt": now_iso(), } with self.storage.lock: self.storage.users.append(user) self.storage.save_users() LOGGER.info("[REGISTER] success email=%s user_id=%s", email, user["id"]) return 201, {"data": {"id": user["id"], "email": user["email"], "name": user["name"]}, "id": user["id"], "email": user["email"], "name": user["name"]} def me(self, handler: BaseHTTPRequestHandler) -> tuple[int, dict[str, Any]]: user = self.current_user(handler) if not user: LOGGER.warning("[ME] unauthorized path=%s", handler.path) return make_error("Unauthorized", status=401) LOGGER.info("[ME] success user_id=%s email=%s", user["id"], user["email"]) payload = {"id": user["id"], "email": user["email"], "name": user.get("name") or user.get("nickname") or user["email"]} return 200, {"data": payload, **payload} def matches_filter(self, game: dict[str, Any], filters: list[str]) -> bool: for expression in filters: if ":" not in expression: continue parts = expression.split(":") if len(parts) < 3: continue field_expr = parts[0] op = parts[1] value = ":".join(parts[2:]) field_path = field_expr.split("->") base = game if field_path[0] == "metadata": base = game.get("metadata", {}) field_path = field_path[1:] actual = nested_get(base, field_path) if len(field_path) > 1 else base.get(field_path[0]) actual_str = json.dumps(actual, ensure_ascii=False) if isinstance(actual, (dict, list)) else str(actual) if field_expr == "id" and op == "eq" and game.get("id") != value: return False if op == "ct" and value not in actual_str: return False if op == "nct" and value in actual_str: return False if op == "eq" and actual_str != value: return False if op == "ne" and actual_str == value: return False if op in {"gt", "gte", "lt", "lte"}: actual_cmp = actual_str value_cmp = value if op == "gt" and not (actual_cmp > value_cmp): return False if op == "gte" and not (actual_cmp >= value_cmp): return False if op == "lt" and not (actual_cmp < value_cmp): return False if op == "lte" and not (actual_cmp <= value_cmp): return False return True def get_games(self, handler: BaseHTTPRequestHandler) -> tuple[int, dict[str, Any]]: query = parse_qs(urlparse(handler.path).query) filters = query.get("filter[]", []) per_page = int(query.get("perPage", ["200"])[-1] or "200") sort_items = query.get("sort[]", []) games = [self.repair_game_metadata_for_listing(game) for game in self.storage.games.values()] filtered = [game for game in games if self.matches_filter(game, filters)] for sort_expr in reversed(sort_items): sort_parts = sort_expr.split(":") if len(sort_parts) < 2: continue field_expr, direction = sort_parts[0], sort_parts[1].upper() field_path = field_expr.split("->") reverse = direction in {"DSC", "DESC"} def sort_value(game: dict[str, Any]) -> str: base = game local_path = list(field_path) if local_path and local_path[0] == "metadata": base = game.get("metadata", {}) local_path = local_path[1:] if not local_path: return "" if len(local_path) > 1: return str(nested_get(base, local_path) or "") return str(base.get(local_path[0], "")) filtered.sort( key=sort_value, reverse=reverse, ) deduped: dict[str, dict[str, Any]] = {} for game in filtered: key = game_dedupe_key(game) or game.get("id", "") existing = deduped.get(key) if existing is None: deduped[key] = game continue existing_players = real_player_count(normalize_player_metadata_slots(existing.get("metadata", {}))) current_players = real_player_count(normalize_player_metadata_slots(game.get("metadata", {}))) existing_updated = str(existing.get("updated_at") or "") current_updated = str(game.get("updated_at") or "") if current_players > existing_players or ( current_players == existing_players and current_updated >= existing_updated ): deduped[key] = game deduped_games = list(deduped.values()) result = deduped_games if per_page == 0 else deduped_games[:per_page] response_items = [] skipped_ids = [] for game in result: metadata = game.get("metadata", {}) if not is_client_safe_game_metadata(metadata): skipped_ids.append(game.get("id")) continue response_items.append( { "id": game["id"], "metadata": metadata, "created_at": game.get("created_at"), "updated_at": game.get("updated_at"), } ) LOGGER.info( "[GAME] list filters=%s sort=%s per_page=%s matched=%s returned=%s", filters, sort_items, per_page, len(deduped_games), len(response_items), ) if skipped_ids: LOGGER.warning("[GAME] list skipped malformed game ids=%s", skipped_ids) LOGGER.info( "[GAME] list response preview=%s", json.dumps(response_items[:2], ensure_ascii=False, default=str)[:2000], ) return make_response(response_items) def create_game(self, handler: BaseHTTPRequestHandler, body: Any) -> tuple[int, dict[str, Any]]: user = self.current_user(handler) if not user: LOGGER.warning("[GAME] create unauthorized") return make_error("Unauthorized", status=401) payload = coerce_payload_dict(body) if not payload: LOGGER.warning("[GAME] create invalid payload user_id=%s", user["id"]) return make_error("Invalid payload") LOGGER.info( "[GAME] create payload user_id=%s keys=%s body=%s", user["id"], sorted(payload.keys()), json.dumps(payload, ensure_ascii=False, default=str)[:1000], ) game_id = str( payload.get("GameId") or payload.get("gameId") or payload.get("id") or payload.get("Id") or "" ).strip() if not game_id: game_id = str(uuid.uuid4()) LOGGER.info("[GAME] create auto-generated game_id=%s user_id=%s", game_id, user["id"]) with self.storage.lock: if game_id in self.storage.games: LOGGER.warning("[GAME] create duplicate game_id=%s user_id=%s", game_id, user["id"]) return make_error("Game already exists", status=409) cfg = payload.get("GameStartConfiguration") or payload.get("gameStartConfiguration") or {} num_players = int( cfg.get("NumberOfPlayersAsInt") or cfg.get("numberOfPlayersAsInt") or payload.get("NumberOfPlayersAsInt") or payload.get("numberOfPlayersAsInt") or payload.get("playersCount") or 4 ) player_entry = { "PlayerNetworkId": user["id"], "email": user["email"], "name": user.get("name") or user.get("nickname"), "nickname": user.get("nickname") or user.get("name"), "createdAt": now_iso(), "LastSeenEventRevision": 0, } metadata = payload.get("metadata", payload) if "PlayerMetadata" not in metadata: metadata["PlayerMetadata"] = [player_entry] if "NetworkGameProgressInformation" not in metadata: metadata["NetworkGameProgressInformation"] = {"CurrentGameStatus": "Open"} if "GameStartConfiguration" not in metadata: metadata["GameStartConfiguration"] = { "IsPrivate": bool(cfg.get("IsPrivate", cfg.get("isPrivate", False))), "NumberOfPlayersAsInt": num_players, } sync_network_game_status(metadata) game = { "id": game_id, "metadata": metadata, "events": [], "created_at": now_iso(), "updated_at": now_iso(), } new_key = game_dedupe_key(game) to_remove = [] for existing_id, existing_game in self.storage.games.items(): if existing_id == game_id: continue if game_dedupe_key(existing_game) == new_key: to_remove.append(existing_id) for existing_id in to_remove: LOGGER.warning("[GAME] create removing stale duplicate game_id=%s", existing_id) self.storage.games.pop(existing_id, None) self.storage.games[game_id] = game self.storage.save_games() LOGGER.info("[GAME] created game_id=%s owner_user_id=%s", game_id, user["id"]) return 201, { "id": game_id, "data": { "id": game_id, "metadata": game.get("metadata", {}), "created_at": game.get("created_at"), "updated_at": game.get("updated_at"), }, } def join_game(self, handler: BaseHTTPRequestHandler, game_id: str, body: Any) -> tuple[int, dict[str, Any]]: user = self.current_user(handler) if not user: LOGGER.warning("[GAME] join unauthorized game_id=%s", game_id) return make_error("Unauthorized", status=401) with self.storage.lock: game = self.storage.games.get(game_id) if not game: LOGGER.warning("[GAME] join missing game game_id=%s user_id=%s", game_id, user["id"]) return make_error("Game not found", status=404) metadata = game.setdefault("metadata", {}) players = metadata.setdefault("PlayerMetadata", []) for existing in players: if isinstance(existing, dict) and existing.get("PlayerNetworkId") == user["id"]: LOGGER.info("[GAME] join skipped already joined game_id=%s user_id=%s", game_id, user["id"]) return 200, { "id": game_id, "data": { "id": game_id, "metadata": game.get("metadata", {}), "created_at": game.get("created_at"), "updated_at": game.get("updated_at"), }, } payload = coerce_payload_dict(body) join_payload = payload.get("metadata", payload) if payload else {} if not join_payload.get("PlayerNetworkId"): join_payload["PlayerNetworkId"] = user["id"] if not join_payload.get("NickName"): join_payload["NickName"] = user.get("name") or user.get("nickname") or user["email"] if not join_payload.get("GameId"): join_payload["GameId"] = game_id if not join_payload.get("LastSeenEventRevision"): join_payload["LastSeenEventRevision"] = 0 slot_index = -1 for idx, existing in enumerate(players): if existing is None: slot_index = idx break if slot_index >= 0: players[slot_index] = join_payload else: players.append(join_payload) sync_network_game_status(metadata) game["updated_at"] = now_iso() self.storage.save_games() LOGGER.info( "[GAME] joined game_id=%s user_id=%s status=%s", game_id, user["id"], nested_get(metadata, ["NetworkGameProgressInformation", "CurrentGameStatus"]), ) return 200, { "id": game_id, "data": { "id": game_id, "metadata": game.get("metadata", {}), "created_at": game.get("created_at"), "updated_at": game.get("updated_at"), }, } def leave_game(self, handler: BaseHTTPRequestHandler, game_id: str, body: Any) -> tuple[int, dict[str, Any]]: user = self.current_user(handler) if not user: LOGGER.warning("[GAME] leave unauthorized game_id=%s", game_id) return make_error("Unauthorized", status=401) with self.storage.lock: game = self.storage.games.get(game_id) if not game: LOGGER.warning("[GAME] leave missing game game_id=%s user_id=%s", game_id, user["id"]) return make_error("Game not found", status=404) metadata = game.setdefault("metadata", {}) players = metadata.get("PlayerMetadata", []) if isinstance(players, dict) and "$values" in players: players = players["$values"] metadata["PlayerMetadata"] = players new_players = [] for entry in players: if not isinstance(entry, dict): continue if entry.get("PlayerNetworkId") == user["id"]: continue new_players.append(entry) metadata["PlayerMetadata"] = new_players sync_network_game_status(metadata) game["updated_at"] = now_iso() self.storage.save_games() LOGGER.info("[GAME] left game_id=%s user_id=%s", game_id, user["id"]) return make_response(True) def get_events(self, game_id: str) -> tuple[int, dict[str, Any]]: game = self.storage.games.get(game_id) if not game: return make_error("Game not found", status=404) return make_response(game.get("events", [])) def get_game_snapshot(self, game_id: str) -> tuple[int, dict[str, Any]]: game = self.storage.games.get(game_id) if not game: return make_error("Game not found", status=404) return make_response( { "id": game_id, "metadata": game.get("metadata", {}), "events": game.get("events", []), "created_at": game.get("created_at"), "updated_at": game.get("updated_at"), } ) def save_events_and_metadata(self, game_id: str, body: Any) -> tuple[int, dict[str, Any]]: game = self.storage.games.setdefault( game_id, {"id": game_id, "metadata": {}, "events": [], "created_at": now_iso(), "updated_at": now_iso()}, ) if isinstance(body, dict): if "events" in body: incoming_events = normalize_events_payload(body["events"]) max_revision = max((event_revision(evt) for evt in game.get("events", [])), default=0) for item in incoming_events: if not isinstance(item, dict): continue event = dict(item) if not event_revision(event): max_revision += 1 set_event_revision(event, max_revision) else: max_revision = max(max_revision, event_revision(event)) game["events"].append(event) if "metadata" in body and isinstance(body["metadata"], dict): game["metadata"] = body["metadata"] sync_network_game_status(game["metadata"]) sync_progress_from_events(game) game["updated_at"] = now_iso() self.storage.save_games() LOGGER.info( "[EVENTS_META] saved game_id=%s event_count=%s current_player_id=%s status=%s", game_id, len(game.get("events", [])), nested_get(game.get("metadata", {}), ["NetworkGameProgressInformation", "CurrentPlayerId"]), nested_get(game.get("metadata", {}), ["NetworkGameProgressInformation", "CurrentGameStatus"]), ) return make_response(True) def save_events(self, game_id: str, body: Any) -> tuple[int, dict[str, Any]]: if not isinstance(body, dict): return make_error("Invalid payload") game = self.storage.games.get(game_id) if not game: return make_error("Game not found", status=404) incoming = normalize_events_payload(body.get("events", body.get("data", body))) if not isinstance(incoming, list): return make_error("events must be a list") with self.storage.lock: events = game.setdefault("events", []) max_revision = max((event_revision(evt) for evt in events), default=0) for item in incoming: if not isinstance(item, dict): continue event = dict(item) if not event_revision(event): max_revision += 1 set_event_revision(event, max_revision) else: max_revision = max(max_revision, event_revision(event)) events.append(event) if event.get("$type") == "GameStarted": game.setdefault("metadata", {}).setdefault( "NetworkGameProgressInformation", {} )["CurrentGameStatus"] = "Started" if "metadata" in body and isinstance(body["metadata"], dict): game["metadata"] = body["metadata"] sync_network_game_status(game["metadata"]) sync_progress_from_events(game) game["updated_at"] = now_iso() ensure_game_metadata_shape(game) self.storage.save_games() LOGGER.info( "[EVENTS] saved game_id=%s count=%s max_revision=%s current_player_id=%s", game_id, len(incoming), max_revision, nested_get(game.get("metadata", {}), ["NetworkGameProgressInformation", "CurrentPlayerId"]), ) return make_response({"saved": len(incoming), "maxRevision": max_revision}) def delete_events(self, game_id: str) -> tuple[int, dict[str, Any]]: with self.storage.lock: game = self.storage.games.get(game_id) if not game: return make_error("Game not found", status=404) game["events"] = [] game["updated_at"] = now_iso() self.storage.save_games() return make_message_response("ok") def update_last_seen( self, handler: BaseHTTPRequestHandler, game_id: str, body: Any ) -> tuple[int, dict[str, Any]]: user = self.current_user(handler) if not user: return make_error("Unauthorized", status=401) game = self.storage.games.get(game_id) if not game: return make_error("Game not found", status=404) revision = normalize_revision( body.get("LastSeenEventRevision") if isinstance(body, dict) else 0 ) with self.storage.lock: metadata = game.setdefault("metadata", {}) players = metadata.setdefault("PlayerMetadata", []) for player in players: if not isinstance(player, dict): continue if player.get("PlayerNetworkId") == user["id"]: player["LastSeenEventRevision"] = revision game["updated_at"] = now_iso() self.storage.save_games() return make_response(True) return make_error("Player not found in game", status=404) class BrassRequestHandler(BaseHTTPRequestHandler): api: BrassAPI def _send(self, status: int, payload: Any) -> None: data = json.dumps(payload, ensure_ascii=False).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(data))) self.end_headers() self.wfile.write(data) def do_GET(self) -> None: # noqa: N802 self.handle_request() def do_POST(self) -> None: # noqa: N802 self.handle_request() def do_PUT(self) -> None: # noqa: N802 self.handle_request() def do_DELETE(self) -> None: # noqa: N802 self.handle_request() def log_message(self, format: str, *args: Any) -> None: return def handle_request(self) -> None: try: status, payload = self.api.route(self) except Exception as exc: # noqa: BLE001 LOGGER.exception("[REQ] unhandled error method=%s path=%s", self.command, self.path) status, payload = make_error(f"Internal server error: {exc}", 500) LOGGER.info("[RES] %s %s status=%s remote=%s", self.command, self.path, status, self.client_address[0]) self._send(status, payload) def build_server(host: str, port: int) -> ThreadingHTTPServer: storage = Storage.load() api = BrassAPI(storage) class Handler(BrassRequestHandler): pass Handler.api = api server = ThreadingHTTPServer((host, port), Handler) return server def main() -> int: setup_logging() parser = argparse.ArgumentParser() parser.add_argument("--host", default="0.0.0.0") parser.add_argument("--port", type=int, default=SERVER_PORT) args = parser.parse_args() storage = Storage.load() print("=" * 60) print(" Brass Birmingham Server") print(f"Loaded: {len(storage.users)} users, {len(storage.games)} games") print(f"Local: http://{get_local_ip()}:{args.port}") print("=" * 60) server = build_server(args.host, args.port) server.serve_forever() return 0 if __name__ == "__main__": raise SystemExit(main())