path: root/py_modules/lsfg_vk/plugin.py

"""
Main plugin class for the lsfg-vk Decky Loader plugin.

This plugin provides services for installing and managing the lsfg-vk 
Vulkan layer for frame generation on Steam Deck.
"""

import os
import json
import subprocess
import urllib.request
import ssl
import hashlib
from typing import Dict, Any
from pathlib import Path

import decky

from .installation import InstallationService
from .dll_detection import DllDetectionService
from .configuration import ConfigurationService
from .config_schema import ConfigurationManager
from .flatpak_service import FlatpakService


class Plugin:
    """
    Main plugin class for lsfg-vk management.
    
    This class provides a unified interface for installation, configuration,
    and DLL detection services. It implements the Decky Loader plugin lifecycle
    methods (_main, _unload, _uninstall, _migration).
    """
    
    def __init__(self):
        """Initialize the plugin with all necessary services"""
        self.installation_service = InstallationService()
        self.dll_detection_service = DllDetectionService()
        self.configuration_service = ConfigurationService()
        self.flatpak_service = FlatpakService()

    async def install_lsfg_vk(self) -> Dict[str, Any]:
        """Install lsfg-vk by extracting the zip file to ~/.local
        
        Returns:
            InstallationResponse dict with success status and message/error
        """
        return self.installation_service.install()

    async def check_lsfg_vk_installed(self) -> Dict[str, Any]:
        """Check if lsfg-vk is already installed
        
        Returns:
            InstallationCheckResponse dict with installation status and paths
        """
        return self.installation_service.check_installation()

    async def uninstall_lsfg_vk(self) -> Dict[str, Any]:
        """Uninstall lsfg-vk by removing the installed files
        
        Returns:
            UninstallationResponse dict with success status and removed files
        """
        return self.installation_service.uninstall()

    async def check_lossless_scaling_dll(self) -> Dict[str, Any]:
        """Check if Lossless Scaling DLL is available at the expected paths
        
        Returns:
            DllDetectionResponse dict with detection status and path info
        """
        return self.dll_detection_service.check_lossless_scaling_dll()

    async def check_lossless_scaling_dll_and_update_config(self) -> Dict[str, Any]:
        """Check for DLL and automatically update configuration if found
        
        This method should only be used during installation or when explicitly
        requested by the user, not for routine DLL detection checks.
        
        Returns:
            DllDetectionResponse dict with detection status and path info
        """
        result = self.dll_detection_service.check_lossless_scaling_dll()
        
        result_dict = dict(result)
        
        if result.get("detected") and result.get("path"):
            try:
                dll_path = result["path"]
                if dll_path:
                    update_result = self.configuration_service.update_dll_path(dll_path)
                    if update_result.get("success"):
                        result_dict["config_updated"] = True
                        result_dict["message"] = f"DLL detected and configuration updated: {dll_path}"
                    else:
                        result_dict["config_updated"] = False
                        result_dict["message"] = f"DLL detected but config update failed: {update_result.get('error', 'Unknown error')}"
            except Exception as e:
                result_dict["config_updated"] = False
                result_dict["message"] = f"DLL detected but config update failed: {str(e)}"
        
        return result_dict

    async def get_dll_stats(self) -> Dict[str, Any]:
        """Get detailed statistics about the detected DLL
        
        Returns:
            Dict containing DLL path, SHA256 hash, and other stats
        """
        try:
            dll_result = self.dll_detection_service.check_lossless_scaling_dll()
            
            if not dll_result.get("detected") or not dll_result.get("path"):
                return {
                    "success": False,
                    "error": "DLL not detected",
                    "dll_path": None,
                    "dll_sha256": None
                }
            
            dll_path = dll_result["path"]
            if dll_path is None:
                return {
                    "success": False,
                    "error": "DLL path is None",
                    "dll_path": None,
                    "dll_sha256": None
                }
            
            dll_path_obj = Path(dll_path)
            
            sha256_hash = hashlib.sha256()
            try:
                with open(dll_path_obj, "rb") as f:
                    for chunk in iter(lambda: f.read(4096), b""):
                        sha256_hash.update(chunk)
                dll_sha256 = sha256_hash.hexdigest()
            except Exception as e:
                return {
                    "success": False,
                    "error": f"Failed to calculate SHA256: {str(e)}",
                    "dll_path": dll_path,
                    "dll_sha256": None
                }
            
            return {
                "success": True,
                "dll_path": dll_path,
                "dll_sha256": dll_sha256,
                "dll_source": dll_result.get("source"),
                "error": None
            }
            
        except Exception as e:
            return {
                "success": False,
                "error": f"Failed to get DLL stats: {str(e)}",
                "dll_path": None,
                "dll_sha256": None
            }

    async def get_lsfg_config(self) -> Dict[str, Any]:
        """Read current lsfg script configuration
        
        Returns:
            ConfigurationResponse dict with current configuration or error
        """
        return self.configuration_service.get_config()

    async def get_config_schema(self) -> Dict[str, Any]:
        """Get configuration schema information for frontend
        
        Returns:
            Dict with field names, types, defaults, and profile information
        """
        try:
            profiles_response = self.configuration_service.get_profiles()
            
            schema_data = {
                "field_names": ConfigurationManager.get_field_names(),
                "field_types": {name: field_type.value for name, field_type in ConfigurationManager.get_field_types().items()},
                "defaults": ConfigurationManager.get_defaults()
            }
            
            if profiles_response.get("success"):
                schema_data["profiles"] = profiles_response.get("profiles", [])
                schema_data["current_profile"] = profiles_response.get("current_profile")
            else:
                schema_data["profiles"] = ["decky-lsfg-vk"]
                schema_data["current_profile"] = "decky-lsfg-vk"
            
            return schema_data
            
        except (ValueError, KeyError, AttributeError) as e:
            self.configuration_service.log.warning(f"Failed to get full schema, using fallback: {e}")
            return {
                "field_names": ConfigurationManager.get_field_names(),
                "field_types": {name: field_type.value for name, field_type in ConfigurationManager.get_field_types().items()},
                "defaults": ConfigurationManager.get_defaults(),
                "profiles": ["decky-lsfg-vk"],
                "current_profile": "decky-lsfg-vk"
            }

    async def update_lsfg_config(self, config: Dict[str, Any]) -> Dict[str, Any]:
        """Update lsfg TOML configuration using object-based API (single source of truth)
        
        Args:
            config: Configuration data dictionary containing all settings
            
        Returns:
            ConfigurationResponse dict with success status
        """
        validated_config = ConfigurationManager.validate_config(config)
        
        return self.configuration_service.update_config_from_dict(validated_config)

    async def update_dll_path(self, dll_path: str) -> Dict[str, Any]:
        """Update the DLL path in the configuration when detected
        
        Args:
            dll_path: Path to the detected Lossless.dll file
            
        Returns:
            ConfigurationResponse dict with success status
        """
        return self.configuration_service.update_dll_path(dll_path)

    async def get_profiles(self) -> Dict[str, Any]:
        """Get list of all profiles and current profile
        
        Returns:
            ProfilesResponse dict with profile list and current profile
        """
        return self.configuration_service.get_profiles()

    async def create_profile(self, profile_name: str, source_profile: str = None) -> Dict[str, Any]:
        """Create a new profile
        
        Args:
            profile_name: Name for the new profile
            source_profile: Optional source profile to copy from (default: current profile)
            
        Returns:
            ProfileResponse dict with success status
        """
        return self.configuration_service.create_profile(profile_name, source_profile)

    async def delete_profile(self, profile_name: str) -> Dict[str, Any]:
        """Delete a profile
        
        Args:
            profile_name: Name of the profile to delete
            
        Returns:
            ProfileResponse dict with success status
        """
        return self.configuration_service.delete_profile(profile_name)

    async def rename_profile(self, old_name: str, new_name: str) -> Dict[str, Any]:
        """Rename a profile
        
        Args:
            old_name: Current profile name
            new_name: New profile name
            
        Returns:
            ProfileResponse dict with success status
        """
        return self.configuration_service.rename_profile(old_name, new_name)

    async def set_current_profile(self, profile_name: str) -> Dict[str, Any]:
        """Set the current active profile
        
        Args:
            profile_name: Name of the profile to set as current
            
        Returns:
            ProfileResponse dict with success status
        """
        return self.configuration_service.set_current_profile(profile_name)

    async def update_profile_config(self, profile_name: str, config: Dict[str, Any]) -> Dict[str, Any]:
        """Update configuration for a specific profile
        
        Args:
            profile_name: Name of the profile to update
            config: Configuration data dictionary containing settings
            
        Returns:
            ConfigurationResponse dict with success status
        """
        validated_config = ConfigurationManager.validate_config(config)
        
        return self.configuration_service.update_profile_config(profile_name, validated_config)

    async def check_for_plugin_update(self) -> Dict[str, Any]:
        """Check for plugin updates by comparing current version with most recent GitHub release
        
        Checks for the most recent release including pre-releases, not just the latest stable.
        
        Returns:
            Dict containing update information:
            {
                "update_available": bool,
                "current_version": str,
                "latest_version": str,
                "release_notes": str,
                "release_date": str,
                "download_url": str,
                "error": str (if error occurred)
            }
        """
        try:
            package_json_path = Path(decky.DECKY_PLUGIN_DIR) / "package.json"
            current_version = "0.0.0"
            
            if package_json_path.exists():
                try:
                    with open(package_json_path, 'r', encoding='utf-8') as f:
                        package_data = json.load(f)
                        current_version = package_data.get('version', '0.0.0')
                except Exception as e:
                    decky.logger.warning(f"Failed to read package.json: {e}")
            
            api_url = "https://api.github.com/repos/xXJSONDeruloXx/decky-lsfg-vk/releases"
            
            try:
                ssl_context = ssl.create_default_context()
                ssl_context.check_hostname = False
                ssl_context.verify_mode = ssl.CERT_NONE
                
                with urllib.request.urlopen(api_url, context=ssl_context) as response:
                    releases_data = json.loads(response.read().decode('utf-8'))
                
                if not releases_data:
                    raise Exception("No releases found")
                    
                release_data = releases_data[0]
                
                latest_version = release_data.get('tag_name', '').lstrip('v')
                release_notes = release_data.get('body', '')
                release_date = release_data.get('published_at', '')
                
                download_url = ""
                for asset in release_data.get('assets', []):
                    if asset.get('name', '').endswith('.zip'):
                        download_url = asset.get('browser_download_url', '')
                        break
                
                update_available = self._compare_versions(current_version, latest_version)
                
                return {
                    "success": True,
                    "update_available": update_available,
                    "current_version": current_version,
                    "latest_version": latest_version,
                    "release_notes": release_notes,
                    "release_date": release_date,
                    "download_url": download_url
                }
                
            except Exception as e:
                decky.logger.error(f"Failed to fetch release info: {e}")
                return {
                    "success": False,
                    "error": f"Failed to check for updates: {str(e)}"
                }
                
        except Exception as e:
            return {
                "success": False,
                "error": f"Update check failed: {str(e)}"
            }

    async def download_plugin_update(self, download_url: str) -> Dict[str, Any]:
        """Download the plugin update zip file to ~/Downloads
        
        Args:
            download_url: URL to download the plugin zip from
            
        Returns:
            Dict containing download result:
            {
                "success": bool,
                "download_path": str,
                "error": str (if error occurred)
            }
        """
        try:
            downloads_dir = Path.home() / "Downloads"
            downloads_dir.mkdir(exist_ok=True)
            download_path = downloads_dir / "decky-lsfg-vk.zip"
            
            if download_path.exists():
                download_path.unlink()
            
            decky.logger.info(f"Downloading plugin update from {download_url}")
            
            try:
                ssl_context = ssl.create_default_context()
                ssl_context.check_hostname = False
                ssl_context.verify_mode = ssl.CERT_NONE
                
                with urllib.request.urlopen(download_url, context=ssl_context) as response:
                    with open(download_path, 'wb') as f:
                        f.write(response.read())
                
                if download_path.exists() and download_path.stat().st_size > 0:
                    decky.logger.info(f"Plugin update downloaded successfully to {download_path}")
                    return {
                        "success": True,
                        "download_path": str(download_path)
                    }
                else:
                    return {
                        "success": False,
                        "error": "Download completed but file is empty or missing"
                    }
                    
            except Exception as e:
                decky.logger.error(f"Download failed: {e}")
                return {
                    "success": False,
                    "error": f"Download failed: {str(e)}"
                }
                
        except Exception as e:
            return {
                "success": False,
                "error": f"Download preparation failed: {str(e)}"
            }

    def _compare_versions(self, current: str, latest: str) -> bool:
        """Compare two version strings to determine if an update is available
        
        Args:
            current: Current version string (e.g., "1.2.3")
            latest: Latest version string (e.g., "1.2.4")
            
        Returns:
            True if latest version is newer than current version
        """
        try:
            current_parts = current.lstrip('v').split('.')
            latest_parts = latest.lstrip('v').split('.')
            
            max_len = max(len(current_parts), len(latest_parts))
            current_parts.extend(['0'] * (max_len - len(current_parts)))
            latest_parts.extend(['0'] * (max_len - len(latest_parts)))
            
            for i in range(max_len):
                try:
                    current_num = int(current_parts[i])
                    latest_num = int(latest_parts[i])
                    
                    if latest_num > current_num:
                        return True
                    elif latest_num < current_num:
                        return False
                except ValueError:
                    if latest_parts[i] > current_parts[i]:
                        return True
                    elif latest_parts[i] < current_parts[i]:
                        return False
            
            return False
            
        except (IndexError, AttributeError, TypeError) as e:
            self.configuration_service.log.warning(f"Version comparison failed: {e}")
            return False

    async def get_launch_option(self) -> Dict[str, Any]:
        """Get the launch option that users need to set for their games
        
        Returns:
            Dict containing the launch option string and instructions
        """
        return {
            "launch_option": "~/lsfg %command%",
            "instructions": "Add this to your game's launch options in Steam Properties",
            "explanation": "The lsfg script is created during installation and sets up the environment for the plugin"
        }

    async def get_config_file_content(self) -> Dict[str, Any]:
        """Get the current config file content
        
        Returns:
            Dict containing the config file content or error message
        """
        try:
            config_path = self.configuration_service.config_file_path
            if not config_path.exists():
                return {
                    "success": False,
                    "content": None,
                    "path": str(config_path),
                    "error": "Config file does not exist"
                }
            
            content = config_path.read_text(encoding='utf-8')
            return {
                "success": True,
                "content": content,
                "path": str(config_path),
                "error": None
            }
        except Exception as e:
            return {
                "success": False,
                "content": None,
                "path": str(config_path) if 'config_path' in locals() else "unknown",
                "error": f"Error reading config file: {str(e)}"
            }

    async def get_launch_script_content(self) -> Dict[str, Any]:
        """Get the content of the launch script file
        
        Returns:
            FileContentResponse dict with file content or error information
        """
        try:
            script_path = self.installation_service.get_launch_script_path()
            
            if not os.path.exists(script_path):
                return {
                    "success": False,
                    "error": f"Launch script not found at {script_path}",
                    "path": str(script_path)
                }
            
            with open(script_path, 'r') as file:
                content = file.read()
                
            return {
                "success": True,
                "content": content,
                "path": str(script_path)
            }
            
        except Exception as e:
            decky.logger.error(f"Error reading launch script: {e}")
            return {
                "success": False,
                "error": str(e)
            }

    async def check_fgmod_directory(self) -> Dict[str, Any]:
        """Check if the fgmod directory exists in the home directory
        
        Returns:
            Dict with exists status and directory path
        """
        try:
            home_path = Path(decky.DECKY_USER_HOME)
            fgmod_path = home_path / "fgmod"
            
            exists = fgmod_path.exists() and fgmod_path.is_dir()
            
            return {
                "success": True,
                "exists": exists,
                "path": str(fgmod_path)
            }
            
        except Exception as e:
            decky.logger.error(f"Error checking fgmod directory: {e}")
            return {
                "success": False,
                "exists": False,
                "error": str(e)
            }

    async def check_flatpak_extension_status(self) -> Dict[str, Any]:
        """Check status of lsfg-vk Flatpak runtime extensions
        
        Returns:
            FlatpakExtensionStatus dict with installation status for both runtime versions
        """
        return self.flatpak_service.get_extension_status()

    async def install_flatpak_extension(self, version: str) -> Dict[str, Any]:
        """Install lsfg-vk Flatpak runtime extension
        
        Args:
            version: Runtime version to install ("23.08" or "24.08")
            
        Returns:
            BaseResponse dict with success status and message/error
        """
        return self.flatpak_service.install_extension(version)

    async def uninstall_flatpak_extension(self, version: str) -> Dict[str, Any]:
        """Uninstall lsfg-vk Flatpak runtime extension
        
        Args:
            version: Runtime version to uninstall ("23.08" or "24.08")
            
        Returns:
            BaseResponse dict with success status and message/error
        """
        return self.flatpak_service.uninstall_extension(version)

    async def get_flatpak_apps(self) -> Dict[str, Any]:
        """Get list of installed Flatpak apps and their lsfg-vk override status
        
        Returns:
            FlatpakAppInfo dict with apps list and override status
        """
        return self.flatpak_service.get_flatpak_apps()

    async def set_flatpak_app_override(self, app_id: str) -> Dict[str, Any]:
        """Set lsfg-vk overrides for a Flatpak app
        
        Args:
            app_id: Flatpak application ID
            
        Returns:
            FlatpakOverrideResponse dict with operation result
        """
        return self.flatpak_service.set_app_override(app_id)

    async def remove_flatpak_app_override(self, app_id: str) -> Dict[str, Any]:
        """Remove lsfg-vk overrides for a Flatpak app
        
        Args:
            app_id: Flatpak application ID
            
        Returns:
            FlatpakOverrideResponse dict with operation result
        """
        return self.flatpak_service.remove_app_override(app_id)
    
    async def _main(self):
        """
        Main entry point for the plugin.
        
        This method is called by Decky Loader when the plugin is loaded.
        Any initialization code should go here.
        """
        decky.logger.info("decky-lsfg-vk plugin loaded")

    async def _unload(self):
        """
        Cleanup tasks when the plugin is unloaded.
        
        This method is called by Decky Loader when the plugin is being unloaded.
        Any cleanup code should go here.
        """
        decky.logger.info("decky-lsfg-vk plugin unloaded")

    async def _uninstall(self):
        """
        Called when the plugin is uninstalled.
        
        This method is called by Decky Loader when the plugin is being uninstalled.
        Performs cleanup of plugin files and flatpak extensions.
        """
        decky.logger.info("decky-lsfg-vk plugin being uninstalled")
        
        # Clean up lsfg-vk files when the plugin is uninstalled
        self.installation_service.cleanup_on_uninstall()
        
        # Also clean up flatpak extensions if they are installed
        try:
            decky.logger.info("Checking for flatpak extensions to uninstall")
            
            extension_status = self.flatpak_service.get_extension_status()
            
            if extension_status.get("success"):
                if extension_status.get("installed_23_08"):
                    decky.logger.info("Uninstalling lsfg-vk flatpak runtime 23.08")
                    result = self.flatpak_service.uninstall_extension("23.08")
                    if result.get("success"):
                        decky.logger.info("Successfully uninstalled flatpak runtime 23.08")
                    else:
                        decky.logger.warning(f"Failed to uninstall flatpak runtime 23.08: {result.get('error')}")
                
                if extension_status.get("installed_24_08"):
                    decky.logger.info("Uninstalling lsfg-vk flatpak runtime 24.08")
                    result = self.flatpak_service.uninstall_extension("24.08")
                    if result.get("success"):
                        decky.logger.info("Successfully uninstalled flatpak runtime 24.08")
                    else:
                        decky.logger.warning(f"Failed to uninstall flatpak runtime 24.08: {result.get('error')}")
                        
                decky.logger.info("Flatpak extension cleanup completed")
            else:
                decky.logger.info(f"Could not check flatpak status for cleanup: {extension_status.get('error')}")
                
        except Exception as e:
            decky.logger.error(f"Error during flatpak cleanup: {e}")
        
        decky.logger.info("decky-lsfg-vk plugin uninstall cleanup completed")

    async def _migration(self):
        """
        Migrations that should be performed before entering `_main()`.
        
        This method is called by Decky Loader for plugin migrations.
        Currently migrates logs, settings, and runtime data from old locations.
        """
        decky.logger.info("Running decky-lsfg-vk plugin migrations")
        
        decky.migrate_logs(os.path.join(decky.DECKY_USER_HOME,
                                       ".config", "decky-lossless-scaling-vk", "lossless-scaling-vk.log"))
        
        decky.migrate_settings(
            os.path.join(decky.DECKY_HOME, "settings", "lossless-scaling-vk.json"),
            os.path.join(decky.DECKY_USER_HOME, ".config", "decky-lossless-scaling-vk"))
        
        decky.migrate_runtime(
            os.path.join(decky.DECKY_HOME, "lossless-scaling-vk"),
            os.path.join(decky.DECKY_USER_HOME, ".local", "share", "decky-lossless-scaling-vk"))
        
        decky.logger.info("decky-lsfg-vk plugin migrations completed")