pristontale-eu-linux/ptlinux/depot.py

"""Native, verified updater for the public PristonTale EU depot."""

from __future__ import annotations

from base64 import b64decode
from binascii import Error as BinAsciiError
from concurrent.futures import FIRST_COMPLETED, Future, ThreadPoolExecutor, wait
from dataclasses import asdict, dataclass
from datetime import datetime
import hashlib
from http.client import HTTPException
import json
import os
from pathlib import Path, PurePosixPath
import re
import stat
import threading
import time
from typing import Callable, Iterable, Mapping, Protocol
from urllib.error import HTTPError
from urllib.parse import quote, urlparse
from urllib.request import Request, urlopen
from uuid import uuid4

from . import __version__
from .paths import AppPaths


CDN_CONFIGURATION_URL = "https://cdn-main.chatplex.io/pristontale-eu/cdn_configuration.json"
DEPOT_ID = 1
FALLBACK_SERVERS = (
	"cdn-fr-sbg1.chatplex.io/pristontale-eu",
	"cdn-us-ore1.chatplex.io/pristontale-eu",
)
READ_SIZE = 128 * 1024
HASH_READ_SIZE = 1024 * 1024
DOWNLOAD_WORKERS = 8
DOWNLOAD_RETRY_DELAYS_SECONDS = (1.0, 2.0, 4.0, 8.0, 16.0)
PROGRESS_UPDATE_INTERVAL_SECONDS = 0.075
MAX_CREATE_IF_MISSING_BYTES = 16 * 1024 * 1024
MAX_LINE_ENDING_RECOVERY_BYTES = 1024 * 1024
UNMANAGED_EXECUTABLE_SUFFIXES = frozenset(
	{
		".asi",
		".bat",
		".cmd",
		".com",
		".cpl",
		".desktop",
		".dll",
		".drv",
		".exe",
		".jar",
		".js",
		".lnk",
		".lua",
		".msi",
		".ocx",
		".ps1",
		".py",
		".reg",
		".scr",
		".sh",
		".so",
		".url",
		".vbe",
		".vbs",
		".wsf",
	}
)
ALWAYS_VERIFY_PATHS = frozenset(
	{"game.exe", "game.dll", "gamese.exe", "gamese.dll"}
)
SHA1_PATTERN = re.compile(r"^[0-9a-fA-F]{40}$")
CONTENT_RANGE_PATTERN = re.compile(r"^bytes\s+(\d+)-(\d+)/(\d+)$", re.IGNORECASE)
DEPOT_LAUNCH_CATALOG_SCHEMA = 1


class DepotError(RuntimeError):
	"""Raised when a depot response or update operation cannot be trusted."""


class _RetryableDownloadError(DepotError):
	"""One transient network or content-integrity failure."""

	def __init__(self, message: str, *, discard_partial: bool = False) -> None:
		super().__init__(message)
		self.discard_partial = discard_partial


@dataclass(frozen=True)
class DepotFile:
	"""One exact file in the published depot build."""

	path: str
	build: int
	size: int
	sha1: str


@dataclass(frozen=True)
class CreateIfMissing:
	"""A small publisher-provided file created only when absent."""

	path: str
	content_base64: str


@dataclass(frozen=True)
class DepotStartOption:
	"""One publisher-advertised realm launch target."""

	name: str
	path: str
	arguments: str


DEFAULT_DEPOT_START_OPTION = DepotStartOption(
	name="Babel",
	path="Game.exe",
	arguments="PTEU_REBORN",
)


@dataclass(frozen=True)
class DepotSnapshot:
	"""The current, verified shape of the public game depot."""

	region: str
	server: str
	name: str
	build: int
	version: str
	files: tuple[DepotFile, ...]
	deleted_files: tuple[str, ...]
	create_if_missing: tuple[CreateIfMissing, ...]
	start_options: tuple[DepotStartOption, ...] = (DEFAULT_DEPOT_START_OPTION,)
	default_start_option: str = DEFAULT_DEPOT_START_OPTION.name


@dataclass(frozen=True)
class DepotLaunchCatalog:
	"""Small cached subset of a depot manifest needed for offline Play."""

	build: int
	start_options: tuple[DepotStartOption, ...]
	default_start_option: str


def launch_catalog_from_snapshot(snapshot: DepotSnapshot) -> DepotLaunchCatalog:
	return DepotLaunchCatalog(
		build=snapshot.build,
		start_options=snapshot.start_options,
		default_start_option=snapshot.default_start_option,
	)


def select_depot_start_option(
	catalog: DepotLaunchCatalog | None,
	preferred_realm: str | None = None,
) -> DepotStartOption:
	"""Choose a published realm, falling back to Babel when possible."""

	current = catalog or DepotLaunchCatalog(
		build=0,
		start_options=(DEFAULT_DEPOT_START_OPTION,),
		default_start_option=DEFAULT_DEPOT_START_OPTION.name,
	)
	if preferred_realm:
		preferred = preferred_realm.casefold()
		for option in current.start_options:
			if option.name.casefold() == preferred:
				return option
	for option in current.start_options:
		if option.name.casefold() == DEFAULT_DEPOT_START_OPTION.name.casefold():
			return option
	for option in current.start_options:
		if option.name.casefold() == current.default_start_option.casefold():
			return option
	return current.start_options[0]


def cache_depot_launch_catalog(path: Path, snapshot: DepotSnapshot) -> None:
	"""Cache only publisher-advertised launch metadata for network-free Play."""

	catalog = launch_catalog_from_snapshot(snapshot)
	_write_json(
		path,
		{
			"schema": DEPOT_LAUNCH_CATALOG_SCHEMA,
			"build": catalog.build,
			"default_start_option": catalog.default_start_option,
			"start_options": [asdict(option) for option in catalog.start_options],
		},
	)


def read_cached_depot_launch_catalog(path: Path) -> DepotLaunchCatalog | None:
	"""Read the last valid launch catalog without performing network I/O."""

	try:
		value = json.loads(path.read_text(encoding="utf-8"))
		if not isinstance(value, dict):
			return None
		if value.get("schema") != DEPOT_LAUNCH_CATALOG_SCHEMA:
			return None
		build = _positive_int(value.get("build"), "cached depot build")
		raw_options = value.get("start_options")
		if not isinstance(raw_options, list):
			return None
		manifest_options = [
			{
				"Name": item.get("name"),
				"Path": item.get("path"),
				"Arguments": item.get("arguments"),
			}
			if isinstance(item, dict)
			else item
			for item in raw_options
		]
		options, default_name = _parse_start_options(
			manifest_options,
			value.get("default_start_option"),
		)
	except (OSError, UnicodeError, json.JSONDecodeError, DepotError):
		return None
	return DepotLaunchCatalog(
		build=build,
		start_options=options,
		default_start_option=default_name,
	)


@dataclass(frozen=True)
class PlannedDownload:
	file: DepotFile
	reason: str


@dataclass(frozen=True)
class UpdatePlan:
	"""The safe changes needed to make a local tree match one depot snapshot."""

	downloads: tuple[PlannedDownload, ...]
	deletes: tuple[str, ...]
	files_total: int
	bytes_total: int


@dataclass(frozen=True)
class SyncProgress:
	"""A thread-safe progress snapshot emitted by ``sync_managed_game``."""

	phase: str
	files_completed: int
	files_total: int
	bytes_completed: int
	bytes_total: int
	current_file: str | None
	bytes_transferred: int = 0


@dataclass(frozen=True)
class DepotSyncResult:
	"""Persistent result of one native depot synchronization."""

	region: str
	server: str
	name: str
	build: int
	version: str
	files_total: int
	files_downloaded: int
	bytes_downloaded: int
	files_deleted: int
	files_created: int
	files_quarantined: int
	quarantine_directory: str | None
	repair: bool
	dry_run: bool
	completed_at: str


@dataclass(frozen=True)
class QuarantineResult:
	"""Recoverable unmanaged entries removed before authenticating an import."""

	files: int
	directory: str | None


ProgressCallback = Callable[[SyncProgress], None]


class HttpResponse(Protocol):
	"""The small response surface used by the native depot client."""

	headers: Mapping[str, str]

	def read(self, size: int = -1) -> bytes: ...

	def close(self) -> None: ...


HttpOpener = Callable[..., HttpResponse]


class DepotClient:
	"""Small HTTP client for the publisher's documented public depot endpoints."""

	def __init__(
		self,
		*,
		timeout_seconds: float = 60.0,
		opener: HttpOpener = urlopen,
		retry_delays: tuple[float, ...] = DOWNLOAD_RETRY_DELAYS_SECONDS,
		sleeper: Callable[[float], None] = time.sleep,
	) -> None:
		if timeout_seconds <= 0:
			raise ValueError("timeout_seconds must be positive")
		if any(delay < 0 for delay in retry_delays):
			raise ValueError("retry delays must not be negative")
		self.timeout_seconds = timeout_seconds
		self._opener = opener
		self._retry_delays = retry_delays
		self._sleeper = sleeper
		self._candidate_servers: tuple[str, ...] = ()

	def fetch_snapshot(self, *, preferred_region: str = "Europe") -> DepotSnapshot:
		"""Fetch the live depot and build manifests, trying trusted regional fallbacks."""

		configuration: Mapping[str, object] | None = None
		configuration_error: DepotError | None = None
		try:
			configuration = self._get_json(CDN_CONFIGURATION_URL)
		except DepotError as exc:
			configuration_error = exc

		candidates = _server_candidates(configuration, preferred_region)
		self._candidate_servers = tuple(server for _region, server in candidates)
		last_error = configuration_error
		for region, server in candidates:
			try:
				depot_manifest = self._get_json(_depot_manifest_url(server))
				build = _positive_int(depot_manifest.get("Build"), "depot Build")
				build_manifest = self._get_json(_build_manifest_url(server, build))
				return _parse_snapshot(region, server, depot_manifest, build_manifest)
			except DepotError as exc:
				last_error = exc

		if last_error is not None:
			raise DepotError(f"could not load the PristonTale EU depot: {last_error}")
		raise DepotError("could not load the PristonTale EU depot")

	def download_file(
		self,
		server: str,
		file: DepotFile,
		destination: Path,
		*,
		progress: Callable[[int], None] | None = None,
	) -> None:
		"""Download, resume, and verify one file before replacing its destination.

		The current publisher CDN normalizes a small number of legacy text files
		from CRLF to LF while its manifest still hashes the CRLF bytes.  A
		candidate is accepted only when restoring bare LF line endings produces
		the manifest's exact size and SHA-1.
		"""

		servers = _ordered_download_servers(server, self._candidate_servers)
		temporary = _download_partial_path(destination)
		last_error: _RetryableDownloadError | None = None
		attempts = len(self._retry_delays) + 1
		for attempt_index in range(attempts):
			attempt_server = servers[attempt_index % len(servers)]
			try:
				self._download_file_once(
					attempt_server,
					file,
					destination,
					progress=progress,
				)
				return
			except _RetryableDownloadError as exc:
				last_error = exc
				if exc.discard_partial:
					temporary.unlink(missing_ok=True)
				if attempt_index >= len(self._retry_delays):
					break
				self._sleeper(self._retry_delays[attempt_index])

		assert last_error is not None
		resume_hint = (
			"; partial data was retained for the next update run"
			if temporary.is_file() and not temporary.is_symlink()
			else ""
		)
		raise DepotError(
			f"download failed for {file.path} after {attempts} attempts: "
			f"{last_error}{resume_hint}"
		) from last_error

	def _download_file_once(
		self,
		server: str,
		file: DepotFile,
		destination: Path,
		*,
		progress: Callable[[int], None] | None,
	) -> None:
		temporary = _download_partial_path(destination)
		written = 0
		digest = hashlib.sha1(usedforsecurity=False)
		if temporary.is_symlink():
			raise DepotError(f"refusing to use symbolic-link partial file for {file.path}")
		if temporary.exists():
			if not temporary.is_file():
				raise DepotError(f"partial download is not a file for {file.path}")
			try:
				with temporary.open("rb") as partial_handle:
					while True:
						chunk = partial_handle.read(HASH_READ_SIZE)
						if not chunk:
							break
						written += len(chunk)
						digest.update(chunk)
			except OSError as exc:
				raise DepotError(f"cannot inspect partial download for {file.path}: {exc}") from exc

			if written > file.size:
				temporary.unlink()
				written = 0
				digest = hashlib.sha1(usedforsecurity=False)
			elif written == file.size:
				if digest.hexdigest().casefold() == file.sha1.casefold():
					temporary.replace(destination)
					if progress is not None:
						progress(written)
					return
				temporary.unlink()
				written = 0
				digest = hashlib.sha1(usedforsecurity=False)
			elif written and _recover_crlf_manifest_file(temporary, destination, file):
				temporary.unlink(missing_ok=True)
				if progress is not None:
					progress(file.size)
				return
		if progress is not None:
			progress(written)

		requested_offset = written
		headers = _headers()
		if requested_offset:
			headers["Range"] = f"bytes={requested_offset}-"
		url = _file_url(server, file)
		request = Request(url, headers=headers, method="GET")
		try:
			response = self._opener(request, timeout=self.timeout_seconds)
		except HTTPError as exc:
			if exc.code == 416 and requested_offset:
				raise _RetryableDownloadError(
					f"server rejected the saved offset for {file.path}",
					discard_partial=True,
				) from exc
			raise _RetryableDownloadError(
				f"HTTP {exc.code} from {server} for {file.path}"
			) from exc
		except (OSError, HTTPException) as exc:
			raise _RetryableDownloadError(
				f"could not connect to {server} for {file.path}: {exc}"
			) from exc

		try:
			append = False
			if requested_offset:
				content_range = response.headers.get("Content-Range")
				status = getattr(response, "status", None)
				if status == 206 or content_range is not None:
					try:
						range_start, _range_end, range_total = _content_range(content_range)
					except DepotError as exc:
						raise _RetryableDownloadError(
							f"invalid resume response from {server} for {file.path}: {exc}"
						) from exc
					if range_start != requested_offset or range_total != file.size:
						raise _RetryableDownloadError(
							f"resume response did not match {file.path}: "
							f"offset {range_start}/{requested_offset}, size {range_total}/{file.size}"
						)
					append = True
				else:
					# A server may ignore Range and send a normal 200 response.  In
					# that case safely restart this file using the full response body.
					written = 0
					digest = hashlib.sha1(usedforsecurity=False)
					if progress is not None:
						progress(written)

			try:
				content_length = _content_length(response.headers.get("Content-Length"))
			except DepotError as exc:
				raise _RetryableDownloadError(
					f"invalid response length from {server} for {file.path}: {exc}"
				) from exc
			remaining = file.size - written
			if content_length is not None and content_length > remaining:
				raise _RetryableDownloadError(
					f"download exceeds expected size for {file.path}: "
					f"server declared {content_length}, expected at most {remaining}"
				)

			mode = "ab" if append else "wb"
			try:
				with temporary.open(mode) as handle:
					while True:
						try:
							chunk = response.read(READ_SIZE)
						except (OSError, HTTPException) as exc:
							raise _RetryableDownloadError(
								f"could not read {file.path} from {server}: {exc}"
							) from exc
						if not chunk:
							break
						written += len(chunk)
						if written > file.size:
							raise _RetryableDownloadError(
								f"download exceeds expected size for {file.path}",
								discard_partial=True,
							)
						digest.update(chunk)
						handle.write(chunk)
						if progress is not None:
							progress(written)
			except _RetryableDownloadError:
				raise
			except OSError as exc:
				raise DepotError(f"cannot write partial download for {file.path}: {exc}") from exc

			if written != file.size or digest.hexdigest().casefold() != file.sha1.casefold():
				if _recover_crlf_manifest_file(temporary, destination, file):
					temporary.unlink(missing_ok=True)
					if progress is not None:
						progress(file.size)
					return
				if written != file.size:
					raise _RetryableDownloadError(
						f"incomplete response from {server} for {file.path}: "
						f"{written}/{file.size} bytes"
					)
				raise _RetryableDownloadError(
					f"SHA-1 mismatch for {file.path} from {server}",
					discard_partial=True,
				)
			temporary.replace(destination)
		finally:
			try:
				response.close()
			except (OSError, HTTPException):
				pass

	def _get_json(self, url: str) -> Mapping[str, object]:
		request = Request(url, headers=_headers(), method="GET")
		try:
			response = self._opener(request, timeout=self.timeout_seconds)
		except (OSError, HTTPException) as exc:
			raise DepotError(f"request failed for {url}: {exc}") from exc
		payload = b""
		try:
			payload = response.read()
		except (OSError, HTTPException) as exc:
			raise DepotError(f"could not read {url}: {exc}") from exc
		finally:
			try:
				response.close()
			except (OSError, HTTPException):
				pass
		try:
			value = json.loads(payload.decode("utf-8-sig"))
		except (UnicodeDecodeError, json.JSONDecodeError) as exc:
			raise DepotError(f"invalid JSON from {url}") from exc
		if not isinstance(value, dict):
			raise DepotError(f"unexpected JSON root from {url}")
		return value


def plan_update(
	game_directory: Path,
	snapshot: DepotSnapshot,
	*,
	repair: bool = False,
) -> UpdatePlan:
	"""Inspect a local game tree without writing to it."""

	root = _prepare_game_root(game_directory, create=False)
	downloads: list[PlannedDownload] = []
	for file in snapshot.files:
		target = _safe_target(root, file.path)
		if target.is_symlink():
			downloads.append(PlannedDownload(file, "symbolic link"))
			continue
		if not target.is_file():
			downloads.append(PlannedDownload(file, "missing"))
			continue
		try:
			if target.stat().st_size != file.size:
				downloads.append(PlannedDownload(file, "size mismatch"))
				continue
		except OSError as exc:
			raise DepotError(f"cannot inspect {file.path}: {exc}") from exc
		if (
			repair
			or file.build == snapshot.build
			or file.path.casefold() in ALWAYS_VERIFY_PATHS
		):
			if _sha1(target).casefold() != file.sha1.casefold():
				downloads.append(PlannedDownload(file, "hash mismatch"))

	deletes: list[str] = []
	for relative_path in snapshot.deleted_files:
		target = _safe_target(root, relative_path)
		if target.is_symlink() or target.is_file():
			deletes.append(relative_path)
		elif target.exists():
			raise DepotError(
				f"refusing to delete non-file depot target: {relative_path}"
			)

	return UpdatePlan(
		downloads=tuple(downloads),
		deletes=tuple(deletes),
		files_total=len(snapshot.files),
		bytes_total=sum(item.file.size for item in downloads),
	)


def sync_managed_game(
	paths: AppPaths,
	*,
	repair: bool = False,
	quarantine_unmanaged: bool = False,
	dry_run: bool = False,
	progress: ProgressCallback | None = None,
	client: DepotClient | None = None,
) -> DepotSyncResult:
	"""Synchronize the managed game directory using only official public data."""

	paths.create_base_directories()
	depot_client = client or DepotClient()
	_emit_progress(progress, "manifest", 0, 0, 0, 0, None)
	snapshot = depot_client.fetch_snapshot()
	quarantine = QuarantineResult(0, None)
	if quarantine_unmanaged and not dry_run:
		quarantine = _quarantine_unmanaged_files(paths, snapshot)
	plan = plan_update(paths.source_game, snapshot, repair=repair)
	_emit_progress(
		progress,
		"planned",
		0,
		len(plan.downloads),
		0,
		plan.bytes_total,
		None,
	)

	if dry_run:
		return _sync_result(
			snapshot,
			plan,
			0,
			0,
			0,
			repair,
			True,
			quarantine=quarantine,
		)

	root = _prepare_game_root(paths.source_game, create=True)
	_download_plan(depot_client, snapshot, root, plan, progress)
	deleted = _apply_deletions(root, plan.deletes)
	created = _apply_create_if_missing(root, snapshot.create_if_missing)
	result = _sync_result(
		snapshot,
		plan,
		len(plan.downloads),
		plan.bytes_total,
		deleted,
		repair,
		False,
		created=created,
		quarantine=quarantine,
	)
	_write_json(paths.native_depot_state, asdict(result))
	cache_depot_launch_catalog(paths.depot_launch_catalog_cache, snapshot)
	_emit_progress(
		progress,
		"complete",
		len(plan.downloads),
		len(plan.downloads),
		plan.bytes_total,
		plan.bytes_total,
		None,
	)
	return result


def _quarantine_unmanaged_files(
	paths: AppPaths,
	snapshot: DepotSnapshot,
) -> QuarantineResult:
	"""Move non-depot import contents aside without following links."""

	root = _prepare_game_root(paths.source_game, create=False)
	if not root.exists():
		return QuarantineResult(0, None)
	allowed = {
		file.path for file in snapshot.files
	} | {
		entry.path for entry in snapshot.create_if_missing
	}
	candidates: list[tuple[Path, str]] = []

	def walk_error(error: OSError) -> None:
		raise DepotError(f"cannot inspect imported game tree: {error}") from error

	for directory, directory_names, file_names in os.walk(
		root,
		topdown=True,
		followlinks=False,
		onerror=walk_error,
	):
		parent = Path(directory)
		for name in tuple(directory_names):
			path = parent / name
			try:
				status = path.lstat()
			except OSError as exc:
				raise DepotError(f"cannot inspect imported entry {path}: {exc}") from exc
			if stat.S_ISDIR(status.st_mode):
				continue
			directory_names.remove(name)
			candidates.append((path, path.relative_to(root).as_posix()))
		for name in file_names:
			path = parent / name
			relative = path.relative_to(root).as_posix()
			try:
				status = path.lstat()
			except OSError as exc:
				raise DepotError(f"cannot inspect imported entry {path}: {exc}") from exc
			if not stat.S_ISREG(status.st_mode):
				candidates.append((path, relative))
				continue
			if relative in allowed:
				continue
			if _safe_imported_user_file(path, relative):
				continue
			candidates.append((path, relative))

	if not candidates:
		return QuarantineResult(0, None)
	base = paths.game_import_quarantine.resolve(strict=False)
	data_root = paths.data.resolve(strict=False)
	try:
		base.relative_to(data_root)
	except ValueError as exc:
		raise DepotError(f"import quarantine is outside the data root: {base}") from exc
	stamp = datetime.now().astimezone().strftime("%Y%m%d-%H%M%S")
	quarantine = base / f"{stamp}-{uuid4().hex}"
	files_root = quarantine / "files"
	files_root.mkdir(parents=True, exist_ok=False)
	manifest = {
		"schema": 1,
		"complete": False,
		"created_at": datetime.now().astimezone().isoformat(),
		"source": str(root),
		"files": [relative for _path, relative in candidates],
	}
	_write_json(quarantine / "quarantine.json", manifest)
	for source, relative in candidates:
		destination = files_root.joinpath(*PurePosixPath(relative).parts)
		destination.parent.mkdir(parents=True, exist_ok=True)
		try:
			source.replace(destination)
		except OSError as exc:
			raise DepotError(
				f"could not quarantine imported entry {relative}: {exc}"
			) from exc
	manifest["complete"] = True
	_write_json(quarantine / "quarantine.json", manifest)
	return QuarantineResult(len(candidates), str(quarantine))


def _safe_imported_user_file(path: Path, relative: str) -> bool:
	parts = PurePosixPath(relative).parts
	if not parts or parts[0].casefold() != "save":
		return False
	if path.suffix.casefold() in UNMANAGED_EXECUTABLE_SUFFIXES:
		return False
	try:
		with path.open("rb") as handle:
			prefix = handle.read(4)
	except OSError as exc:
		raise DepotError(f"cannot inspect imported user file {path}: {exc}") from exc
	return not (
		prefix.startswith(b"MZ")
		or prefix.startswith(b"\x7fELF")
		or prefix.startswith(b"#!")
	)


def _download_plan(
	client: DepotClient,
	depot_snapshot: DepotSnapshot,
	root: Path,
	plan: UpdatePlan,
	progress: ProgressCallback | None,
) -> None:
	if not plan.downloads:
		return

	lock = threading.Lock()
	emit_lock = threading.Lock()
	current_bytes: dict[str, int] = {}
	completed_files = 0
	completed_bytes = 0
	transferred_bytes = 0
	last_progress_emitted_at = 0.0

	def emit_download_progress(
		current_file: str,
		*,
		force: bool = False,
	) -> None:
		"""Throttle progress I/O without holding the download-state lock.

		The workers can produce many chunk and file-completion updates per second.
		The callback writes to a subprocess pipe for the native dashboard, so
		retaining every event would make that pipe and its UI consumer the bottleneck.
		"""

		nonlocal last_progress_emitted_at
		if progress is None:
			return
		with emit_lock:
			now = time.monotonic()
			if not force and now - last_progress_emitted_at < PROGRESS_UPDATE_INTERVAL_SECONDS:
				return
			last_progress_emitted_at = now
			with lock:
				event = SyncProgress(
					"downloading",
					completed_files,
					len(plan.downloads),
					completed_bytes + sum(current_bytes.values()),
					plan.bytes_total,
					current_file,
					transferred_bytes,
				)
			progress(event)

	def download(planned_download: PlannedDownload) -> None:
		target = _safe_target(root, planned_download.file.path)
		_ensure_target_parent(root, target)

		def report_file_bytes(written: int) -> None:
			nonlocal transferred_bytes
			with lock:
				previous = current_bytes.get(planned_download.file.path)
				if previous is not None and written > previous:
					transferred_bytes += written - previous
				current_bytes[planned_download.file.path] = written
			emit_download_progress(planned_download.file.path)

		client.download_file(
			depot_snapshot.server,
			planned_download.file,
			target,
			progress=report_file_bytes if progress is not None else None,
		)

	worker_count = min(DOWNLOAD_WORKERS, len(plan.downloads))
	remaining_items = iter(plan.downloads)
	with ThreadPoolExecutor(max_workers=worker_count) as executor:
		future_items: dict[Future[None], PlannedDownload] = {}

		def submit_next() -> bool:
			try:
				next_item = next(remaining_items)
			except StopIteration:
				return False
			future_items[executor.submit(download, next_item)] = next_item
			return True

		try:
			for _worker in range(worker_count):
				if not submit_next():
					break
			while future_items:
				done, _pending = wait(
					tuple(future_items),
					return_when=FIRST_COMPLETED,
				)
				completed_items: list[PlannedDownload] = []
				for future in done:
					completed_item = future_items.pop(future)
					future.result()
					completed_items.append(completed_item)
				for completed_item in completed_items:
					with lock:
						completed_files += 1
						completed_bytes += completed_item.file.size
						current_bytes.pop(completed_item.file.path, None)
						is_final_download = completed_files == len(plan.downloads)
					emit_download_progress(
						completed_item.file.path,
						force=is_final_download,
					)
				for _item in completed_items:
					submit_next()
		except BaseException:
			for future in future_items:
				future.cancel()
			raise


def _apply_deletions(root: Path, relative_paths: Iterable[str]) -> int:
	deleted = 0
	for relative_path in relative_paths:
		target = _safe_target(root, relative_path)
		if target.is_symlink() or target.is_file():
			target.unlink()
			deleted += 1
	return deleted


def _apply_create_if_missing(
	root: Path,
	entries: Iterable[CreateIfMissing],
) -> int:
	created = 0
	for entry in entries:
		target = _safe_target(root, entry.path)
		if target.is_symlink():
			raise DepotError(f"refusing to follow symbolic link for {entry.path}")
		if target.exists():
			continue
		content = _decode_create_if_missing(entry)
		_ensure_target_parent(root, target)
		temporary = target.with_name(f".{target.name}.{uuid4().hex}.partial")
		try:
			with temporary.open("xb") as handle:
				handle.write(content)
				handle.flush()
				os.fsync(handle.fileno())
			temporary.replace(target)
		except BaseException:
			temporary.unlink(missing_ok=True)
			raise
		created += 1
	return created


def _sync_result(
	snapshot: DepotSnapshot,
	plan: UpdatePlan,
	files_downloaded: int,
	bytes_downloaded: int,
	files_deleted: int,
	repair: bool,
	dry_run: bool,
	*,
	created: int = 0,
	quarantine: QuarantineResult = QuarantineResult(0, None),
) -> DepotSyncResult:
	return DepotSyncResult(
		region=snapshot.region,
		server=snapshot.server,
		name=snapshot.name,
		build=snapshot.build,
		version=snapshot.version,
		files_total=plan.files_total,
		files_downloaded=files_downloaded,
		bytes_downloaded=bytes_downloaded,
		files_deleted=files_deleted,
		files_created=created,
		files_quarantined=quarantine.files,
		quarantine_directory=quarantine.directory,
		repair=repair,
		dry_run=dry_run,
		completed_at=datetime.now().astimezone().isoformat(),
	)


def _parse_snapshot(
	region: str,
	server: str,
	depot_manifest: Mapping[str, object],
	build_manifest: Mapping[str, object],
) -> DepotSnapshot:
	build = _positive_int(depot_manifest.get("Build"), "depot Build")
	build_manifest_build = _positive_int(build_manifest.get("b"), "build manifest b")
	if build_manifest_build != build:
		raise DepotError(
			f"depot build {build} does not match build manifest {build_manifest_build}"
		)
	name = _string(depot_manifest.get("Name"), "depot Name")
	version = _string(depot_manifest.get("BuildVersionText"), "depot BuildVersionText")
	files_value = build_manifest.get("f")
	if not isinstance(files_value, list):
		raise DepotError("build manifest f must be an array")
	files: list[DepotFile] = []
	seen_paths: set[str] = set()
	for value in files_value:
		if not isinstance(value, dict):
			raise DepotError("build manifest contains a non-object file entry")
		path = _normalise_relative_path(value.get("p"), "file path")
		if path.casefold() in seen_paths:
			raise DepotError(f"build manifest has duplicate file path: {path}")
		seen_paths.add(path.casefold())
		file_build = _positive_int(value.get("b"), f"file build for {path}")
		size = _non_negative_int(value.get("s"), f"file size for {path}")
		sha1 = _string(value.get("h"), f"file hash for {path}")
		if not SHA1_PATTERN.fullmatch(sha1):
			raise DepotError(f"invalid SHA-1 hash for {path}")
		files.append(DepotFile(path, file_build, size, sha1.casefold()))

	deleted = _parse_deleted_paths(build_manifest.get("df"), seen_paths)
	create_if_missing = _parse_create_if_missing(depot_manifest.get("CreateIfMissing"))
	start_options, default_start_option = _parse_start_options(
		depot_manifest.get("StartOptions"),
		depot_manifest.get("DefaultStartOption"),
	)
	canonical_file_paths = {file.path.casefold(): file.path for file in files}
	start_options = tuple(
		DepotStartOption(
			name=option.name,
			path=canonical_file_paths.get(option.path.casefold(), option.path),
			arguments=option.arguments,
		)
		for option in start_options
	)
	return DepotSnapshot(
		region=region,
		server=server,
		name=name,
		build=build,
		version=version,
		files=tuple(files),
		deleted_files=deleted,
		create_if_missing=create_if_missing,
		start_options=start_options,
		default_start_option=default_start_option,
	)


def _parse_deleted_paths(value: object, known_paths: set[str]) -> tuple[str, ...]:
	if value is None:
		return ()
	if not isinstance(value, list):
		raise DepotError("build manifest df must be an array")
	paths: list[str] = []
	seen: set[str] = set()
	for item in value:
		path = _normalise_relative_path(item, "deleted file path")
		folded = path.casefold()
		if folded in known_paths:
			raise DepotError(f"manifest lists a file as both present and deleted: {path}")
		if folded in seen:
			continue
		seen.add(folded)
		paths.append(path)
	return tuple(paths)


def _parse_create_if_missing(value: object) -> tuple[CreateIfMissing, ...]:
	if value is None:
		return ()
	if not isinstance(value, list):
		raise DepotError("CreateIfMissing must be an array")
	entries: list[CreateIfMissing] = []
	seen: set[str] = set()
	for item in value:
		if not isinstance(item, dict):
			raise DepotError("CreateIfMissing contains a non-object entry")
		path = _normalise_relative_path(item.get("Path"), "CreateIfMissing path")
		folded = path.casefold()
		if folded in seen:
			raise DepotError(f"CreateIfMissing has duplicate path: {path}")
		seen.add(folded)
		content = _string(item.get("Content"), f"CreateIfMissing content for {path}")
		entry = CreateIfMissing(path, content)
		_decode_create_if_missing(entry)
		entries.append(entry)
	return tuple(entries)


def _parse_start_options(
	value: object,
	default_value: object,
) -> tuple[tuple[DepotStartOption, ...], str]:
	"""Validate the launch targets published by the official depot manifest."""

	# Older test fixtures and historical manifests did not expose StartOptions.
	# Babel remains the launcher-safe compatibility default for that case.
	if value is None:
		return (DEFAULT_DEPOT_START_OPTION,), DEFAULT_DEPOT_START_OPTION.name
	if not isinstance(value, list) or not value:
		raise DepotError("StartOptions must be a non-empty array")
	options: list[DepotStartOption] = []
	seen_names: set[str] = set()
	for item in value:
		if not isinstance(item, dict):
			raise DepotError("StartOptions contains a non-object entry")
		name = _start_option_text(item.get("Name"), "StartOptions Name", 64)
		folded_name = name.casefold()
		if folded_name in seen_names:
			raise DepotError(f"StartOptions has duplicate realm name: {name}")
		seen_names.add(folded_name)
		path = _normalise_relative_path(item.get("Path"), f"StartOptions path for {name}")
		if PurePosixPath(path).suffix.casefold() != ".exe":
			raise DepotError(f"StartOptions path is not an executable for {name}: {path}")
		arguments = _start_option_text(
			item.get("Arguments"),
			f"StartOptions arguments for {name}",
			2048,
		)
		options.append(DepotStartOption(name=name, path=path, arguments=arguments))

	if default_value is None:
		default_name = next(
			(
				option.name
				for option in options
				if option.name.casefold() == DEFAULT_DEPOT_START_OPTION.name.casefold()
			),
			options[0].name,
		)
	else:
		default_name = _start_option_text(
			default_value,
			"DefaultStartOption",
			64,
		)
		matching_default = next(
			(option.name for option in options if option.name.casefold() == default_name.casefold()),
			None,
		)
		if matching_default is None:
			raise DepotError(
				f"DefaultStartOption does not match an advertised realm: {default_name}"
			)
		default_name = matching_default
	return tuple(options), default_name


def _start_option_text(value: object, field: str, maximum_length: int) -> str:
	if not isinstance(value, str):
		raise DepotError(f"{field} must be a string")
	text = value.strip()
	if not text or len(text) > maximum_length or any(ord(character) < 32 for character in text):
		raise DepotError(f"invalid {field}")
	return text


def _decode_create_if_missing(entry: CreateIfMissing) -> bytes:
	try:
		content = b64decode(entry.content_base64, validate=True)
	except (ValueError, BinAsciiError) as exc:
		raise DepotError(f"invalid base64 create-if-missing entry: {entry.path}") from exc
	if len(content) > MAX_CREATE_IF_MISSING_BYTES:
		raise DepotError(f"create-if-missing entry is too large: {entry.path}")
	return content


def _server_candidates(
	configuration: Mapping[str, object] | None,
	preferred_region: str,
) -> tuple[tuple[str, str], ...]:
	entries: list[tuple[str, str]] = []
	if configuration is not None:
		regions = configuration.get("Regions")
		default_region = configuration.get("DefaultRegion")
		if isinstance(regions, list):
			ordered = [preferred_region]
			if isinstance(default_region, str) and default_region not in ordered:
				ordered.append(default_region)
			for region_name in ordered:
				for region in regions:
					if not isinstance(region, dict) or region.get("Name") != region_name:
						continue
					servers = region.get("Servers")
					if isinstance(servers, list):
						for server in servers:
							if not isinstance(server, str):
								continue
							try:
								entries.append((region_name, _normalise_server(server)))
							except DepotError:
								continue
	seen: set[str] = set()
	unique: list[tuple[str, str]] = []
	for region, server in [*entries, *(("Fallback", value) for value in FALLBACK_SERVERS)]:
		if server in seen:
			continue
		seen.add(server)
		unique.append((region, server))
	return tuple(unique)


def _normalise_server(value: str) -> str:
	server = value.strip().strip("/")
	if not server or "://" in server or any(character.isspace() for character in server):
		raise DepotError(f"invalid CDN server: {value!r}")
	parsed = urlparse(f"https://{server}")
	if not parsed.netloc or parsed.query or parsed.fragment or "@" in parsed.netloc:
		raise DepotError(f"invalid CDN server: {value!r}")
	parts = PurePosixPath(parsed.path).parts
	if any(part in (".", "..") for part in parts):
		raise DepotError(f"invalid CDN server path: {value!r}")
	return f"{parsed.netloc}{parsed.path.rstrip('/')}"


def _depot_manifest_url(server: str) -> str:
	return f"https://{server}/depots/{DEPOT_ID}/manifest.json"


def _build_manifest_url(server: str, build: int) -> str:
	return f"https://{server}/depots/{DEPOT_ID}/builds/{build}.json"


def _file_url(server: str, file: DepotFile) -> str:
	return (
		f"https://{server}/depots/{DEPOT_ID}/builds/{file.build}/"
		f"{quote(file.path, safe='/')}"
	)


def _ordered_download_servers(
	primary: str,
	candidates: Iterable[str],
) -> tuple[str, ...]:
	ordered: list[str] = []
	seen: set[str] = set()
	for value in (primary, *candidates):
		server = _normalise_server(value)
		if server in seen:
			continue
		seen.add(server)
		ordered.append(server)
	return tuple(ordered)


def _download_partial_path(destination: Path) -> Path:
	return destination.with_name(f".{destination.name}.ptlinux.partial")


def _recover_crlf_manifest_file(
	temporary: Path,
	destination: Path,
	file: DepotFile,
) -> bool:
	"""Restore publisher-normalized legacy text only on an exact hash match."""

	try:
		if temporary.stat().st_size > MAX_LINE_ENDING_RECOVERY_BYTES:
			return False
		data = temporary.read_bytes()
	except OSError as exc:
		raise DepotError(f"could not inspect downloaded {file.path}: {exc}") from exc
	if b"\r" in data:
		return False
	recovered = data.replace(b"\n", b"\r\n")
	if len(recovered) != file.size:
		return False
	if (
		hashlib.sha1(recovered, usedforsecurity=False).hexdigest().casefold()
		!= file.sha1.casefold()
	):
		return False
	recovered_temporary = destination.with_name(
		f".{destination.name}.{os.getpid()}.{uuid4().hex}.recovered"
	)
	try:
		with recovered_temporary.open("xb") as handle:
			handle.write(recovered)
			handle.flush()
			os.fsync(handle.fileno())
		recovered_temporary.replace(destination)
	except BaseException:
		recovered_temporary.unlink(missing_ok=True)
		raise
	return True


def _prepare_game_root(game_directory: Path, *, create: bool) -> Path:
	root = game_directory.expanduser()
	if root.is_symlink():
		raise DepotError(f"managed game directory must not be a symbolic link: {root}")
	if root.exists() and not root.is_dir():
		raise DepotError(f"managed game path is not a directory: {root}")
	if create:
		root.mkdir(parents=True, exist_ok=True)
	return root.resolve(strict=False)


def _safe_target(root: Path, relative_path: str) -> Path:
	parts = _relative_parts(relative_path, "depot path")
	target = root.joinpath(*parts)
	try:
		target.resolve(strict=False).relative_to(root)
	except ValueError as exc:
		raise DepotError(f"depot path escapes the managed game directory: {relative_path}") from exc
	return target


def _ensure_target_parent(root: Path, target: Path) -> None:
	try:
		relative_parent = target.parent.relative_to(root)
	except ValueError as exc:
		raise DepotError(f"target escapes managed game directory: {target}") from exc
	cursor = root
	for part in relative_parent.parts:
		cursor = cursor / part
		if cursor.is_symlink():
			raise DepotError(f"refusing to traverse symbolic link in game tree: {cursor}")
		if cursor.exists():
			if not cursor.is_dir():
				raise DepotError(f"cannot create game directory over file: {cursor}")
		else:
			try:
				cursor.mkdir()
			except FileExistsError:
				# Another download worker can create a shared parent after
				# ``exists()`` and before ``mkdir()``.  Accept only the directory
				# that the losing worker expected; never accept a file or symlink.
				if cursor.is_symlink():
					raise DepotError(
						f"refusing to traverse symbolic link in game tree: {cursor}"
					)
				if not cursor.is_dir():
					raise DepotError(f"cannot create game directory over file: {cursor}")
			except OSError as exc:
				raise DepotError(f"cannot create game directory {cursor}: {exc}") from exc


def _normalise_relative_path(value: object, field: str) -> str:
	if not isinstance(value, str):
		raise DepotError(f"{field} must be a string")
	parts = _relative_parts(value, field)
	return "/".join(parts)


def _relative_parts(value: str, field: str) -> tuple[str, ...]:
	if not value or "\x00" in value:
		raise DepotError(f"invalid {field}")
	normalised = value.replace("\\", "/")
	if normalised.startswith("/") or normalised.startswith("//"):
		raise DepotError(f"{field} must be relative: {value!r}")
	parts = PurePosixPath(normalised).parts
	if not parts or any(part in ("", ".", "..") or ":" in part for part in parts):
		raise DepotError(f"invalid {field}: {value!r}")
	return tuple(parts)


def _positive_int(value: object, field: str) -> int:
	if isinstance(value, bool) or not isinstance(value, int) or value <= 0:
		raise DepotError(f"{field} must be a positive integer")
	return value


def _non_negative_int(value: object, field: str) -> int:
	if isinstance(value, bool) or not isinstance(value, int) or value < 0:
		raise DepotError(f"{field} must be a non-negative integer")
	return value


def _string(value: object, field: str) -> str:
	if not isinstance(value, str) or not value.strip():
		raise DepotError(f"{field} must be a non-empty string")
	return value


def _sha1(path: Path) -> str:
	digest = hashlib.sha1(usedforsecurity=False)
	try:
		with path.open("rb") as handle:
			for chunk in iter(lambda: handle.read(HASH_READ_SIZE), b""):
				digest.update(chunk)
	except OSError as exc:
		raise DepotError(f"cannot hash {path}: {exc}") from exc
	return digest.hexdigest()


def _content_length(value: str | None) -> int | None:
	if value is None:
		return None
	try:
		length = int(value)
	except ValueError as exc:
		raise DepotError(f"invalid Content-Length: {value!r}") from exc
	if length < 0:
		raise DepotError(f"invalid Content-Length: {value!r}")
	return length


def _content_range(value: str | None) -> tuple[int, int, int]:
	if value is None:
		raise DepotError("206 response omitted Content-Range")
	match = CONTENT_RANGE_PATTERN.fullmatch(value.strip())
	if match is None:
		raise DepotError(f"invalid Content-Range: {value!r}")
	start, end, total = (int(part) for part in match.groups())
	if start > end or end >= total:
		raise DepotError(f"invalid Content-Range: {value!r}")
	return start, end, total


def _headers() -> dict[str, str]:
	return {"User-Agent": f"PT-Linux/{__version__} (+https://pristontale.eu/)"}


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")
	try:
		temporary.write_text(
			json.dumps(value, indent=2, sort_keys=True) + "\n",
			encoding="utf-8",
		)
		temporary.replace(path)
	except BaseException:
		temporary.unlink(missing_ok=True)
		raise


def _emit_progress(
	callback: ProgressCallback | None,
	phase: str,
	files_completed: int,
	files_total: int,
	bytes_completed: int,
	bytes_total: int,
	current_file: str | None,
) -> None:
	if callback is not None:
		callback(
			SyncProgress(
				phase,
				files_completed,
				files_total,
				bytes_completed,
				bytes_total,
				current_file,
			)
		)