path: root/py_modules/lsfg_vk/installation.py

"""Installation and in-place migration service for lsfg-vk v2."""

import json
import platform
import shutil
import tarfile
import tempfile
import traceback
from pathlib import Path
from typing import Any, Dict

from .base_service import BaseService
from .config_schema import ConfigurationManager, ProfileData
from .constants import (
    ARMADA_DEVICE_ENV,
    BIN_DIR,
    JSON_FILENAME,
    LEGACY_JSON_FILENAME,
    LEGACY_LIB_FILENAME,
    LIB_FILENAME,
    get_layer_archive_filename,
)
from .types import InstallationCheckResponse, InstallationResponse, UninstallationResponse


class InstallationService(BaseService):
    """Install architecture-matched v2 assets and migrate the public config."""

    def __init__(self, logger=None):
        super().__init__(logger)
        self.lib_file = self.local_lib_dir / LIB_FILENAME
        self.json_file = self.local_share_dir / JSON_FILENAME
        self.legacy_lib_file = self.local_lib_dir / LEGACY_LIB_FILENAME
        self.legacy_json_file = self.local_share_dir / LEGACY_JSON_FILENAME
        self._config_recovery_backup: Path | None = None

    def install(self) -> InstallationResponse:
        try:
            plugin_dir = Path(__file__).parent.parent.parent
            archive_name = get_layer_archive_filename(self._is_arm_architecture())
            archive_path = plugin_dir / BIN_DIR / archive_name
            if not archive_path.exists():
                error_msg = f"{archive_name} not found at {archive_path}"
                self.log.error(error_msg)
                return self._error_response(InstallationResponse, error_msg, message="")

            self._ensure_directories()
            self._extract_and_install_files(archive_path)
            profile_data = self._create_config_file()
            self._create_lsfg_launch_script(profile_data)
            self._remove_legacy_layer_files()

            message = "lsfg-vk v2 installed successfully"
            if self._config_recovery_backup is not None:
                message += f"; unsupported config backed up to {self._config_recovery_backup.name}"
            self.log.info("lsfg-vk v2 installed successfully from %s", archive_name)
            return self._success_response(InstallationResponse, message)
        except (OSError, tarfile.TarError, shutil.Error) as error:
            self.log.error("Error installing lsfg-vk: %s", error)
            return self._error_response(InstallationResponse, str(error), message="")
        except Exception as error:
            self.log.error("Unexpected error installing lsfg-vk: %s", error)
            return self._error_response(InstallationResponse, str(error), message="")

    def _is_arm_architecture(self) -> bool:
        """Detect the native game host, including Decky running under FEX on Armada."""
        if platform.machine().lower() in ("aarch64", "arm64"):
            return True
        if ARMADA_DEVICE_ENV.is_file():
            self.log.info("Detected native AArch64 Armada host through device-env")
            return True
        try:
            with Path("/proc/1/exe").open("rb") as host_executable:
                elf_header = host_executable.read(20)
            if elf_header[:4] == b"\x7fELF" and elf_header[5] in (1, 2):
                byte_order = "little" if elf_header[5] == 1 else "big"
                if int.from_bytes(elf_header[18:20], byte_order) == 183:
                    self.log.info("Detected native AArch64 host through PID 1")
                    return True
        except OSError as error:
            self.log.debug("Could not inspect native host architecture: %s", error)
        return False

    @staticmethod
    def _copy_plugin_file(source: Path, destination: Path, mode: int = 0o644) -> None:
        """Copy payload content without preserving FEX-incompatible metadata."""
        shutil.copyfile(source, destination)
        destination.chmod(mode)

    def _extract_and_install_files(self, archive_path: Path) -> None:
        """Extract just the v2 library and manifest from an owner release archive."""
        destinations = {
            LIB_FILENAME: self.lib_file,
            JSON_FILENAME: self.json_file,
        }
        found = set()
        with tarfile.open(archive_path, "r:xz") as archive:
            with tempfile.TemporaryDirectory() as temp_dir:
                temp_root = Path(temp_dir)
                for member in archive.getmembers():
                    if not member.isfile():
                        continue
                    filename = Path(member.name).name
                    destination = destinations.get(filename)
                    if destination is None:
                        continue
                    source = archive.extractfile(member)
                    if source is None:
                        continue
                    temporary_file = temp_root / filename
                    with source, temporary_file.open("wb") as output:
                        shutil.copyfileobj(source, output)
                    destination.parent.mkdir(parents=True, exist_ok=True)
                    if filename == JSON_FILENAME:
                        self._copy_and_fix_json_file(temporary_file, destination)
                    else:
                        self._copy_plugin_file(temporary_file, destination)
                    found.add(filename)
                    self.log.info("Installed %s to %s", filename, destination)

        missing = [name for name in destinations if name not in found]
        if missing:
            raise OSError("Archive did not contain required lsfg-vk files: " + ", ".join(missing))

    def _copy_and_fix_json_file(self, source: Path, destination: Path) -> None:
        """Point the manifest at ~/.local/lib after Decky installs it."""
        try:
            data = json.loads(source.read_text(encoding="utf-8"))
            data["layer"]["library_path"] = "../../../lib/liblsfg-vk-layer.so"
            destination.write_text(json.dumps(data, indent=2) + "\n", encoding="utf-8")
            destination.chmod(0o644)
        except (json.JSONDecodeError, KeyError, OSError) as error:
            self.log.error("Error fixing JSON file %s: %s", source, error)
            self._copy_plugin_file(source, destination)

    def _create_config_file(self) -> ProfileData:
        """Migrate v1 once, normalize v2, and retain Decky launch workarounds."""
        from .dll_detection import DllDetectionService

        dll_service = DllDetectionService(self.log)
        profile_data: ProfileData
        was_legacy = False
        self._config_recovery_backup = None
        if self.config_file_path.exists():
            content = self.config_file_path.read_text(encoding="utf-8")
            was_legacy = ConfigurationManager.is_legacy_v1(content)
            try:
                profile_data = ConfigurationManager.parse_toml_content_multi_profile(content)
            except (ValueError, KeyError, TypeError) as error:
                self._config_recovery_backup = self._backup_config(content, "unrecognized")
                self.log.warning(
                    "Could not parse existing config; saved it to %s and using defaults: %s",
                    self._config_recovery_backup,
                    error,
                )
                default = ConfigurationManager.get_defaults_with_dll_detection(dll_service)
                profile_data = ProfileData(
                    current_profile="decky-lsfg-vk",
                    profiles={"decky-lsfg-vk": default},
                    global_config={"dll": default["dll"], "allow_fp16": default["allow_fp16"]},
                )
            if was_legacy:
                self._backup_legacy_config(content)
        else:
            default = ConfigurationManager.get_defaults_with_dll_detection(dll_service)
            profile_data = ProfileData(
                current_profile="decky-lsfg-vk",
                profiles={"decky-lsfg-vk": default},
                global_config={"dll": default["dll"], "allow_fp16": default["allow_fp16"]},
            )

        profile_data = self._merge_config_with_defaults(profile_data, dll_service)
        script_values = self._read_script_values()
        current_profile = profile_data["current_profile"]
        profile_data["profiles"][current_profile] = ConfigurationManager.merge_config_with_script(
            profile_data["profiles"][current_profile], script_values
        )
        self._write_file(
            self.config_file_path,
            ConfigurationManager.generate_toml_content_multi_profile(profile_data),
            0o644,
        )
        return profile_data

    def _backup_config(self, content: str, label: str) -> Path:
        backup_path = self.config_file_path.with_name(f"{self.config_file_path.name}.{label}.bak")
        if not backup_path.exists():
            self._write_file(backup_path, content, 0o644)
            self.log.info("Backed up %s configuration to %s", label, backup_path)
        return backup_path

    def _backup_legacy_config(self, content: str) -> None:
        self._backup_config(content, "v1")

    def _read_script_values(self) -> Dict[str, Any]:
        if not self.lsfg_launch_script_path.exists():
            return {}
        try:
            return ConfigurationManager.parse_script_content(
                self.lsfg_launch_script_path.read_text(encoding="utf-8")
            )
        except OSError as error:
            self.log.warning("Could not preserve launch-script settings: %s", error)
            return {}

    def _create_lsfg_launch_script(self, profile_data: ProfileData = None) -> None:
        """Write the v2 launcher after config migration has completed."""
        from .configuration import ConfigurationService

        if profile_data is None:
            profile_data = ConfigurationManager.parse_toml_content_multi_profile(
                self.config_file_path.read_text(encoding="utf-8")
            )
        configuration_service = ConfigurationService(logger=self.log)
        configuration_service.user_home = self.user_home
        configuration_service.config_dir = self.config_dir
        configuration_service.config_file_path = self.config_file_path
        configuration_service.lsfg_script_path = self.lsfg_launch_script_path
        script = configuration_service._generate_script_content_for_profile(profile_data)
        self._write_file(self.lsfg_launch_script_path, script, 0o755)

    def _merge_config_with_defaults(self, existing: ProfileData, dll_service) -> ProfileData:
        defaults = ConfigurationManager.get_defaults_with_dll_detection(dll_service)
        global_config = dict(existing.get("global_config", {}))
        global_config.setdefault("dll", defaults["dll"])
        global_config.setdefault("allow_fp16", defaults["allow_fp16"])

        profiles: Dict[str, Any] = {}
        for name, existing_config in existing.get("profiles", {}).items():
            profiles[name] = ConfigurationManager.validate_config({
                **defaults,
                **existing_config,
                **global_config,
            })
        if not profiles:
            profiles["decky-lsfg-vk"] = ConfigurationManager.validate_config(defaults)

        current_profile = existing.get("current_profile", "decky-lsfg-vk")
        if current_profile not in profiles:
            current_profile = "decky-lsfg-vk" if "decky-lsfg-vk" in profiles else next(iter(profiles))
        return ProfileData(
            current_profile=current_profile,
            profiles=profiles,
            global_config=global_config,
        )

    def _remove_legacy_layer_files(self) -> None:
        for path in (self.legacy_lib_file, self.legacy_json_file):
            if self._remove_if_exists(path):
                self.log.info("Removed superseded v1 layer file %s", path)

    def get_launch_script_path(self) -> str:
        return str(self.lsfg_launch_script_path)

    def check_installation(self) -> InstallationCheckResponse:
        try:
            lib_exists = self.lib_file.exists()
            json_exists = self.json_file.exists()
            legacy_installed = self.legacy_lib_file.exists() or self.legacy_json_file.exists()
            return {
                "installed": lib_exists and json_exists,
                "legacy_installed": legacy_installed,
                "lib_exists": lib_exists,
                "json_exists": json_exists,
                "script_exists": self.config_file_path.exists(),
                "lib_path": str(self.lib_file),
                "json_path": str(self.json_file),
                "script_path": str(self.config_file_path),
                "error": None,
            }
        except OSError as error:
            return {
                "installed": False,
                "legacy_installed": False,
                "lib_exists": False,
                "json_exists": False,
                "script_exists": False,
                "lib_path": str(self.lib_file),
                "json_path": str(self.json_file),
                "script_path": str(self.config_file_path),
                "error": str(error),
            }

    def uninstall(self) -> UninstallationResponse:
        try:
            removed = []
            for path in (
                self.lib_file,
                self.json_file,
                self.legacy_lib_file,
                self.legacy_json_file,
                self.lsfg_launch_script_path,
            ):
                if self._remove_if_exists(path):
                    removed.append(str(path))
            if not removed:
                return self._success_response(UninstallationResponse, "No lsfg-vk files found to remove", removed_files=None)
            return self._success_response(
                UninstallationResponse,
                f"lsfg-vk uninstalled successfully. Removed {len(removed)} files.",
                removed_files=removed,
            )
        except OSError as error:
            return self._error_response(UninstallationResponse, str(error), message="", removed_files=None)

    def cleanup_on_uninstall(self) -> None:
        try:
            self.uninstall()
        except Exception as error:
            self.log.error("Error cleaning up lsfg-vk files during uninstall: %s", error)
            self.log.error("Traceback: %s", traceback.format_exc())