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 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


@dataclass(frozen=True)
class Settings:
	realm: str = "Babel"
	width: int = 1600
	height: int = 900
	output_width: int = 1600
	output_height: int = 900
	refresh_rate: int = 60
	fullscreen: bool = False
	borderless: bool = True
	suppress_unresponsive_warnings: bool = True
	scaler: str = "fit"
	filter: str = "linear"
	runtime: str = "auto"
	proton_sync: str = "auto"
	proton_path: str | None = None
	game_source: str | None = None

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


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(),
		width=_integer(value, "width", 1600),
		height=_integer(value, "height", 900),
		output_width=_integer(value, "output_width", 1600),
		output_height=_integer(value, "output_height", 900),
		refresh_rate=_integer(value, "refresh_rate", 60),
		fullscreen=_boolean(value, "fullscreen", False),
		borderless=_boolean(value, "borderless", True),
		suppress_unresponsive_warnings=_boolean(
			value,
			"suppress_unresponsive_warnings",
			True,
		),
		scaler=_string(value, "scaler", "fit"),
		filter=_string(value, "filter", "linear"),
		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"),
	)
	_validate(settings)
	return settings


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)
	)


def _write_settings(
	paths: AppPaths,
	settings: Settings,
	existing: dict[str, object] | None,
) -> None:
	payload = dict(existing or {})
	payload.update(asdict(settings))
	payload["schema"] = SETTINGS_SCHEMA
	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")
	for name, value in (
		("width", settings.width),
		("height", settings.height),
		("output_width", settings.output_width),
		("output_height", settings.output_height),
	):
		if value < 640 or value > 16384:
			raise ValueError(f"{name} must be between 640 and 16384")
	if settings.refresh_rate < 24 or settings.refresh_rate > 1000:
		raise ValueError("refresh_rate must be between 24 and 1000")
	if settings.scaler not in {"auto", "integer", "fit", "fill", "stretch"}:
		raise ValueError(f"unsupported legacy scaler setting: {settings.scaler}")
	if settings.filter not in {"linear", "nearest", "fsr", "nis", "pixel"}:
		raise ValueError(f"unsupported legacy filter setting: {settings.filter}")
	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}")


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


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


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