pristontale-eu-linux/ptlinux/runtime.py

"""Portable game runtime selection and Proton/UMU preparation."""

from __future__ import annotations

from dataclasses import asdict, dataclass
from datetime import datetime
import hashlib
import json
import os
from pathlib import Path
import platform
import shutil
import sys
from typing import Mapping, Sequence

from .command import CommandOutcome, CommandRunner, OutputLineFilter
from .gamebox import build_gamebox_command
from .paths import AppPaths
from .settings import Settings


WINDOWS_GAME_EXE = r"C:\Program Files (x86)\Pristontale EU\Game.exe"
NATIVE_D3D43_DLLS = ("d3dx9_43", "d3dcompiler_43")
DXVK_D3D9_OVERRIDE = "native,builtin"
DLL_OVERRIDES_KEY = r"HKCU\Software\Wine\DllOverrides"
WINETRICKS_CODEPAGE_PROBE = (
	'winetricks_early_wine_arch cmd.exe /c "chcp 65001 > nul & echo %$1%"'
)
WINETRICKS_PROTON7_PROBE = 'winetricks_early_wine_arch cmd.exe /c "echo %$1%"'
RUNTIME_DEFAULT_SYNC = "runtime-default"
CACHYOS_MARKER = "cachyos"
CURRENT_UMU_DISTRIBUTIONS = frozenset(
	{"arch", "cachyos", "endeavouros", "garuda", "manjaro"}
)
GAME_RUNTIME_CONFIGURATION_VERSION = 1


class RuntimeErrorPT(RuntimeError):
	"""Raised when a supported game runtime cannot be prepared."""


@dataclass(frozen=True)
class RuntimeSelection:
	kind: str
	executable: str
	proton_path: str | None
	steam_root: str | None


def select_runtime(
	settings: Settings,
	environment: Mapping[str, str] | None = None,
	*,
	distribution_id: str | None = None,
	kernel_release: str | None = None,
) -> RuntimeSelection:
	env = os.environ if environment is None else environment
	umu = shutil.which("umu-run", path=env.get("PATH"))
	current_umu_host = prefers_current_umu_host(
		distribution_id=distribution_id,
		kernel_release=kernel_release,
	)
	auto_container = (
		settings.runtime == "auto"
		and bool(umu)
		and not current_umu_host
	)
	if settings.runtime == "umu":
		if umu:
			return RuntimeSelection("umu", umu, None, None)
		raise RuntimeErrorPT("UMU was requested but umu-run was not found")
	if settings.proton_path and settings.runtime in {
		"auto",
		"proton7",
		"proton7-container",
	}:
		proton = Path(settings.proton_path).expanduser().resolve()
		if not proton.is_file():
			raise RuntimeErrorPT(f"configured Proton script does not exist: {proton}")
		steam_root = _steam_root_for_proton(proton)
		if settings.runtime == "proton7-container":
			if not umu:
				raise RuntimeErrorPT(
					"containerized Proton 7 was requested but umu-run was not found"
				)
			return RuntimeSelection(
				"proton7-container",
				umu,
				str(proton),
				str(steam_root),
			)
		if auto_container:
			assert umu is not None
			return RuntimeSelection(
				"proton7-container",
				umu,
				str(proton),
				str(steam_root),
			)
		return RuntimeSelection("proton7", sys.executable, str(proton), str(steam_root))
	# Current UMU avoids several rolling-distribution container/media failures and
	# remains the broad Arch-family compatibility default. Direct Proton 7 is also
	# known to authenticate on CachyOS, so this is a compatibility policy rather
	# than an assertion that Proton 7 itself causes server-side Code 1/Code 3.
	if settings.runtime == "auto" and current_umu_host and umu:
		return RuntimeSelection("umu", umu, None, None)
	if settings.runtime in {"auto", "proton7", "proton7-container"}:
		proton_result = find_steam_proton7(env)
		if proton_result is not None:
			proton, steam_root = proton_result
			if settings.runtime == "proton7-container" or auto_container:
				if not umu:
					raise RuntimeErrorPT(
						"containerized Proton 7 was requested but umu-run was not found"
					)
				return RuntimeSelection(
					"proton7-container",
					umu,
					str(proton),
					str(steam_root),
				)
			return RuntimeSelection("proton7", sys.executable, str(proton), str(steam_root))
		if settings.runtime in {"proton7", "proton7-container"}:
			raise RuntimeErrorPT("Steam Proton 7.0 was requested but not found")
	if settings.runtime in {"auto", "umu"}:
		if umu:
			return RuntimeSelection("umu", umu, None, None)
		raise RuntimeErrorPT(
			"no supported game runtime found; install umu-launcher or Steam Proton 7.0"
		)
	raise RuntimeErrorPT(f"unsupported runtime selection: {settings.runtime}")


def find_steam_proton7(
	environment: Mapping[str, str] | None = None,
) -> tuple[Path, Path] | None:
	env = os.environ if environment is None else environment
	home_value = env.get("HOME")
	if not home_value:
		return None
	home = Path(home_value).expanduser()
	roots: list[Path] = []
	explicit = env.get("STEAM_COMPAT_CLIENT_INSTALL_PATH")
	if explicit:
		roots.append(Path(explicit).expanduser())
	roots.extend(
		(
			home / ".local/share/Steam",
			home / ".steam/steam",
			home / "snap/steam/common/.local/share/Steam",
			home / ".var/app/com.valvesoftware.Steam/.local/share/Steam",
		)
	)
	seen: set[Path] = set()
	for root in roots:
		resolved = root.resolve()
		if resolved in seen:
			continue
		seen.add(resolved)
		proton = resolved / "steamapps/common/Proton 7.0/proton"
		if proton.is_file():
			return proton, resolved
	return None


def prepare_game_runtime(
	paths: AppPaths,
	selection: RuntimeSelection,
	runner: CommandRunner,
) -> list[CommandOutcome]:
	if not runner.dry_run:
		paths.create_base_directories()
	results: list[CommandOutcome] = []
	if selection.kind == "umu":
		environment = _umu_environment(paths)
		if not native_d3d_ready(paths.umu_prefix):
			results.append(
				runner.run(
					[selection.executable, "winetricks", "-q", "d3dx9_43", "d3dcompiler_43"],
					env_overrides=environment,
					timeout_seconds=1800.0,
				)
			)
	elif selection.kind in {"proton7", "proton7-container"}:
		if not selection.proton_path or not selection.steam_root:
			raise RuntimeErrorPT("incomplete Proton 7 runtime selection")
		if not runner.dry_run and _game_runtime_ready(paths, selection):
			return results
		if not runner.dry_run:
			paths.compat_data.mkdir(parents=True, exist_ok=True)
		proton = Path(selection.proton_path)
		environment = _proton_environment(paths, selection)
		prefix_verb = "runinprefix" if _proton_prefix_ready(paths.proton_prefix) else "run"
		results.append(
			runner.run(
				[
					sys.executable,
					"-m",
					"ptlinux.proton_compat",
					str(proton),
					prefix_verb,
					"cmd",
					"/c",
					"exit",
				],
				env_overrides=environment,
				timeout_seconds=600.0,
			)
		)
		if not runner.dry_run:
			_ensure_proton_game_link(paths)
		results.append(
			runner.run(
				[
					sys.executable,
					"-m",
					"ptlinux.proton_compat",
					str(proton),
					"runinprefix",
					"reg.exe",
					"add",
					DLL_OVERRIDES_KEY,
					"/v",
					"d3d9",
					"/t",
					"REG_SZ",
					"/d",
					DXVK_D3D9_OVERRIDE,
					"/f",
				],
				env_overrides=environment,
				timeout_seconds=60.0,
			)
		)
		if not native_d3d_ready(paths.proton_prefix):
			wine_tools = _find_proton_wine_tools(proton)
			winetricks = shutil.which("winetricks")
			if wine_tools is None or not winetricks:
				raise RuntimeErrorPT("Proton Winetricks helpers are unavailable")
			wine, wineserver = wine_tools
			compatible_winetricks = prepare_proton7_winetricks(
				Path(winetricks),
				paths.cache / "compat/winetricks-proton7",
				dry_run=runner.dry_run,
			)
			results.append(
				runner.run(
					[
						str(compatible_winetricks),
						"-q",
						"d3dx9_43",
						"d3dcompiler_43",
					],
					env_overrides={
						"WINEPREFIX": str(paths.proton_prefix),
						"WINE": str(wine),
						"WINESERVER": str(wineserver),
						"WINEDEBUG": "-all",
					},
					timeout_seconds=1800.0,
				)
			)
	else:
		raise RuntimeErrorPT(f"unsupported runtime kind: {selection.kind}")
	if not runner.dry_run:
		_write_json(
			paths.game_runtime_marker,
			{
				"configuration_version": GAME_RUNTIME_CONFIGURATION_VERSION,
				"configured_at": datetime.now().astimezone().isoformat(),
				"native_d3d43": True,
				"dxvk_d3d9_override": DXVK_D3D9_OVERRIDE,
				"proton_identity": _proton_identity(selection),
				"runtime": asdict(selection),
			},
		)
	return results


def game_runtime_ready(paths: AppPaths, selection: RuntimeSelection) -> bool:
	"""Return whether another client may reuse the prepared runtime read-only."""

	if selection.kind == "umu":
		return native_d3d_ready(paths.umu_prefix)
	if selection.kind in {"proton7", "proton7-container"}:
		return _game_runtime_ready(paths, selection)
	return False


def run_game(
	paths: AppPaths,
	settings: Settings,
	selection: RuntimeSelection,
	runner: CommandRunner,
	*,
	use_gamebox: bool = True,
	timeout_seconds: float = 12 * 60 * 60,
	check: bool = True,
	extra_environment: Mapping[str, str] | None = None,
	output_line_filter: OutputLineFilter | None = None,
) -> CommandOutcome:
	if not paths.runtime_view.is_dir() and not runner.dry_run:
		raise RuntimeErrorPT("runtime view is missing; prepare the game first")
	if selection.kind == "umu":
		game_command = [
			selection.executable,
			str(paths.runtime_view / "Game.exe"),
			"PTEU_REBORN",
		]
		environment = _umu_environment(paths)
	elif selection.kind == "proton7-container":
		if not selection.proton_path:
			raise RuntimeErrorPT("Proton script path is missing")
		game_command = [
			selection.executable,
			str(paths.runtime_view / "Game.exe"),
			"PTEU_REBORN",
		]
		environment = _umu_proton7_environment(
			paths,
			selection,
			synchronization=resolve_proton_sync_mode(settings.proton_sync),
		)
	else:
		if not selection.proton_path:
			raise RuntimeErrorPT("Proton script path is missing")
		game_command = [
			sys.executable,
			"-m",
			"ptlinux.proton_compat",
			selection.proton_path,
			"runinprefix",
			WINDOWS_GAME_EXE,
			"PTEU_REBORN",
		]
		environment = _proton_environment(
			paths,
			selection,
			synchronization=resolve_proton_sync_mode(settings.proton_sync),
		)
	if use_gamebox:
		game_command = build_gamebox_command(game_command, settings)
	if extra_environment:
		environment.update(extra_environment)
	environment["PYTHONPATH"] = _runtime_python_path(
		os.environ.get("PYTHONPATH"),
	)
	return runner.run(
		game_command,
		cwd=paths.runtime_view,
		env_overrides=environment,
		timeout_seconds=timeout_seconds,
		check=check,
		stream_output=True,
		output_line_filter=output_line_filter,
	)


def run_proton_prefix_command(
	paths: AppPaths,
	selection: RuntimeSelection,
	runner: CommandRunner,
	arguments: Sequence[str],
	*,
	check: bool = True,
	timeout_seconds: float = 60.0,
) -> CommandOutcome:
	"""Run a maintenance command in the pinned Proton 7 prefix."""

	if selection.kind not in {"proton7", "proton7-container"}:
		raise RuntimeErrorPT("the login trace requires the pinned Proton 7 runtime")
	if not selection.proton_path or not selection.steam_root:
		raise RuntimeErrorPT("incomplete Proton 7 runtime selection")
	return runner.run(
		[
			sys.executable,
			"-m",
			"ptlinux.proton_compat",
			selection.proton_path,
			"runinprefix",
			*arguments,
		],
		env_overrides=_proton_environment(paths, selection),
		timeout_seconds=timeout_seconds,
		check=check,
	)


def run_runtime_prefix_command(
	paths: AppPaths,
	selection: RuntimeSelection,
	runner: CommandRunner,
	arguments: Sequence[str],
	*,
	check: bool = True,
	timeout_seconds: float = 60.0,
) -> CommandOutcome:
	"""Run a credential-free helper inside the selected game's Wine prefix."""

	if not arguments:
		raise ValueError("runtime-prefix command arguments cannot be empty")
	if selection.kind == "umu":
		command = [selection.executable, *arguments]
		environment = _umu_environment(paths)
	elif selection.kind == "proton7-container":
		if not selection.proton_path or not selection.steam_root:
			raise RuntimeErrorPT("incomplete Proton 7 runtime selection")
		command = [selection.executable, *arguments]
		environment = _umu_proton7_environment(paths, selection)
	elif selection.kind == "proton7":
		if not selection.proton_path or not selection.steam_root:
			raise RuntimeErrorPT("incomplete Proton 7 runtime selection")
		command = [
			sys.executable,
			"-m",
			"ptlinux.proton_compat",
			selection.proton_path,
			"runinprefix",
			*arguments,
		]
		environment = _proton_environment(paths, selection)
	else:
		raise RuntimeErrorPT(f"unsupported runtime kind: {selection.kind}")
	return runner.run(
		command,
		env_overrides=environment,
		timeout_seconds=timeout_seconds,
		check=check,
	)


def _runtime_python_path(existing: str | None) -> str:
	"""Make the installed package importable after changing into runtime_view."""
	package_root = str(Path(__file__).resolve().parent.parent)
	if not existing:
		return package_root
	return os.pathsep.join((package_root, existing))


def _umu_environment(paths: AppPaths) -> dict[str, str]:
	return {
		"WINEPREFIX": str(paths.umu_prefix),
		"GAMEID": "umu-pristontale-eu",
		"STORE": "none",
		"DXVK_STATE_CACHE_PATH": str(paths.dxvk_cache),
		"WINEDEBUG": "-all",
	}


def _umu_proton7_environment(
	paths: AppPaths,
	selection: RuntimeSelection,
	*,
	synchronization: str = RUNTIME_DEFAULT_SYNC,
) -> dict[str, str]:
	"""Run the retained Proton 7 build inside its Steam Runtime container."""

	if not selection.proton_path:
		raise RuntimeErrorPT("Proton script path is missing")
	environment = {
		# UMU expects the Proton compatibility-data root and manages its pfx child.
		# Pointing it at proton_prefix would create a nested pfx -> . compatibility
		# symlink; compat_data preserves the exact prefix prepared by Proton 7.
		"WINEPREFIX": str(paths.compat_data),
		"GAMEID": "umu-pristontale-eu",
		"STORE": "none",
		"PROTONPATH": str(Path(selection.proton_path).parent),
		"PROTON_VERB": "waitforexitandrun",
		"PROTONFIXES_DISABLE": "1",
		"DXVK_STATE_CACHE_PATH": str(paths.dxvk_cache),
		"WINEDEBUG": "-all",
	}
	environment.update(_proton_sync_environment(synchronization))
	return environment


def _proton_environment(
	paths: AppPaths,
	selection: RuntimeSelection,
	*,
	synchronization: str = RUNTIME_DEFAULT_SYNC,
) -> dict[str, str]:
	if not selection.steam_root:
		raise RuntimeErrorPT("Steam root is missing for Proton")
	environment = {
		"STEAM_COMPAT_CLIENT_INSTALL_PATH": selection.steam_root,
		"STEAM_COMPAT_DATA_PATH": str(paths.compat_data),
		"STEAM_COMPAT_INSTALL_PATH": str(paths.runtime_view),
		"SteamAppId": "0",
		"SteamGameId": "0",
		"WINEPREFIX": str(paths.proton_prefix),
		"WINEDEBUG": "-all",
		"PROTON_LOG": "0",
		"DXVK_STATE_CACHE_PATH": str(paths.dxvk_cache),
	}
	environment.update(_proton_sync_environment(synchronization))
	return environment


def detect_distribution_id() -> str:
	"""Return the freedesktop distribution ID without invoking distro tools."""

	try:
		value = platform.freedesktop_os_release().get("ID", "")
	except OSError:
		return ""
	return value.strip().casefold()


def is_cachyos_host(
	*,
	distribution_id: str | None = None,
	kernel_release: str | None = None,
) -> bool:
	"""Identify CachyOS narrowly for runtime compatibility policy."""

	distribution = detect_distribution_id() if distribution_id is None else distribution_id
	kernel = platform.release() if kernel_release is None else kernel_release
	return (
		CACHYOS_MARKER in distribution.casefold()
		or CACHYOS_MARKER in kernel.casefold()
	)


def prefers_current_umu_host(
	*,
	distribution_id: str | None = None,
	kernel_release: str | None = None,
) -> bool:
	"""Use current UMU automatically on rolling Arch-family hosts."""

	distribution = detect_distribution_id() if distribution_id is None else distribution_id
	return (
		distribution.strip().casefold() in CURRENT_UMU_DISTRIBUTIONS
		or is_cachyos_host(
			distribution_id=distribution,
			kernel_release=kernel_release,
		)
	)


def resolve_proton_sync_mode(
	configured: str,
	*,
	distribution_id: str | None = None,
	kernel_release: str | None = None,
) -> str:
	"""Resolve the configured mode, applying the narrow CachyOS compatibility fallback."""

	if configured not in {"auto", "fsync", "esync"}:
		raise ValueError(f"unsupported Proton synchronization mode: {configured}")
	if configured != "auto":
		return configured
	if is_cachyos_host(
		distribution_id=distribution_id,
		kernel_release=kernel_release,
	):
		return "esync"
	return RUNTIME_DEFAULT_SYNC


def _proton_sync_environment(synchronization: str) -> dict[str, str]:
	if synchronization == RUNTIME_DEFAULT_SYNC:
		return {}
	if synchronization == "esync":
		return {
			"PROTON_NO_FSYNC": "1",
			"WINEFSYNC": "0",
			"PROTON_NO_ESYNC": "0",
			"WINEESYNC": "1",
		}
	if synchronization == "fsync":
		return {
			"PROTON_NO_FSYNC": "0",
			"WINEFSYNC": "1",
			"PROTON_NO_ESYNC": "0",
			"WINEESYNC": "1",
		}
	raise ValueError(f"unsupported resolved Proton synchronization mode: {synchronization}")


def _ensure_proton_game_link(paths: AppPaths) -> None:
	link = paths.proton_prefix / "drive_c/Program Files (x86)/Pristontale EU"
	link.parent.mkdir(parents=True, exist_ok=True)
	if link.is_symlink():
		if link.resolve() == paths.runtime_view.resolve():
			return
		raise RuntimeErrorPT(f"Proton game link points somewhere unexpected: {link}")
	if link.exists():
		if link.is_dir() and not any(link.iterdir()):
			link.rmdir()
		else:
			raise RuntimeErrorPT(f"refusing to replace existing Proton game path: {link}")
	link.symlink_to(paths.runtime_view, target_is_directory=True)


def _steam_root_for_proton(proton: Path) -> Path:
	parts = proton.parts
	try:
		index = parts.index("steamapps")
	except ValueError as exc:
		raise RuntimeErrorPT(
			"configured Proton path must be inside a Steam steamapps directory"
		) from exc
	return Path(*parts[:index])


def native_d3d_ready(prefix: Path) -> bool:
	directory = prefix / "drive_c/windows/syswow64"
	if not all((directory / f"{name}.dll").is_file() for name in NATIVE_D3D43_DLLS):
		return False
	try:
		registry = (prefix / "user.reg").read_text(
			encoding="utf-8",
			errors="replace",
		).casefold()
	except OSError:
		return False
	return all(_native_override_ready(registry, name) for name in NATIVE_D3D43_DLLS)


def _native_override_ready(registry: str, name: str) -> bool:
	return any(
		entry in registry
		for entry in (
			f'"{name}"="native,builtin"',
			f'"{name}"="native"',
			f'"*{name}"="native"',
		)
	)


def prepare_proton7_winetricks(
	source: Path,
	destination: Path,
	*,
	dry_run: bool = False,
) -> Path:
	try:
		content = source.read_bytes()
	except OSError as exc:
		raise RuntimeErrorPT(f"could not read Winetricks script {source}: {exc}") from exc
	probe = WINETRICKS_CODEPAGE_PROBE.encode("utf-8")
	replacement = WINETRICKS_PROTON7_PROBE.encode("utf-8")
	if probe not in content:
		return source
	if dry_run:
		return destination
	destination.parent.mkdir(parents=True, exist_ok=True)
	patched = content.replace(
		probe,
		replacement,
		1,
	)
	temporary = destination.with_name(f".{destination.name}.{os.getpid()}.tmp")
	temporary.write_bytes(patched)
	temporary.chmod(source.stat().st_mode & 0o777)
	temporary.replace(destination)
	return destination


def _proton_prefix_ready(prefix: Path) -> bool:
	return (prefix / "user.reg").is_file() and (
		prefix / "drive_c/windows/system32/cmd.exe"
	).exists()


def _game_runtime_ready(paths: AppPaths, selection: RuntimeSelection) -> bool:
	marker = _read_json(paths.game_runtime_marker)
	identity = _proton_identity(selection)
	if marker is None or identity is None:
		return False
	if marker.get("configuration_version") != GAME_RUNTIME_CONFIGURATION_VERSION:
		return False
	if marker.get("runtime") != asdict(selection):
		return False
	if marker.get("proton_identity") != identity:
		return False
	if marker.get("native_d3d43") is not True:
		return False
	if marker.get("dxvk_d3d9_override") != DXVK_D3D9_OVERRIDE:
		return False
	if not _proton_prefix_ready(paths.proton_prefix):
		return False
	if not native_d3d_ready(paths.proton_prefix):
		return False
	if not _d3d9_override_ready(paths.proton_prefix):
		return False
	return _proton_game_link_ready(paths)


def _d3d9_override_ready(prefix: Path) -> bool:
	try:
		registry = (prefix / "user.reg").read_text(
			encoding="utf-8",
			errors="replace",
		).casefold()
	except OSError:
		return False
	return f'"d3d9"="{DXVK_D3D9_OVERRIDE}"' in registry


def _proton_game_link_ready(paths: AppPaths) -> bool:
	link = paths.proton_prefix / "drive_c/Program Files (x86)/Pristontale EU"
	if not link.is_symlink():
		return False
	try:
		return link.resolve(strict=True) == paths.runtime_view.resolve(strict=True)
	except (OSError, RuntimeError):
		return False


def _proton_identity(selection: RuntimeSelection) -> dict[str, str] | None:
	if not selection.proton_path:
		return None
	proton = Path(selection.proton_path)
	identity_paths = {
		"proton": proton,
		"version": proton.parent / "version",
		"toolmanifest": proton.parent / "toolmanifest.vdf",
	}
	identity: dict[str, str] = {}
	for name, path in identity_paths.items():
		try:
			identity[name] = _sha256(path)
		except OSError:
			return None
	return identity


def _find_proton_wine_tools(proton: Path) -> tuple[Path, Path] | None:
	for directory in (proton.parent / "files/bin", proton.parent / "dist/bin"):
		wine = directory / "wine"
		wineserver = directory / "wineserver"
		if wine.is_file() and wineserver.is_file():
			return wine, wineserver
	return None


def _write_json(path: Path, value: object) -> None:
	path.parent.mkdir(parents=True, exist_ok=True)
	temporary = path.with_name(f".{path.name}.{os.getpid()}.tmp")
	temporary.write_text(json.dumps(value, indent=2, sort_keys=True) + "\n", encoding="utf-8")
	temporary.replace(path)


def _read_json(path: Path) -> dict[str, object] | None:
	try:
		value = json.loads(path.read_text(encoding="utf-8"))
	except (OSError, UnicodeError, json.JSONDecodeError):
		return None
	return value if isinstance(value, dict) else None


def _sha256(path: Path) -> str:
	digest = hashlib.sha256()
	with path.open("rb") as handle:
		for chunk in iter(lambda: handle.read(1024 * 1024), b""):
			digest.update(chunk)
	return digest.hexdigest()