path: root/py_modules/lsfg_vk/flatpak_service.py

import os
import shutil
import subprocess
from pathlib import Path
from typing import Any, Dict, List

from .base_service import BaseService
from .config_schema import ConfigurationManager
from .dll_detection import DllDetectionService
from .types import BaseResponse


class FlatpakService(BaseService):
    EXTENSION_ID = "org.freedesktop.Platform.VulkanLayer.lsfgvk"
    SUPPORTED_RUNTIMES = ("24.08", "25.08")

    def __init__(self, logger=None):
        super().__init__(logger)
        self.flatpak_command = None

    def _get_clean_env(self) -> Dict[str, str]:
        env = os.environ.copy()
        env.pop("LD_LIBRARY_PATH", None)
        path_entries = [entry for entry in env.get("PATH", "").split(":") if entry]
        for entry in ("/usr/bin", "/usr/local/bin", "/bin"):
            if entry not in path_entries:
                path_entries.insert(0, entry)
        env["PATH"] = ":".join(path_entries)
        return env

    def check_flatpak_available(self) -> bool:
        env = self._get_clean_env()
        self.flatpak_command = shutil.which("flatpak", path=env["PATH"])
        return self.flatpak_command is not None

    def _run_flatpak_command(self, args: List[str], **kwargs):
        if self.flatpak_command is None and not self.check_flatpak_available():
            raise FileNotFoundError("Flatpak command not available")
        return subprocess.run(
            [self.flatpak_command, *args],
            env=self._get_clean_env(),
            **kwargs,
        )

    @classmethod
    def _extension_ref(cls, version: str) -> str:
        return f"{cls.EXTENSION_ID}/x86_64/{version}"

    @classmethod
    def _validate_runtime(cls, version: str) -> None:
        if version not in cls.SUPPORTED_RUNTIMES:
            raise ValueError("Unsupported Flatpak runtime")

    def get_extension_status(self) -> Dict[str, Any]:
        try:
            if not self.check_flatpak_available():
                raise FileNotFoundError("Flatpak is not available on this system")

            result = self._run_flatpak_command(
                ["list", "--user", "--runtime", "--columns=application,arch,branch"],
                capture_output=True,
                text=True,
                check=True,
            )
            installed = {
                tuple(line.split("\t")[:3])
                for line in result.stdout.splitlines()
                if line.strip()
            }
            return self._success_response(
                BaseResponse,
                "Flatpak runtime status retrieved",
                installed_24_08=(self.EXTENSION_ID, "x86_64", "24.08") in installed,
                installed_25_08=(self.EXTENSION_ID, "x86_64", "25.08") in installed,
            )
        except Exception as error:
            return self._error_response(
                BaseResponse,
                str(error),
                installed_24_08=False,
                installed_25_08=False,
            )

    def install_extension(self, version: str) -> Dict[str, Any]:
        try:
            self._validate_runtime(version)
            if not self.check_flatpak_available():
                raise FileNotFoundError("Flatpak is not available on this system")
            result = self._run_flatpak_command(
                [
                    "install",
                    "--user",
                    "--noninteractive",
                    "--or-update",
                    "flathub",
                    f"{self.EXTENSION_ID}//{version}",
                ],
                capture_output=True,
                text=True,
            )
            if result.returncode != 0:
                raise OSError(result.stderr.strip() or "Flatpak installation failed")
            return self._success_response(
                BaseResponse,
                f"lsfg-vk {version} runtime extension installed",
            )
        except Exception as error:
            return self._error_response(BaseResponse, str(error))

    def uninstall_extension(self, version: str) -> Dict[str, Any]:
        try:
            self._validate_runtime(version)
            if not self.check_flatpak_available():
                raise FileNotFoundError("Flatpak is not available on this system")
            result = self._run_flatpak_command(
                ["uninstall", "--user", "--noninteractive", self._extension_ref(version)],
                capture_output=True,
                text=True,
            )
            if result.returncode != 0:
                raise OSError(result.stderr.strip() or "Flatpak uninstall failed")
            return self._success_response(
                BaseResponse,
                f"lsfg-vk {version} runtime extension uninstalled",
            )
        except Exception as error:
            return self._error_response(BaseResponse, str(error))

    def _dll_directory(self) -> Path:
        if self.config_file_path.exists():
            try:
                profile_data = ConfigurationManager.parse_toml_content_multi_profile(
                    self.config_file_path.read_text(encoding="utf-8")
                )
                dll_path = profile_data["global_config"].get("dll")
                if dll_path:
                    return Path(dll_path).parent
            except Exception:
                pass

        result = DllDetectionService(self.log).check_lossless_scaling_dll()
        if result.get("detected") and result.get("path"):
            return Path(result["path"]).parent
        return self.user_home / ".local/share/Steam/steamapps/common"

    def _override_output(self, app_id: str) -> str:
        result = self._run_flatpak_command(
            ["override", "--user", "--show", app_id],
            capture_output=True,
            text=True,
        )
        return result.stdout if result.returncode == 0 else ""

    def _override_paths(self) -> Dict[str, str]:
        return {
            "config_dir": str(self.config_dir),
            "config_file": str(self.config_file_path),
            "dll_dir": str(self._dll_directory()),
            "legacy_dll": str(
                self.user_home
                / ".local/share/Steam/steamapps/common/Lossless Scaling/Lossless.dll"
            ),
            "legacy_script": str(self.lsfg_launch_script_path),
        }

    def _check_app_override_status(self, app_id: str) -> Dict[str, bool]:
        output = self._override_output(app_id)
        paths = self._override_paths()
        return {
            "filesystem": (
                paths["config_dir"] in output
                and paths["dll_dir"] in output
            ),
            "env": f"LSFGVK_CONFIG={paths['config_file']}" in output,
        }

    def get_flatpak_apps(self) -> Dict[str, Any]:
        try:
            if not self.check_flatpak_available():
                raise FileNotFoundError("Flatpak is not available on this system")
            result = self._run_flatpak_command(
                ["list", "--user", "--app", "--columns=name,application"],
                capture_output=True,
                text=True,
                check=True,
            )
            apps = []
            for line in result.stdout.splitlines():
                parts = line.split("\t", 1)
                if len(parts) != 2:
                    continue
                status = self._check_app_override_status(parts[1])
                apps.append(
                    {
                        "app_id": parts[1],
                        "app_name": parts[0],
                        "has_filesystem_override": status["filesystem"],
                        "has_env_override": status["env"],
                    }
                )
            return self._success_response(
                BaseResponse,
                f"Found {len(apps)} Flatpak applications",
                apps=apps,
                total_apps=len(apps),
            )
        except Exception as error:
            return self._error_response(
                BaseResponse,
                str(error),
                apps=[],
                total_apps=0,
            )

    def set_app_override(self, app_id: str) -> Dict[str, Any]:
        try:
            if not self.check_flatpak_available():
                raise FileNotFoundError("Flatpak is not available on this system")
            paths = self._override_paths()
            result = self._run_flatpak_command(
                [
                    "override",
                    "--user",
                    f"--filesystem={paths['config_dir']}:rw",
                    f"--filesystem={paths['dll_dir']}:ro",
                    f"--env=LSFGVK_CONFIG={paths['config_file']}",
                    f"--nofilesystem={paths['legacy_dll']}",
                    f"--nofilesystem={paths['legacy_script']}",
                    "--unset-env=LSFG_CONFIG",
                    app_id,
                ],
                capture_output=True,
                text=True,
            )
            if result.returncode != 0:
                raise OSError(result.stderr.strip() or "Failed to set Flatpak overrides")
            return self._success_response(
                BaseResponse,
                f"lsfg-vk overrides set for {app_id}",
                app_id=app_id,
                operation="set",
            )
        except Exception as error:
            return self._error_response(
                BaseResponse,
                str(error),
                app_id=app_id,
                operation="set",
            )

    def remove_app_override(self, app_id: str) -> Dict[str, Any]:
        try:
            if not self.check_flatpak_available():
                raise FileNotFoundError("Flatpak is not available on this system")
            paths = self._override_paths()
            result = self._run_flatpak_command(
                [
                    "override",
                    "--user",
                    f"--nofilesystem={paths['config_dir']}",
                    f"--nofilesystem={paths['dll_dir']}",
                    f"--nofilesystem={paths['legacy_dll']}",
                    f"--nofilesystem={paths['legacy_script']}",
                    "--unset-env=LSFGVK_CONFIG",
                    "--unset-env=LSFG_CONFIG",
                    app_id,
                ],
                capture_output=True,
                text=True,
            )
            if result.returncode != 0:
                raise OSError(result.stderr.strip() or "Failed to remove Flatpak overrides")
            return self._success_response(
                BaseResponse,
                f"lsfg-vk overrides removed for {app_id}",
                app_id=app_id,
                operation="remove",
            )
        except Exception as error:
            return self._error_response(
                BaseResponse,
                str(error),
                app_id=app_id,
                operation="remove",
            )

    def migrate_v2(self) -> None:
        if not self.check_flatpak_available():
            return

        status = self.get_extension_status()
        for version, key in (
            ("24.08", "installed_24_08"),
            ("25.08", "installed_25_08"),
        ):
            if not status.get(key):
                continue
            result = self._run_flatpak_command(
                ["update", "--user", "--noninteractive", self._extension_ref(version)],
                capture_output=True,
                text=True,
            )
            if result.returncode != 0:
                self.log.warning(result.stderr.strip())

        apps_result = self._run_flatpak_command(
            ["list", "--user", "--app", "--columns=application"],
            capture_output=True,
            text=True,
        )
        if apps_result.returncode != 0:
            return

        for app_id in apps_result.stdout.splitlines():
            app_id = app_id.strip()
            if not app_id:
                continue
            if "LSFG_CONFIG=" in self._override_output(app_id):
                result = self.set_app_override(app_id)
                if not result.get("success"):
                    self.log.warning(result.get("error"))