pristontale-eu-linux/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 secrets
import threading
from typing import Any

from .locking import launcher_operation_lock
from .paths import AppPaths


SETTINGS_SCHEMA = 1
_TITLE_FRAME_DEFAULT_MIGRATION_KEY = "title_frame_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"
	borderless: bool = False
	suppress_unresponsive_warnings: bool = True
	runtime: str = "auto"
	proton_sync: str = "auto"
	proton_path: str | None = None
	game_source: str | None = None
	download_speed_limit_mbps: float | None = None
	low_bandwidth_mode: bool = False

	@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)
	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)
		_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:
	settings = Settings(
		realm=_string(value, "realm", "Babel").strip(),
		ui_scale=_string(value, "ui_scale", "auto"),
		borderless=_migrated_borderless(value),
		suppress_unresponsive_warnings=_boolean(
			value,
			"suppress_unresponsive_warnings",
			True,
		),
		runtime=_string(value, "runtime", "auto"),
		proton_sync=_string(value, "proton_sync", "auto"),
		proton_path=_optional_string(value, "proton_path"),
		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),
	)
	_validate(settings)
	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 bool(_RETIRED_SETTINGS.intersection(value))
	)


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
	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"}:
		raise ValueError(f"unsupported runtime selection: {settings.runtime}")
	if settings.proton_sync not in {"auto", "fsync", "esync"}:
		raise ValueError(f"unsupported Proton synchronization mode: {settings.proton_sync}")
	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))