""" ServerService — orchestrates all server lifecycle operations. Delegates game-specific work to the adapter. """ from __future__ import annotations import logging import shutil from pathlib import Path from fastapi import HTTPException, status from sqlalchemy.engine import Connection from adapters.registry import GameAdapterRegistry from core.dal.config_repository import ConfigRepository from core.dal.event_repository import EventRepository from core.dal.server_repository import ServerRepository from core.servers.process_manager import ProcessManager from core.utils.file_utils import ensure_server_dirs, get_server_dir logger = logging.getLogger(__name__) def _ok_response(data): return {"success": True, "data": data, "error": None} class ServerService: def __init__(self, db: Connection): self._db = db self._server_repo = ServerRepository(db) self._config_repo = ConfigRepository(db) self._event_repo = EventRepository(db) # ── CRUD ────────────────────────────────────────────────────────────────── def list_servers(self, game_type: str | None = None) -> list[dict]: """Return server list with live metrics merged in.""" servers = self._server_repo.get_all(game_type) return [self._enrich_server(s) for s in servers] def get_server(self, server_id: int) -> dict: server = self._server_repo.get_by_id(server_id) if server is None: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail={"code": "NOT_FOUND", "message": f"Server {server_id} not found"}, ) return self._enrich_server(server) def _enrich_server(self, server: dict) -> dict: """Add live CPU/RAM/player count from DB.""" from core.dal.metrics_repository import MetricsRepository from core.dal.player_repository import PlayerRepository result = dict(server) metrics = MetricsRepository(self._db).get_latest(server["id"]) if metrics: result["cpu_percent"] = metrics["cpu_percent"] result["ram_mb"] = metrics["ram_mb"] else: result["cpu_percent"] = None result["ram_mb"] = None result["player_count"] = PlayerRepository(self._db).count(server["id"]) return result def create_server( self, name: str, game_type: str, exe_path: str, game_port: int, rcon_port: int | None = None, description: str | None = None, auto_restart: bool = False, max_restarts: int = 3, ) -> dict: # Validate adapter exists try: adapter = GameAdapterRegistry.get(game_type) except KeyError: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail={"code": "GAME_TYPE_NOT_FOUND", "message": f"Unknown game type: {game_type}"}, ) # Validate exe process_config = adapter.get_process_config() exe_name = Path(exe_path).name if exe_name not in process_config.get_allowed_executables(): from adapters.exceptions import ExeNotAllowedError raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail={ "code": "EXE_NOT_ALLOWED", "message": f"Executable '{exe_name}' not allowed", "allowed": process_config.get_allowed_executables(), }, ) # Determine rcon_port if not provided if rcon_port is None: rcon_port = process_config.get_default_rcon_port(game_port) # Check port conflicts against running servers from core.utils.port_checker import check_ports_against_running_servers conflicts = check_ports_against_running_servers(game_port, rcon_port, None, self._db) if conflicts: raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail={ "code": "PORT_IN_USE", "message": f"Ports already in use: {conflicts}", }, ) # Create DB row server_id = self._server_repo.create( name=name, game_type=game_type, exe_path=exe_path, game_port=game_port, rcon_port=rcon_port, description=description, auto_restart=auto_restart, max_restarts=max_restarts, ) # Create directory layout layout = process_config.get_server_dir_layout() ensure_server_dirs(server_id, layout) # Seed default config sections config_gen = adapter.get_config_generator() schema_version = config_gen.get_config_version() for section in config_gen.get_sections(): defaults = config_gen.get_defaults(section) sensitive = config_gen.get_sensitive_fields(section) self._config_repo.upsert_section( server_id=server_id, game_type=game_type, section=section, config_data=defaults, schema_version=schema_version, sensitive_fields=sensitive, ) self._event_repo.insert(server_id, "created", actor="admin") return self.get_server(server_id) def update_server(self, server_id: int, **updates) -> dict: self.get_server(server_id) # raises 404 if not found filtered = {k: v for k, v in updates.items() if v is not None} if filtered: self._server_repo.update(server_id, **filtered) return self.get_server(server_id) def delete_server(self, server_id: int) -> None: server = self.get_server(server_id) if server["status"] not in ("stopped", "crashed", "error"): raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail={ "code": "SERVER_NOT_STOPPED", "message": "Server must be stopped before deletion", }, ) self._server_repo.delete(server_id) # Delete server directory server_dir = get_server_dir(server_id) if server_dir.exists(): shutil.rmtree(str(server_dir), ignore_errors=True) # ── Lifecycle ───────────────────────────────────────────────────────────── def start(self, server_id: int) -> dict: """ Full start sequence: 1. Load server + adapter 2. Validate exe 3. Check ports 4. Write config files (atomic) 5. Build launch args 6. Start process 7. Start monitoring threads 8. Return status """ from adapters.exceptions import ( ConfigWriteError, ExeNotAllowedError, LaunchArgsError, ConfigValidationError, ) from core.utils.port_checker import check_ports_against_running_servers server = self.get_server(server_id) if server["status"] in ("running", "starting"): raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail={"code": "SERVER_ALREADY_RUNNING", "message": "Server is already running"}, ) adapter = GameAdapterRegistry.get(server["game_type"]) process_config = adapter.get_process_config() config_gen = adapter.get_config_generator() # Validate exe exe_name = Path(server["exe_path"]).name if exe_name not in process_config.get_allowed_executables(): raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail={ "code": "EXE_NOT_ALLOWED", "message": f"Executable '{exe_name}' not in adapter allowlist", "allowed": process_config.get_allowed_executables(), }, ) # Check ports conflicts = check_ports_against_running_servers( server["game_port"], server.get("rcon_port"), server_id, self._db ) if conflicts: raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail={"code": "PORT_IN_USE", "message": f"Ports in use: {conflicts}"}, ) # Load config sections (decrypt sensitive fields for config generation) sensitive_by_section = { s: config_gen.get_sensitive_fields(s) for s in config_gen.get_sections() } sections = self._config_repo.get_all_sections(server_id, sensitive_by_section) # Remove _meta from each section before passing to adapter raw_sections = { section: {k: v for k, v in data.items() if k != "_meta"} for section, data in sections.items() } # Inject port into sections so build_launch_args can use it if "_port" not in raw_sections: raw_sections["_port"] = server["game_port"] # Get mod args if adapter supports mods mod_args: list[str] = [] if adapter.has_capability("mod_manager"): from sqlalchemy import text mods = self._db.execute( text(""" SELECT m.folder_path, sm.is_server_mod, sm.sort_order FROM server_mods sm JOIN mods m ON m.id = sm.mod_id WHERE sm.server_id = :sid ORDER BY sm.sort_order """), {"sid": server_id}, ).fetchall() mod_list = [dict(r._mapping) for r in mods] mod_args = adapter.get_mod_manager().build_mod_args(mod_list) # Write config files (atomic) server_dir = get_server_dir(server_id) try: config_gen.write_configs(server_id, server_dir, raw_sections) except ConfigWriteError as e: self._server_repo.update_status(server_id, "error") self._event_repo.insert(server_id, "config_write_error", detail={"error": str(e)}) raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail={"code": "CONFIG_WRITE_ERROR", "message": str(e)}, ) except ConfigValidationError as e: raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail={"code": "INVALID_CONFIG", "message": str(e), "errors": e.errors}, ) # Build launch args try: launch_args = config_gen.build_launch_args(raw_sections, mod_args) except LaunchArgsError as e: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail={"code": "INVALID_CONFIG", "message": str(e)}, ) # Start process pm = ProcessManager.get() with pm.get_operation_lock(server_id): pid = pm.start(server_id, server["exe_path"], launch_args, cwd=str(server_dir)) # Update DB from datetime import datetime, timezone self._server_repo.update_status( server_id, "starting", pid=pid, started_at=datetime.now(timezone.utc).isoformat() ) self._event_repo.insert(server_id, "started", detail={"pid": pid}) # Start monitoring threads try: from core.threads.thread_registry import ThreadRegistry ThreadRegistry.start_server_threads(server_id, self._db) except Exception as e: logger.warning("Could not start monitoring threads: %s", e) return {"status": "starting", "pid": pid} def stop(self, server_id: int, force: bool = False) -> dict: server = self.get_server(server_id) if server["status"] in ("stopped", "crashed"): raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail={"code": "SERVER_NOT_RUNNING", "message": "Server is not running"}, ) # Mark as "stopping" so ProcessMonitorThread doesn't treat this as a crash self._server_repo.update_status(server_id, "stopping") # Stop monitoring threads first so they don't fight with shutdown try: from core.threads.thread_registry import ThreadRegistry ThreadRegistry.stop_server_threads(server_id) except Exception as exc: logger.warning("Failed to stop monitoring threads for server %d during stop: %s", server_id, exc) # Try graceful shutdown via remote admin if not force: try: pm = ProcessManager.get() logger.info("Sending graceful shutdown to server %d", server_id) except Exception as e: logger.warning("Graceful shutdown failed: %s, falling back to terminate", e) pm = ProcessManager.get() with pm.get_operation_lock(server_id): exited = pm.stop(server_id, timeout=30) if not exited: logger.warning("Server %d did not exit in 30s, force-killing", server_id) pm.kill(server_id) from datetime import datetime, timezone self._server_repo.update_status( server_id, "stopped", pid=None, stopped_at=datetime.now(timezone.utc).isoformat() ) from core.dal.player_repository import PlayerRepository PlayerRepository(self._db).clear(server_id) self._event_repo.insert(server_id, "stopped") return {"status": "stopped"} def restart(self, server_id: int) -> dict: self.stop(server_id) return self.start(server_id) def kill(self, server_id: int) -> dict: server = self.get_server(server_id) # Mark as "stopping" so ProcessMonitorThread doesn't treat this as a crash self._server_repo.update_status(server_id, "stopping") # Stop monitoring threads first try: from core.threads.thread_registry import ThreadRegistry ThreadRegistry.stop_server_threads(server_id) except Exception as exc: logger.warning("Failed to stop monitoring threads for server %d during kill: %s", server_id, exc) pm = ProcessManager.get() with pm.get_operation_lock(server_id): pm.kill(server_id) from datetime import datetime, timezone self._server_repo.update_status(server_id, "stopped", pid=None, stopped_at=datetime.now(timezone.utc).isoformat()) from core.dal.player_repository import PlayerRepository PlayerRepository(self._db).clear(server_id) self._event_repo.insert(server_id, "killed") return {"status": "stopped"} # ── Config ──────────────────────────────────────────────────────────────── def get_config(self, server_id: int) -> dict: self.get_server(server_id) adapter = GameAdapterRegistry.get( self._server_repo.get_by_id(server_id)["game_type"] ) config_gen = adapter.get_config_generator() sensitive_by_section = { s: config_gen.get_sensitive_fields(s) for s in config_gen.get_sections() } sections = self._config_repo.get_all_sections(server_id, sensitive_by_section) # Mask sensitive fields in response (replace actual value with "***") for section, data in sections.items(): sf = config_gen.get_sensitive_fields(section) for field in sf: if field in data and data[field]: data[field] = "***" return sections def get_config_section(self, server_id: int, section: str) -> dict: server = self.get_server(server_id) adapter = GameAdapterRegistry.get(server["game_type"]) config_gen = adapter.get_config_generator() if section not in config_gen.get_sections(): raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail={"code": "NOT_FOUND", "message": f"Config section '{section}' not found"}, ) sensitive = config_gen.get_sensitive_fields(section) data = self._config_repo.get_section(server_id, section, sensitive) if data is None: data = config_gen.get_defaults(section) data["_meta"] = {"config_version": 0, "schema_version": config_gen.get_config_version()} # Mask sensitive fields for field in sensitive: if field in data and data[field]: data[field] = "***" return data def update_config_section( self, server_id: int, section: str, data: dict, expected_version: int | None = None, ) -> dict: server = self.get_server(server_id) adapter = GameAdapterRegistry.get(server["game_type"]) config_gen = adapter.get_config_generator() sections = config_gen.get_sections() if section not in sections: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail={"code": "NOT_FOUND", "message": f"Config section '{section}' not found"}, ) # Validate against adapter's Pydantic model model_cls = sections[section] # Get current values, merge with update (partial update support) current = self._config_repo.get_section( server_id, section, config_gen.get_sensitive_fields(section) ) if current: merged = {k: v for k, v in current.items() if k != "_meta"} else: merged = config_gen.get_defaults(section) # Apply updates for k, v in data.items(): if k not in ("_meta", "config_version"): merged[k] = v # Validate try: model_cls(**merged) except Exception as e: raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail={"code": "INVALID_CONFIG", "message": str(e)}, ) sensitive = config_gen.get_sensitive_fields(section) try: new_version = self._config_repo.upsert_section( server_id=server_id, game_type=server["game_type"], section=section, config_data=merged, schema_version=config_gen.get_config_version(), sensitive_fields=sensitive, expected_config_version=expected_version, ) except ValueError as e: error_msg = str(e) if "CONFIG_VERSION_CONFLICT" in error_msg: current_version = int(error_msg.split(":")[1]) current_data = self.get_config_section(server_id, section) raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail={ "code": "CONFIG_VERSION_CONFLICT", "message": "Config was modified by another user. Re-read and merge.", "current_config": current_data, "current_version": current_version, }, ) raise self._event_repo.insert( server_id, "config_updated", detail={"section": section, "version": new_version} ) return self.get_config_section(server_id, section) def get_config_preview(self, server_id: int) -> dict[str, str]: server = self.get_server(server_id) adapter = GameAdapterRegistry.get(server["game_type"]) config_gen = adapter.get_config_generator() sensitive_by_section = { s: config_gen.get_sensitive_fields(s) for s in config_gen.get_sections() } sections = self._config_repo.get_all_sections(server_id, sensitive_by_section) raw_sections = {k: {kk: vv for kk, vv in v.items() if kk != "_meta"} for k, v in sections.items()} server_dir = get_server_dir(server_id) return config_gen.preview_config(server_id, server_dir, raw_sections)