languard-servers-manager/backend/adapters/arma3/mission_manager.py

"""Arma 3 mission manager — handles .pbo mission files, upload, delete, rotation."""
from __future__ import annotations

import logging
import os
import re
from pathlib import Path

from pydantic import BaseModel

from adapters.exceptions import AdapterError
from core.utils.file_utils import get_server_dir, sanitize_filename, safe_delete_file

logger = logging.getLogger(__name__)

_MISSIONS_DIR = "mpmissions"
_ALLOWED_EXTENSION = ".pbo"
_MAX_MISSION_SIZE_MB = 500


class Arma3MissionData(BaseModel):
    """Mission data schema for Arma 3."""
    terrain: str = ""
    difficulty: str = "Regular"


class Arma3MissionManager:
    file_extension = ".pbo"

    def __init__(self, server_id: int | None = None) -> None:
        self._server_id = server_id

    def _missions_dir(self) -> Path:
        return get_server_dir(self._server_id) / _MISSIONS_DIR

    # ── File operations ──

    def list_missions(self) -> list[dict]:
        """
        Scan the mpmissions directory and return all .pbo files.

        Returns list of dicts:
            name: str       — filename without extension
            filename: str   — full filename
            size_bytes: int — file size
        """
        missions_dir = self._missions_dir()
        if not missions_dir.exists():
            return []

        missions = []
        try:
            for entry in missions_dir.iterdir():
                if entry.is_file() and entry.suffix.lower() == _ALLOWED_EXTENSION:
                    missions.append({
                        "name": entry.stem,
                        "filename": entry.name,
                        "size_bytes": entry.stat().st_size,
                    })
        except OSError as exc:
            raise AdapterError(f"Cannot list missions: {exc}") from exc

        missions.sort(key=lambda m: m["filename"].lower())
        return missions

    def upload_mission(self, filename: str, content: bytes) -> dict:
        """
        Save a mission file to the mpmissions directory.

        Args:
            filename: Original filename from the upload (will be sanitized).
            content: Raw file bytes.

        Returns the saved mission dict.
        """
        safe_name = sanitize_filename(filename)
        if not safe_name.lower().endswith(_ALLOWED_EXTENSION):
            raise AdapterError(
                f"Invalid mission file extension. Only {_ALLOWED_EXTENSION} files are allowed."
            )

        size_mb = len(content) / (1024 * 1024)
        if size_mb > _MAX_MISSION_SIZE_MB:
            raise AdapterError(
                f"Mission file too large ({size_mb:.1f} MB). Max is {_MAX_MISSION_SIZE_MB} MB."
            )

        missions_dir = self._missions_dir()
        missions_dir.mkdir(parents=True, exist_ok=True)

        dest_path = missions_dir / safe_name

        # Atomic write: write to .tmp first, then replace
        tmp_path = str(dest_path) + ".tmp"
        try:
            with open(tmp_path, "wb") as f:
                f.write(content)
            os.replace(tmp_path, str(dest_path))
        except OSError as exc:
            safe_delete_file(Path(tmp_path))
            raise AdapterError(f"Cannot save mission file: {exc}") from exc

        logger.info(
            "Mission uploaded for server %d: %s (%d bytes)",
            self._server_id, safe_name, len(content),
        )
        return {
            "name": dest_path.stem,
            "filename": safe_name,
            "size_bytes": len(content),
        }

    def delete_mission(self, filename: str) -> bool:
        """
        Delete a mission file.
        Returns True if deleted, False if not found.
        """
        safe_name = sanitize_filename(filename)
        if not safe_name.lower().endswith(_ALLOWED_EXTENSION):
            raise AdapterError("Invalid mission filename")

        dest_path = self._missions_dir() / safe_name

        # Verify resolved path is inside missions directory (path traversal guard)
        try:
            dest_path.resolve().relative_to(self._missions_dir().resolve())
        except ValueError:
            raise AdapterError("Path traversal detected in filename")

        if not dest_path.exists():
            return False

        try:
            dest_path.unlink()
            logger.info("Mission deleted for server %d: %s", self._server_id, safe_name)
            return True
        except OSError as exc:
            raise AdapterError(f"Cannot delete mission: {exc}") from exc

    # ── Mission rotation config ──

    def parse_mission_filename(self, filename: str) -> dict:
        """
        Parse Arma 3 mission filename.
        Format: MissionName.Terrain.pbo
        """
        name = filename
        if name.endswith(self.file_extension):
            name = name[: -len(self.file_extension)]

        parts = name.rsplit(".", 1)
        if len(parts) == 2:
            return {
                "mission_name": parts[0],
                "terrain": parts[1],
                "filename": filename,
            }
        return {
            "mission_name": name,
            "terrain": "",
            "filename": filename,
        }

    def get_rotation_config(self, rotation_entries: list[dict]) -> str:
        """
        Generate Arma 3 mission rotation config block.
        rotation_entries: list of {mission_name, terrain, difficulty, params_json}
        """
        if not rotation_entries:
            return ""

        lines = ['class Missions {']
        for i, entry in enumerate(rotation_entries):
            mission = entry.get("mission_name", "")
            terrain = entry.get("terrain", "")
            difficulty = entry.get("difficulty", "Regular")
            params = entry.get("params_json", "{}")
            lines.append(f'    class Mission_{i} {{')
            lines.append(f'        template = "{mission}.{terrain}";')
            lines.append(f'        difficulty = "{difficulty}";')
            if params and params != "{}":
                lines.append(f'        params = {params};')
            lines.append('    };')
        lines.append('};')
        return "\n".join(lines)

    def get_missions_dir(self, server_dir: Path) -> Path:
        return server_dir / _MISSIONS_DIR

    def get_mission_data_schema(self) -> type[BaseModel] | None:
        return Arma3MissionData