1333 lines
52 KiB
Python
1333 lines
52 KiB
Python
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 "<empty>")
|
|
if not email or not password:
|
|
LOGGER.warning("[LOGIN] rejected missing credentials email=%s", email or "<empty>")
|
|
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 "<empty>", name or "<empty>")
|
|
if not email or not password:
|
|
LOGGER.warning("[REGISTER] rejected missing credentials email=%s", email or "<empty>")
|
|
return make_error("Missing email or password", status=400)
|
|
if not name:
|
|
LOGGER.warning("[REGISTER] rejected missing name email=%s", email or "<empty>")
|
|
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())
|