refactor: unify flatpak targets with steam profiles

1 files changed, 123 insertions, 16 deletions

diff --git a/py_modules/lsfg_vk/wrapper_service.py b/py_modules/lsfg_vk/wrapper_service.py
index a3cba6e..dae565b 100644
--- a/py_modules/lsfg_vk/wrapper_service.py
+++ b/py_modules/lsfg_vk/wrapper_service.py
@@ -16,8 +16,10 @@ from .constants import WRAPPER_FILENAME
 class WrapperService(BaseService):
     """Persist workaround state and compile it into a safe POSIX wrapper."""
 
-    FORMAT_VERSION = 1
-    MARKER = "# lsfg-vk-wrapper-format: 1"
+    LEGACY_FORMAT_VERSION = 1
+    FORMAT_VERSION = 2
+    LEGACY_MARKER = "# lsfg-vk-wrapper-format: 1"
+    MARKER = "# lsfg-vk-wrapper-format: 2"
     WRAPPER_TOKEN = "~/.lsfg"
     STATE_FIELDS = (
         "dxvkFrameRate",
@@ -82,12 +84,38 @@ class WrapperService(BaseService):
         return state
 
     @classmethod
+    def _validate_transport(cls, raw: Any) -> Dict[str, Any]:
+        if raw is None:
+            return {"kind": "host"}
+        if not isinstance(raw, dict):
+            raise ValueError("Workaround transport must be an object")
+        kind = raw.get("kind")
+        if kind == "host":
+            return {"kind": "host"}
+        if kind == "flatpak":
+            app_id = raw.get("flatpakAppId")
+            if (
+                not isinstance(app_id, str)
+                or not re.fullmatch(
+                    r"^[A-Za-z0-9][A-Za-z0-9-]*(?:\.[A-Za-z0-9][A-Za-z0-9-]*)+$",
+                    app_id,
+                )
+            ):
+                raise ValueError("Flatpak transport requires a valid application ID")
+            return {"kind": "flatpak", "flatpakAppId": app_id}
+        raise ValueError("Workaround transport must be host or flatpak")
+
+    @classmethod
     def _validate_entry(cls, raw: Any) -> Dict[str, Any]:
         if not isinstance(raw, dict):
             raise ValueError("Workaround AppID entry must be an object")
         entry = {
             "state": cls._validate_state(raw.get("state")),
             "command_token_added": raw.get("command_token_added", False),
+            # Version 1 entries had no transport field.  They are preserved as
+            # host entries until the shortcut is explicitly repaired with the
+            # backend's classified transport.
+            "transport": cls._validate_transport(raw.get("transport")),
         }
         if type(entry["command_token_added"]) is not bool:
             raise ValueError("command_token_added must be a boolean")
@@ -105,7 +133,10 @@ class WrapperService(BaseService):
 
     @classmethod
     def _validate_document(cls, raw: Any) -> Dict[str, Any]:
-        if not isinstance(raw, dict) or raw.get("version") != cls.FORMAT_VERSION:
+        if not isinstance(raw, dict) or raw.get("version") not in (
+            cls.LEGACY_FORMAT_VERSION,
+            cls.FORMAT_VERSION,
+        ):
             raise ValueError("Unsupported lsfg-vk workaround state version")
         apps = raw.get("apps")
         if not isinstance(apps, dict):
@@ -142,7 +173,7 @@ class WrapperService(BaseService):
             prefix = "\n".join(self.wrapper_path.read_text(encoding="utf-8").splitlines()[:8])
         except OSError as error:
             raise RuntimeError(f"Could not read lsfg wrapper: {error}") from error
-        return self.MARKER in prefix
+        return self.MARKER in prefix or self.LEGACY_MARKER in prefix
 
     def _assert_wrapper_owned_or_absent(self) -> bool:
         if not self.wrapper_path.exists() and not self.wrapper_path.is_symlink():
@@ -157,6 +188,17 @@ class WrapperService(BaseService):
     def _shell(value: str) -> str:
         return shlex.quote(value)
 
+    @staticmethod
+    def _direct_flatpak_tokens(value: str) -> Optional[list[str]]:
+        """Parse the supported full executable form: /usr/bin/flatpak run APP."""
+        try:
+            tokens = shlex.split(value, posix=True)
+        except ValueError:
+            return None
+        if len(tokens) >= 3 and Path(tokens[0]).name == "flatpak" and tokens[1] == "run":
+            return tokens
+        return None
+
     @classmethod
     def _state_lines(cls, state: Dict[str, Any], shortcut_exe: Optional[str]) -> list[str]:
         lines = ["    unset " + " ".join(cls.MANAGED_ENV_KEYS)]
@@ -194,9 +236,31 @@ class WrapperService(BaseService):
         lines.append(f"    shortcut_exe={cls._shell(shortcut_exe or '')}")
         return lines
 
-    @classmethod
-    def _flatpak_args(cls, state: Dict[str, Any]) -> list[str]:
+    def _dll_directory(self) -> Path:
+        if self.config_file_path.exists():
+            try:
+                content = self.config_file_path.read_text(encoding="utf-8")
+                match = re.search(
+                    r'(?m)^[ \t]*dll[ \t]*=[ \t]*"((?:\\.|[^"\\])*)"',
+                    content,
+                )
+                if match:
+                    configured_dll = json.loads('"' + match.group(1) + '"')
+                    if configured_dll:
+                        return Path(configured_dll).parent
+            except Exception:
+                pass
+        return self.user_home / ".local/share/Steam/steamapps/common/Lossless Scaling"
+
+    def _flatpak_args(self, state: Dict[str, Any]) -> list[str]:
+        config_dir = str(self.config_dir)
+        config_file = str(self.config_file_path)
+        dll_dir = str(self._dll_directory())
         args = [
+            self._shell(f"--filesystem={config_dir}:rw"),
+            self._shell(f"--filesystem={dll_dir}:ro"),
+            self._shell(f"--env=LSFGVK_CONFIG={config_file}"),
+            '"--env=LSFGVK_FLATPAK=1"',
             '"--env=SteamAppId=$appid"',
             '"--unset-env=DISABLE_GAMESCOPE_WSI"',
             '"--unset-env=ENABLE_GAMESCOPE_WSI"' if not state["disableGamescopeWsi"] else
@@ -230,11 +294,10 @@ class WrapperService(BaseService):
             "    fi",
         ]
 
-    @classmethod
-    def _render_wrapper(cls, document: Dict[str, Any]) -> str:
+    def _render_wrapper(self, document: Dict[str, Any]) -> str:
         lines = [
             "#!/bin/sh",
-            cls.MARKER,
+            self.MARKER,
             "# Generated by Decky LSFG-VK; edits will be rejected on the next update.",
             "",
             "appid=",
@@ -260,29 +323,61 @@ class WrapperService(BaseService):
         for appid in sorted(document["apps"], key=lambda value: int(value)):
             entry = document["apps"][appid]
             lines.append(f"  {appid})")
-            lines.extend(cls._state_lines(entry["state"], entry.get("shortcut_exe")))
+            lines.extend(self._state_lines(entry["state"], entry.get("shortcut_exe")))
             lines.append("    ;;")
         lines.extend([
             "esac",
             "",
             'if [ -n "$shortcut_exe" ]; then',
-            '  if [ "${1-}" = "run" ]; then',
-            '    flatpak_command="$1"',
-            "    shift",
         ])
         # The arguments are emitted per branch below so the values are static and
         # the wrapper never needs a JSON parser or another helper executable.
         lines.append('    case "$appid" in')
         for appid in sorted(document["apps"], key=lambda value: int(value)):
             entry = document["apps"][appid]
-            if not entry.get("shortcut_exe", "").endswith("/flatpak"):
+            transport = entry.get("transport", {"kind": "host"})
+            if transport.get("kind") != "flatpak":
                 continue
+            shortcut_exe = entry.get("shortcut_exe", "")
+            direct_flatpak_tokens = self._direct_flatpak_tokens(shortcut_exe)
+            if direct_flatpak_tokens is None and Path(shortcut_exe).name != "flatpak":
+                raise ValueError(
+                    f"Flatpak target {appid} does not use a direct flatpak executable"
+                )
             lines.append(f"      {appid})")
-            lines.extend(cls._flatpak_args(entry["state"]))
+            lines.extend([
+                *(
+                    [
+                        f"        shortcut_exe={self._shell(direct_flatpak_tokens[0])}",
+                        "        set -- "
+                        + " ".join(self._shell(token) for token in direct_flatpak_tokens[1:])
+                        + ' "$@"',
+                    ]
+                    if direct_flatpak_tokens
+                    else []
+                ),
+                '        if [ "${1-}" != "run" ]; then',
+                '          echo "lsfg-vk: Flatpak shortcut must use direct flatpak run transport" >&2',
+                "          exit 64",
+                "        fi",
+                '        flatpak_command="$1"',
+                "        shift",
+                "        flatpak_target=",
+                '        for flatpak_arg in "$@"; do',
+                '          case "$flatpak_arg" in',
+                '            -*) ;;',
+                '            *) flatpak_target="$flatpak_arg"; break ;;',
+                "          esac",
+                "        done",
+                f'        if [ "$flatpak_target" != {self._shell(transport["flatpakAppId"])} ]; then',
+                '          echo "lsfg-vk: Flatpak shortcut application ID changed externally" >&2',
+                "          exit 64",
+                "        fi",
+            ])
+            lines.extend(self._flatpak_args(entry["state"]))
             lines.append("        ;;")
         lines.extend([
             "    esac",
-            "  fi",
             '  exec "$shortcut_exe" "$@"',
             "fi",
             'exec "$@"',
@@ -328,6 +423,7 @@ class WrapperService(BaseService):
             "wrapper_owned": self._wrapper_marker() if document["apps"] else False,
             "shortcut_exe": entry.get("shortcut_exe") if entry else None,
             "command_token_added": entry.get("command_token_added", False) if entry else False,
+            "transport": dict(entry.get("transport", {"kind": "host"})) if entry else None,
         }
 
     def get(self, appid: str) -> Dict[str, Any]:
@@ -354,6 +450,7 @@ class WrapperService(BaseService):
         state: Dict[str, Any],
         shortcut_exe: Optional[str] = None,
         command_token_added: bool = False,
+        transport: Optional[Dict[str, Any]] = None,
     ) -> Dict[str, Any]:
         try:
             normalized = self._valid_appid(appid)
@@ -364,9 +461,19 @@ class WrapperService(BaseService):
                 self._assert_wrapper_owned_or_absent()
                 document, _, _ = self._read_document()
                 previous_entry = document["apps"].get(normalized)
+                selected_transport = self._validate_transport(
+                    transport
+                    if transport is not None
+                    else (
+                        previous_entry.get("transport")
+                        if previous_entry
+                        else None
+                    )
+                )
                 entry: Dict[str, Any] = {
                     "state": validated_state,
                     "command_token_added": bool(command_token_added),
+                    "transport": selected_transport,
                 }
                 if shortcut_exe is not None:
                     entry = self._validate_entry({**entry, "shortcut_exe": shortcut_exe})