pristontale-eu-linux/src/ptlinux/settings.py

"""User-editable launcher settings with conservative defaults."""

from __future__ import annotations

from dataclasses import asdict, dataclass, fields, replace
import json
import math
import os
from pathlib import Path
import re
import secrets
import threading
from typing import Any

from .locking import launcher_operation_lock
from .paths import AppPaths
from .update_resolver import SUPPORTED_CHANNELS


SETTINGS_SCHEMA = 1
DEFAULT_GAME_RUNTIME = "ge-proton"
_TITLE_FRAME_DEFAULT_MIGRATION_KEY = "title_frame_default_migrated"
_GE_PROTON_DEFAULT_MIGRATION_KEY = "ge_proton_default_migrated"
UI_SCALE_CHOICES = frozenset({"auto", "100", "125", "150", "175", "200"})
_RETIRED_SETTINGS = frozenset(
	{
		"width",
		"height",
		"output_width",
		"output_height",
		"refresh_rate",
		"fullscreen",
		"scaler",
		"filter",
		"maximize_button",
	}
)


@dataclass(frozen=True)
class Settings:
	realm: str = "Babel"
	ui_scale: str = "auto"
	telemetry_enabled: bool = True
	game_crash_dump_upload_enabled: bool = False
	borderless: bool = False
	suppress_unresponsive_warnings: bool = True
	runtime: str = DEFAULT_GAME_RUNTIME
	proton_sync: str = "auto"
	proton_path: str | None = None
	renderer: str = "auto"
	game_source: str | None = None
	download_speed_limit_mbps: float | None = None
	low_bandwidth_mode: bool = False
	allow_window_resizing: bool = False
	update_channel: str = "stable"

	@property
	def source_override(self) -> Path | None:
		return Path(self.game_source).expanduser().resolve() if self.game_source else None

	@property
	def download_speed_limit_bytes_per_second(self) -> int | None:
		if self.download_speed_limit_mbps is None:
			return None
		return round(self.download_speed_limit_mbps * 1_000_000)


def load_settings(paths: AppPaths) -> Settings:
	value = _read_settings_object(paths.settings_file)
	if value is None:
		return Settings()
	return _settings_from_object(value)


def save_settings(paths: AppPaths, settings: Settings) -> Settings:
	"""Atomically save a complete settings snapshot.

	Unknown fields from a newer launcher are retained when the existing file is
	valid. Callers editing only part of the configuration should use
	``update_settings`` so an old in-memory snapshot cannot undo another save.
	"""

	_validate(settings)
	validate_proton_path_setting(settings.proton_path)
	paths.config.mkdir(parents=True, exist_ok=True)
	with launcher_operation_lock(paths.settings_lock, "settings-save"):
		existing = _read_settings_object(paths.settings_file)
		_write_settings(paths, settings, existing)
	return settings


def update_settings(paths: AppPaths, **changes: Any) -> Settings:
	"""Merge selected settings into the newest on-disk configuration."""

	unknown = set(changes).difference(_SETTINGS_FIELD_NAMES)
	if unknown:
		names = ", ".join(sorted(unknown))
		raise ValueError(f"unsupported launcher setting: {names}")
	paths.config.mkdir(parents=True, exist_ok=True)
	with launcher_operation_lock(paths.settings_lock, "settings-update"):
		existing = _read_settings_object(paths.settings_file)
		current = (
			_settings_from_object(existing)
			if existing is not None
			else Settings()
		)
		updated = replace(current, **changes)
		_validate(updated)
		validate_proton_path_setting(updated.proton_path)
		_write_settings(paths, updated, existing)
	return updated


def ensure_settings(paths: AppPaths) -> Settings:
	"""Create settings once and materialize defaults added by newer releases."""

	paths.config.mkdir(parents=True, exist_ok=True)
	with launcher_operation_lock(paths.settings_lock, "settings-migrate"):
		existing = _read_settings_object(paths.settings_file)
		if existing is None:
			settings = Settings()
			_write_settings(paths, settings, None)
			return settings
		settings = _settings_from_object(existing)
		if _settings_need_migration(existing):
			_write_settings(paths, settings, existing)
		return settings


def reset_settings(paths: AppPaths) -> Settings:
	"""Replace all launcher settings with the documented defaults."""

	settings = Settings()
	paths.config.mkdir(parents=True, exist_ok=True)
	with launcher_operation_lock(paths.settings_lock, "settings-reset"):
		_write_settings(paths, settings, None)
	return settings


def _read_settings_object(path: Path) -> dict[str, object] | None:
	try:
		content = path.read_bytes()
	except FileNotFoundError:
		return None
	except OSError as exc:
		raise ValueError(f"invalid settings file {path}: {exc}") from exc
	try:
		value = json.loads(_decode_settings(content))
	except (UnicodeError, json.JSONDecodeError) as exc:
		raise ValueError(f"invalid settings file {path}: {exc}") from exc
	if not isinstance(value, dict):
		raise ValueError(f"settings file must contain an object: {path}")
	schema = value.get("schema")
	if schema is not None:
		if isinstance(schema, bool) or not isinstance(schema, int):
			raise ValueError(f"settings schema must be an integer: {path}")
		if schema != SETTINGS_SCHEMA:
			raise ValueError(f"unsupported settings schema {schema}: {path}")
	return value


def _settings_from_object(value: dict[str, object]) -> Settings:
	proton_path = _optional_string(value, "proton_path")
	renderer = _string(value, "renderer", "auto")
	# Older settings exposed only a path field. Recover the exact renderer
	# assignment seen in telemetry without interpreting arbitrary shell input.
	legacy_renderer = _legacy_renderer_assignment(proton_path)
	if legacy_renderer is not None:
		proton_path = None
		if renderer == "auto":
			renderer = legacy_renderer
	settings = Settings(
		realm=_string(value, "realm", "Babel").strip(),
		ui_scale=_string(value, "ui_scale", "auto"),
		telemetry_enabled=_boolean(value, "telemetry_enabled", True),
		game_crash_dump_upload_enabled=_boolean(
			value,
			"game_crash_dump_upload_enabled",
			False,
		),
		borderless=_migrated_borderless(value),
		allow_window_resizing=_boolean(
			value,
			"allow_window_resizing",
			False,
		),
		suppress_unresponsive_warnings=_boolean(
			value,
			"suppress_unresponsive_warnings",
			True,
		),
		runtime=_string(value, "runtime", DEFAULT_GAME_RUNTIME),
		proton_sync=_string(value, "proton_sync", "auto"),
		proton_path=proton_path,
		renderer=renderer,
		game_source=_optional_string(value, "game_source"),
		download_speed_limit_mbps=_optional_positive_number(
			value,
			"download_speed_limit_mbps",
		),
		low_bandwidth_mode=_boolean(value, "low_bandwidth_mode", False),
		update_channel=_string(value, "update_channel", "stable"),
	)
	_validate(settings)
	# Move existing installations to the modern WoW64 profile once. Validate the
	# saved runtime first so migration cannot hide malformed settings. New saves
	# carry the marker, allowing users to choose another runtime afterwards.
	# Keep the optional Proton path for PTR and future explicit legacy choices.
	if not _boolean(value, _GE_PROTON_DEFAULT_MIGRATION_KEY, False):
		return replace(settings, runtime=DEFAULT_GAME_RUNTIME)
	return settings


def _migrated_borderless(value: dict[str, object]) -> bool:
	"""Retire the old implicit borderless default exactly once.

	Older releases wrote ``borderless=true`` for every default installation, so
	they cannot distinguish an inherited value from an explicit choice. The
	migration marker keeps schema 1 backward compatible while letting users opt
	back into Borderless after the decorated default has been applied.
	"""

	borderless = _boolean(value, "borderless", False)
	migrated = _boolean(value, _TITLE_FRAME_DEFAULT_MIGRATION_KEY, False)
	if not migrated:
		return False
	return borderless


def _decode_settings(content: bytes) -> str:
	"""Accept native UTF-8 plus settings saved by common Windows editors."""

	try:
		return content.decode("utf-8-sig")
	except UnicodeDecodeError as utf8_error:
		try:
			return content.decode("cp1252")
		except UnicodeDecodeError:
			raise utf8_error


def _settings_need_migration(value: dict[str, object]) -> bool:
	return (
		value.get("schema") != SETTINGS_SCHEMA
		or not _SETTINGS_FIELD_NAMES.issubset(value)
		or value.get(_TITLE_FRAME_DEFAULT_MIGRATION_KEY) is not True
		or value.get(_GE_PROTON_DEFAULT_MIGRATION_KEY) is not True
		or bool(_RETIRED_SETTINGS.intersection(value))
		or _legacy_renderer_assignment(value.get("proton_path")) is not None
	)


def _legacy_renderer_assignment(value: object) -> str | None:
	if not isinstance(value, str):
		return None
	return {
		"PROTON_USE_WINED3D=1": "wined3d",
		"PROTON_USE_WINED3D=0": "dxvk",
	}.get(value.strip())


def validate_proton_path_setting(value: str | None) -> None:
	"""Reject launch options in a filesystem field, without requiring a mounted disk."""

	if value is None:
		return
	if not value.strip() or any(ord(character) < 32 for character in value):
		raise ValueError("Proton script must be a filesystem path")
	if re.match(r"(?:env\s+)?[A-Za-z_][A-Za-z0-9_]*=", value.strip()):
		raise ValueError(
			"Proton script must be a filesystem path. Choose WineD3D or DXVK "
			"under Renderer; put other environment variables in the launcher "
			"process environment. Leave Proton script blank for automatic detection."
		)
	if "%command%" in value:
		raise ValueError("Proton script accepts a path, not Steam launch options")


def _write_settings(
	paths: AppPaths,
	settings: Settings,
	existing: dict[str, object] | None,
) -> None:
	payload = dict(existing or {})
	for name in _RETIRED_SETTINGS:
		payload.pop(name, None)
	payload.update(asdict(settings))
	payload["schema"] = SETTINGS_SCHEMA
	payload[_TITLE_FRAME_DEFAULT_MIGRATION_KEY] = True
	payload[_GE_PROTON_DEFAULT_MIGRATION_KEY] = True
	temporary = paths.settings_file.with_name(
		f".{paths.settings_file.name}.{os.getpid()}.{threading.get_ident()}."
		f"{secrets.token_hex(8)}.tmp"
	)
	try:
		with temporary.open("x", encoding="utf-8", newline="\n") as handle:
			json.dump(payload, handle, indent=2, sort_keys=True)
			handle.write("\n")
			handle.flush()
			os.fsync(handle.fileno())
		temporary.chmod(0o600)
		os.replace(temporary, paths.settings_file)
		_fsync_directory(paths.config)
	except BaseException:
		temporary.unlink(missing_ok=True)
		raise


def _fsync_directory(path: Path) -> None:
	try:
		descriptor = os.open(path, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))
	except OSError:
		return
	try:
		os.fsync(descriptor)
	except OSError:
		pass
	finally:
		os.close(descriptor)


def _validate(settings: Settings) -> None:
	if (
		len(settings.realm) > 64
		or not settings.realm.strip()
		or any(ord(character) < 32 for character in settings.realm)
	):
		raise ValueError("realm must be a non-empty display name")
	if settings.ui_scale not in UI_SCALE_CHOICES:
		raise ValueError(f"unsupported launcher UI scale: {settings.ui_scale}")
	if settings.runtime not in {"auto", "proton7", "proton7-container", "umu", "ge-proton"}:
		raise ValueError(f"unsupported runtime selection: {settings.runtime}")
	if settings.renderer not in {"auto", "dxvk", "wined3d"}:
		raise ValueError(f"unsupported renderer: {settings.renderer}")
	if settings.proton_sync not in {"auto", "fsync", "esync"}:
		raise ValueError(f"unsupported Proton synchronization mode: {settings.proton_sync}")
	if settings.update_channel not in SUPPORTED_CHANNELS:
		raise ValueError(f"unsupported launcher update channel: {settings.update_channel}")
	if (
		settings.download_speed_limit_mbps is not None
		and (
			not math.isfinite(settings.download_speed_limit_mbps)
			or settings.download_speed_limit_mbps <= 0
			or settings.download_speed_limit_mbps > 10_000
		)
	):
		raise ValueError(
			"download_speed_limit_mbps must be greater than 0 and at most 10000, or null"
		)

def _boolean(value: dict[str, object], key: str, default: bool) -> bool:
	item = value.get(key, default)
	if not isinstance(item, bool):
		raise ValueError(f"{key} must be a boolean")
	return item


def _string(value: dict[str, object], key: str, default: str) -> str:
	item = value.get(key, default)
	if not isinstance(item, str) or not item:
		raise ValueError(f"{key} must be a non-empty string")
	return item


def _optional_string(value: dict[str, object], key: str) -> str | None:
	item = value.get(key)
	if item is None:
		return None
	if not isinstance(item, str) or not item:
		raise ValueError(f"{key} must be null or a non-empty string")
	return item


def _optional_positive_number(value: dict[str, object], key: str) -> float | None:
	item = value.get(key)
	if item is None:
		return None
	if isinstance(item, bool) or not isinstance(item, (int, float)):
		raise ValueError(f"{key} must be null or a positive number")
	number = float(item)
	if not math.isfinite(number) or number <= 0:
		raise ValueError(f"{key} must be null or a positive number")
	return number


_SETTINGS_FIELD_NAMES = frozenset(field.name for field in fields(Settings))