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 collections import deque
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_INITIAL_WORKERS = DOWNLOAD_WORKERS
LOW_BANDWIDTH_WORKERS = 2
LOW_BANDWIDTH_BYTES_PER_SECOND = 5_000_000
DOWNLOAD_RETRY_DELAYS_SECONDS = (1.0, 2.0, 4.0, 8.0, 16.0)
PROGRESS_UPDATE_INTERVAL_SECONDS = 0.075
ADAPTIVE_EVALUATION_INTERVAL_SECONDS = 1.0
ADAPTIVE_RETRY_WINDOW_SECONDS = 15.0
ADAPTIVE_RETRY_RECOVERY_SECONDS = 5.0
TRANSFER_RATE_WINDOW_SECONDS = 5.0
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
DEPOT_FILE_CATALOG_SCHEMA = 1
MAX_PREFLIGHT_ISSUES = 32


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 DownloadOptions:
	"""User-selected limits around the adaptive depot scheduler."""

	bandwidth_limit_bytes_per_second: int | None = None
	low_bandwidth: bool = False
	minimum_workers: int = 1
	maximum_workers: int | None = None

	def __post_init__(self) -> None:
		if not isinstance(self.low_bandwidth, bool):
			raise ValueError("low_bandwidth must be a boolean")
		if (
			self.bandwidth_limit_bytes_per_second is not None
			and (
				isinstance(self.bandwidth_limit_bytes_per_second, bool)
				or self.bandwidth_limit_bytes_per_second <= 0
			)
		):
			raise ValueError("bandwidth limit must be positive or None")
		if isinstance(self.minimum_workers, bool) or self.minimum_workers <= 0:
			raise ValueError("minimum download workers must be positive")
		if self.maximum_workers is not None and (
			isinstance(self.maximum_workers, bool) or self.maximum_workers <= 0
		):
			raise ValueError("maximum download workers must be positive or None")
		if (
			self.maximum_workers is not None
			and self.minimum_workers > self.maximum_workers
		):
			raise ValueError("minimum download workers exceed the maximum")
		if self.low_bandwidth and self.minimum_workers > LOW_BANDWIDTH_WORKERS:
			raise ValueError("low-bandwidth minimum workers exceed its worker cap")

	@property
	def effective_bandwidth_limit(self) -> int | None:
		configured = self.bandwidth_limit_bytes_per_second
		if not self.low_bandwidth:
			return configured
		return min(
			configured if configured is not None else LOW_BANDWIDTH_BYTES_PER_SECOND,
			LOW_BANDWIDTH_BYTES_PER_SECOND,
		)

	@property
	def effective_maximum_workers(self) -> int:
		configured = self.maximum_workers or DOWNLOAD_WORKERS
		if self.low_bandwidth:
			configured = min(configured, LOW_BANDWIDTH_WORKERS)
		return max(self.minimum_workers, configured)


@dataclass(frozen=True)
class DownloadTelemetry:
	"""One internal measurement produced while retrieving a depot file."""

	kind: str
	file: str
	server: str
	attempt: int
	latency_seconds: float | None = None
	bytes_written: int = 0
	write_seconds: float = 0.0
	reason: str | None = None


DownloadObserver = Callable[[DownloadTelemetry], None]


class BandwidthLimiter:
	"""Thread-safe aggregate byte-rate limiter with a one-chunk burst allowance."""

	def __init__(
		self,
		bytes_per_second: int,
		*,
		clock: Callable[[], float] = time.monotonic,
		sleeper: Callable[[float], None] = time.sleep,
	) -> None:
		if isinstance(bytes_per_second, bool) or bytes_per_second <= 0:
			raise ValueError("bandwidth limit must be positive")
		self.bytes_per_second = bytes_per_second
		self._clock = clock
		self._sleeper = sleeper
		self._lock = threading.Lock()
		self._next_available_at = clock()

	def consume(self, byte_count: int) -> None:
		"""Wait until this newly received chunk's aggregate reservation begins.

		The network read happens before this call. Reserving the chunk's *start*
		keeps all workers under one aggregate rate without charging again for time
		already spent receiving it. The first chunk is the bucket's burst allowance.
		"""

		if byte_count < 0:
			raise ValueError("byte count must not be negative")
		if byte_count == 0:
			return
		with self._lock:
			now = self._clock()
			start = max(now, self._next_available_at)
			self._next_available_at = start + (byte_count / self.bytes_per_second)
			delay = start - now
		if delay > 0:
			self._sleeper(delay)


class AdaptiveConcurrencyController:
	"""Select a bounded worker target from live network, host, and plan data."""

	def __init__(
		self,
		options: DownloadOptions,
		*,
		clock: Callable[[], float] = time.monotonic,
		cpu_load: Callable[[], float | None] | None = None,
	) -> None:
		self.options = options
		self._clock = clock
		self._cpu_load = cpu_load or _normalized_cpu_load
		self._lock = threading.Lock()
		self._target = max(
			options.minimum_workers,
			min(DOWNLOAD_INITIAL_WORKERS, options.effective_maximum_workers),
		)
		self._last_evaluation = clock()
		self._transferred_bytes = 0
		self._transfer_samples: deque[tuple[float, int]] = deque(
			[(self._last_evaluation, 0)]
		)
		self._response_samples: deque[tuple[float, float]] = deque()
		self._retry_samples: deque[float] = deque()
		self._retry_count = 0
		self._handled_retry_count = 0
		self._last_retry_at: float | None = None
		self._latency_seconds: float | None = None
		self._disk_samples: deque[tuple[float, int, float]] = deque()

	def record_response(self, latency_seconds: float) -> None:
		if latency_seconds < 0:
			return
		now = self._clock()
		with self._lock:
			self._response_samples.append((now, latency_seconds))
			self._prune_health_samples(now)
			if self._latency_seconds is None:
				self._latency_seconds = latency_seconds
			else:
				self._latency_seconds = (
					(self._latency_seconds * 0.75) + (latency_seconds * 0.25)
				)

	def record_retry(self) -> None:
		now = self._clock()
		with self._lock:
			self._retry_count += 1
			self._retry_samples.append(now)
			self._last_retry_at = now
			self._prune_health_samples(now)

	def record_chunk(self, byte_count: int, write_seconds: float) -> None:
		if byte_count <= 0:
			return
		now = self._clock()
		with self._lock:
			self._transferred_bytes += byte_count
			self._transfer_samples.append((now, self._transferred_bytes))
			_prune_transfer_samples(self._transfer_samples, now)
			self._disk_samples.append((now, byte_count, max(0.0, write_seconds)))
			self._prune_disk_samples(now)

	def _prune_health_samples(self, now: float) -> None:
		cutoff = now - ADAPTIVE_RETRY_WINDOW_SECONDS
		while self._response_samples and self._response_samples[0][0] < cutoff:
			self._response_samples.popleft()
		while self._retry_samples and self._retry_samples[0] < cutoff:
			self._retry_samples.popleft()

	def _prune_disk_samples(self, now: float) -> None:
		cutoff = now - TRANSFER_RATE_WINDOW_SECONDS
		while self._disk_samples and self._disk_samples[0][0] < cutoff:
			self._disk_samples.popleft()

	def target_workers(self, remaining_files: int, remaining_bytes: int) -> int:
		"""Return the next target; active workers are allowed to drain naturally."""

		if remaining_files <= 0:
			return 0
		maximum = min(self.options.effective_maximum_workers, remaining_files)
		minimum = min(self.options.minimum_workers, maximum)
		with self._lock:
			self._target = min(maximum, max(minimum, self._target))
			now = self._clock()
			average_file_size = remaining_bytes / max(1, remaining_files)
			if now - self._last_evaluation < ADAPTIVE_EVALUATION_INTERVAL_SECONDS:
				return self._target
			self._last_evaluation = now
			_prune_transfer_samples(self._transfer_samples, now)
			self._prune_health_samples(now)
			self._prune_disk_samples(now)
			rate = _sample_rate(self._transfer_samples)
			attempts = len(self._response_samples) + len(self._retry_samples)
			error_rate = len(self._retry_samples) / attempts if attempts else 0.0
			latency = self._latency_seconds
			disk_bytes = sum(sample[1] for sample in self._disk_samples)
			disk_seconds = sum(sample[2] for sample in self._disk_samples)
			disk_rate = (
				disk_bytes / disk_seconds
				if disk_seconds > 0
				else None
			)
			try:
				cpu_load = self._cpu_load()
			except (OSError, ValueError):
				cpu_load = None

			new_retries = self._retry_count - self._handled_retry_count
			if new_retries:
				self._handled_retry_count = self._retry_count
				if error_rate >= 0.12 and self._target > minimum:
					backoff = min(2, new_retries)
					self._target = max(minimum, self._target - backoff)
				return self._target

			resource_constrained = (
				(cpu_load is not None and cpu_load >= 0.90)
				or (
					rate is not None
					and disk_rate is not None
					and disk_rate < rate * 0.75
				)
			)
			if resource_constrained and self._target > minimum:
				self._target -= 1
				return self._target

			limit = self.options.effective_bandwidth_limit
			limit_saturated = (
				limit is not None and rate is not None and rate >= limit * 0.90
			)
			healthy = (
				(cpu_load is None or cpu_load < 0.80)
				and (
					rate is None
					or disk_rate is None
					or disk_rate >= rate
				)
			)
			retry_recovery_pending = (
				self._last_retry_at is not None
				and now - self._last_retry_at < ADAPTIVE_RETRY_RECOVERY_SECONDS
			)
			if (
				healthy
				and not retry_recovery_pending
				and not limit_saturated
				and self._target < maximum
			):
				# Parallel requests hide CDN round-trip time and are especially useful
				# for this depot's large population of small files.
				step = 2 if (
					(latency is not None and latency >= 0.75)
					or average_file_size <= 512 * 1024
				) else 1
				self._target = min(maximum, self._target + step)
			return min(maximum, max(minimum, self._target))


@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


@dataclass(frozen=True)
class DepotFileCatalogEntry:
	"""One manifest file plus the identity recorded after verified installation."""

	path: str
	size: int
	sha1: str
	device: int
	inode: int
	mtime_ns: int
	ctime_ns: int
	verified: bool


@dataclass(frozen=True)
class DepotFileCatalog:
	"""Cached manifest used for fast, network-free launch preflight checks."""

	build: int
	files: tuple[DepotFileCatalogEntry, ...]
	created_at: str


@dataclass(frozen=True)
class DepotIntegrityIssue:
	path: str
	reason: str


@dataclass(frozen=True)
class DepotPreflightResult:
	"""Outcome of validating cached file identities before starting the game."""

	build: int
	files_total: int
	verified_files: int
	unverified_files: int
	changed_files_hashed: int
	issue_count: int
	issues: tuple[DepotIntegrityIssue, ...]
	duration_ms: int

	@property
	def ready(self) -> bool:
		return self.issue_count == 0

	@property
	def fully_verified(self) -> bool:
		return self.ready and self.unverified_files == 0


@dataclass(frozen=True)
class _ScannedGameEntry:
	path: Path
	state: os.stat_result
	kind: 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,
	)


def cache_depot_file_catalog(
	path: Path,
	game_directory: Path,
	snapshot: DepotSnapshot,
	*,
	verified_paths: Iterable[str],
	previous: DepotFileCatalog | None = None,
) -> DepotFileCatalog:
	"""Record identities without rehashing files already verified by the updater."""

	root = _prepare_game_root(game_directory, create=False)
	scanned_files = _scan_game_tree(root)
	explicitly_verified = {verified_path.casefold() for verified_path in verified_paths}
	previous_files = (
		{entry.path.casefold(): entry for entry in previous.files}
		if previous is not None
		else {}
	)
	entries: list[DepotFileCatalogEntry] = []
	for file in snapshot.files:
		scanned = scanned_files.get(file.path)
		if scanned is None:
			raise DepotError(f"cannot cache missing depot file identity: {file.path}")
		state = scanned.state
		if scanned.kind != "file" or state.st_size != file.size:
			raise DepotError(f"cannot cache invalid depot file identity: {file.path}")
		prior = previous_files.get(file.path.casefold())
		verified = file.path.casefold() in explicitly_verified
		if (
			not verified
			and prior is not None
			and prior.verified
			and prior.size == file.size
			and prior.sha1 == file.sha1
			and _catalog_identity_matches(prior, state)
		):
			verified = True
		entries.append(_catalog_entry(file, state, verified=verified))

	catalog = DepotFileCatalog(
		build=snapshot.build,
		files=tuple(entries),
		created_at=datetime.now().astimezone().isoformat(),
	)
	_write_json(
		path,
		_depot_file_catalog_payload(catalog),
		compact=True,
	)
	return catalog


def read_cached_depot_file_catalog(path: Path) -> DepotFileCatalog | None:
	"""Read and strictly validate the local manifest/identity cache."""

	try:
		value = json.loads(path.read_text(encoding="utf-8"))
		if not isinstance(value, dict):
			return None
		if value.get("schema") != DEPOT_FILE_CATALOG_SCHEMA:
			return None
		build = _positive_int(value.get("build"), "cached depot file build")
		created_at = _string(value.get("created_at"), "cached depot file timestamp")
		raw_files = value.get("files")
		if not isinstance(raw_files, list):
			return None
		entries: list[DepotFileCatalogEntry] = []
		seen: set[str] = set()
		for raw in raw_files:
			entry = _parse_depot_file_catalog_entry(raw)
			if entry.path.casefold() in seen:
				return None
			seen.add(entry.path.casefold())
			entries.append(entry)
	except (OSError, UnicodeError, json.JSONDecodeError, DepotError):
		return None
	return DepotFileCatalog(build=build, files=tuple(entries), created_at=created_at)


def read_current_depot_file_catalog(paths: AppPaths) -> DepotFileCatalog | None:
	"""Return the file catalog only when it matches the last completed sync."""

	catalog = read_cached_depot_file_catalog(paths.depot_file_catalog)
	if catalog is None:
		return None
	if _completed_depot_build(paths.native_depot_state) != catalog.build:
		return None
	return catalog


def preflight_managed_game(
	paths: AppPaths,
	*,
	catalog: DepotFileCatalog | None = None,
) -> DepotPreflightResult | None:
	"""Check all managed paths cheaply and hash only files whose identity changed."""

	catalog = catalog or read_current_depot_file_catalog(paths)
	if catalog is None:
		return None
	if _completed_depot_build(paths.native_depot_state) != catalog.build:
		return None
	started = time.perf_counter()
	root = _prepare_game_root(paths.source_game, create=False)
	if not root.is_dir():
		return DepotPreflightResult(
			build=catalog.build,
			files_total=len(catalog.files),
			verified_files=0,
			unverified_files=sum(not entry.verified for entry in catalog.files),
			changed_files_hashed=0,
			issue_count=1,
			issues=(DepotIntegrityIssue(".", "managed game directory is missing"),),
			duration_ms=_elapsed_milliseconds(started),
		)
	try:
		scanned_files = _scan_game_tree(root)
	except DepotError as exc:
		return DepotPreflightResult(
			build=catalog.build,
			files_total=len(catalog.files),
			verified_files=0,
			unverified_files=sum(not entry.verified for entry in catalog.files),
			changed_files_hashed=0,
			issue_count=1,
			issues=(DepotIntegrityIssue(".", str(exc)),),
			duration_ms=_elapsed_milliseconds(started),
		)

	verified_files = 0
	unverified_files = 0
	changed_files_hashed = 0
	issue_count = 0
	issues: list[DepotIntegrityIssue] = []
	updated_entries: list[DepotFileCatalogEntry] = []
	cache_changed = False
	for entry in catalog.files:
		reason = None
		state: os.stat_result | None = None
		scanned = scanned_files.get(entry.path)
		if scanned is None:
			reason = "missing"
		elif scanned.kind != "file":
			reason = scanned.kind
		elif scanned.state.st_size != entry.size:
			reason = "size mismatch"
		else:
			state = scanned.state

		updated_entry = entry
		if reason is None and state is not None:
			if not entry.verified:
				unverified_files += 1
			elif _catalog_identity_matches(entry, state):
				verified_files += 1
			else:
				changed_files_hashed += 1
				try:
					assert scanned is not None
					matches = _sha1(scanned.path).casefold() == entry.sha1
				except DepotError as exc:
					reason = f"cannot verify changed file: {exc}"
				else:
					if matches:
						verified_files += 1
						updated_entry = _catalog_entry_from_cached(entry, state)
						cache_changed = True
					else:
						reason = "hash mismatch"

		if reason is not None:
			issue_count += 1
			if len(issues) < MAX_PREFLIGHT_ISSUES:
				issues.append(DepotIntegrityIssue(entry.path, reason))
		updated_entries.append(updated_entry)

	if cache_changed:
		updated_catalog = DepotFileCatalog(
			build=catalog.build,
			files=tuple(updated_entries),
			created_at=datetime.now().astimezone().isoformat(),
		)
		_write_json(
			paths.depot_file_catalog,
			_depot_file_catalog_payload(updated_catalog),
			compact=True,
		)

	return DepotPreflightResult(
		build=catalog.build,
		files_total=len(catalog.files),
		verified_files=verified_files,
		unverified_files=unverified_files,
		changed_files_hashed=changed_files_hashed,
		issue_count=issue_count,
		issues=tuple(issues),
		duration_ms=_elapsed_milliseconds(started),
	)


@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
	hash_verified_paths: tuple[str, ...] = ()


@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
	bytes_remaining: int = 0
	transfer_rate_bytes_per_second: float | None = None
	eta_seconds: float | None = None
	current_server: str | None = None
	retry_count: int = 0
	retry_reason: str | None = None
	active_workers: int = 0
	target_workers: 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
	files_planned: int = 0
	bytes_planned: int = 0


@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,
		download_observer: DownloadObserver | None = None,
	) -> 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, ...] = ()
		self._download_observer = download_observer
		self._bandwidth_limiter: BandwidthLimiter | None = None

	@property
	def download_observer(self) -> DownloadObserver | None:
		return self._download_observer

	@download_observer.setter
	def download_observer(self, observer: DownloadObserver | None) -> None:
		self._download_observer = observer

	@property
	def bandwidth_limiter(self) -> BandwidthLimiter | None:
		return self._bandwidth_limiter

	@bandwidth_limiter.setter
	def bandwidth_limiter(self, limiter: BandwidthLimiter | None) -> None:
		self._bandwidth_limiter = limiter

	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,
					attempt=attempt_index + 1,
					progress=progress,
				)
				return
			except _RetryableDownloadError as exc:
				last_error = exc
				self._notify_download(
					DownloadTelemetry(
						kind="retry",
						file=file.path,
						server=attempt_server,
						attempt=attempt_index + 1,
						reason=str(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,
		*,
		attempt: int,
		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")
		self._notify_download(
			DownloadTelemetry(
				kind="request",
				file=file.path,
				server=server,
				attempt=attempt,
			)
		)
		request_started_at = time.monotonic()
		try:
			response = self._opener(request, timeout=self.timeout_seconds)
		except HTTPError as exc:
			status = exc.code
			exc.close()
			if status == 416 and requested_offset:
				raise _RetryableDownloadError(
					f"server rejected the saved offset for {file.path}",
					discard_partial=True,
				) from exc
			raise _RetryableDownloadError(
				f"HTTP {status} 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
		self._notify_download(
			DownloadTelemetry(
				kind="response",
				file=file.path,
				server=server,
				attempt=attempt,
				latency_seconds=max(0.0, time.monotonic() - request_started_at),
			)
		)

		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,
							)
						if self._bandwidth_limiter is not None:
							self._bandwidth_limiter.consume(len(chunk))
						digest.update(chunk)
						write_started_at = time.monotonic()
						handle.write(chunk)
						self._notify_download(
							DownloadTelemetry(
								kind="chunk",
								file=file.path,
								server=server,
								attempt=attempt,
								bytes_written=len(chunk),
								write_seconds=max(
									0.0,
									time.monotonic() - write_started_at,
								),
							)
						)
						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 _notify_download(self, event: DownloadTelemetry) -> None:
		observer = self._download_observer
		if observer is not None:
			# noinspection PyBroadException
			try:
				observer(event)
			except Exception:
				# Telemetry must never weaken verified download behavior.
				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] = []
	hash_verified_paths: list[str] = []
	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"))
			else:
				hash_verified_paths.append(file.path)

	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),
		hash_verified_paths=tuple(hash_verified_paths),
	)


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,
	download_options: DownloadOptions | None = None,
) -> DepotSyncResult:
	"""Synchronize the managed game directory using only official public data."""

	paths.create_base_directories()
	depot_client = client or DepotClient()
	previous_file_catalog = read_current_depot_file_catalog(paths)
	_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,
		download_options or DownloadOptions(),
	)
	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,
	)
	verified_paths = {
		*plan.hash_verified_paths,
		*(download.file.path for download in plan.downloads),
	}
	cache_depot_file_catalog(
		paths.depot_file_catalog,
		root,
		snapshot,
		verified_paths=verified_paths,
		previous=previous_file_catalog,
	)
	_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,
	options: DownloadOptions,
) -> None:
	if not plan.downloads:
		return

	lock = threading.Lock()
	emit_lock = threading.Lock()
	current_bytes: dict[str, int] = {}
	current_servers: dict[str, str] = {}
	completed_files = 0
	completed_bytes = 0
	transferred_bytes = 0
	retry_count = 0
	retry_reason: str | None = None
	active_workers = 0
	target_workers = 0
	last_progress_emitted_at = 0.0
	started_at = time.monotonic()
	transfer_samples: deque[tuple[float, int]] = deque([(started_at, 0)])
	controller = AdaptiveConcurrencyController(options)
	bandwidth_limit = options.effective_bandwidth_limit
	limiter = BandwidthLimiter(bandwidth_limit) if bandwidth_limit is not None else None
	previous_observer = client.download_observer
	previous_limiter = client.bandwidth_limiter

	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:
				bytes_completed = min(
					plan.bytes_total,
					completed_bytes + sum(current_bytes.values()),
				)
				_prune_transfer_samples(transfer_samples, now)
				transfer_rate = _sample_rate(transfer_samples)
				bytes_remaining = max(0, plan.bytes_total - bytes_completed)
				eta_seconds = (
					bytes_remaining / transfer_rate
					if transfer_rate is not None and transfer_rate > 0
					else None
				)
				event = SyncProgress(
					"downloading",
					completed_files,
					len(plan.downloads),
					bytes_completed,
					plan.bytes_total,
					current_file,
					transferred_bytes,
					bytes_remaining,
					transfer_rate,
					eta_seconds,
					current_servers.get(
						current_file,
						_server_display_name(depot_snapshot.server),
					),
					retry_count,
					retry_reason,
					active_workers,
					target_workers,
				)
			progress(event)

	def observe_download(event: DownloadTelemetry) -> None:
		nonlocal transferred_bytes, retry_count, retry_reason
		if event.kind == "response" and event.latency_seconds is not None:
			controller.record_response(event.latency_seconds)
		elif event.kind == "retry":
			controller.record_retry()
		elif event.kind == "chunk":
			controller.record_chunk(event.bytes_written, event.write_seconds)

		force = event.kind == "retry"
		if event.kind in {"request", "response", "retry", "chunk"}:
			with lock:
				current_servers[event.file] = _server_display_name(event.server)
				if event.kind == "chunk":
					transferred_bytes += event.bytes_written
					now = time.monotonic()
					transfer_samples.append((now, transferred_bytes))
					_prune_transfer_samples(transfer_samples, now)
				if event.kind == "retry":
					retry_count += 1
					retry_reason = event.reason
		if previous_observer is not None:
			previous_observer(event)
		if event.kind != "chunk":
			emit_download_progress(event.file, force=force)

	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:
			with lock:
				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,
		)

	maximum_workers = min(options.effective_maximum_workers, len(plan.downloads))
	# Start large transfers first so they overlap the small-file request workload
	# instead of becoming a single-connection tail at the end of an install.
	remaining_items: deque[PlannedDownload] = deque(
		sorted(plan.downloads, key=lambda item: item.file.size, reverse=True)
	)
	client.download_observer = observe_download
	client.bandwidth_limiter = limiter
	try:
		with ThreadPoolExecutor(max_workers=maximum_workers) as executor:
			future_items: dict[Future[None], PlannedDownload] = {}

			def submit_until(worker_target: int) -> None:
				nonlocal active_workers, target_workers
				with lock:
					target_workers = worker_target
				while remaining_items and len(future_items) < worker_target:
					next_item = remaining_items.popleft()
					with lock:
						active_workers += 1
					try:
						submitted_future = executor.submit(download, next_item)
					except BaseException:
						with lock:
							active_workers -= 1
						remaining_items.appendleft(next_item)
						raise
					future_items[submitted_future] = next_item

			try:
				initial_target = controller.target_workers(
					len(remaining_items),
					plan.bytes_total,
				)
				submit_until(initial_target)
				while future_items:
					done, _pending = wait(
						tuple(future_items),
						timeout=ADAPTIVE_EVALUATION_INTERVAL_SECONDS,
						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:
							active_workers -= 1
							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,
						)
						with lock:
							current_servers.pop(completed_item.file.path, None)
					with lock:
						in_progress_bytes = sum(current_bytes.values())
					remaining_file_count = len(remaining_items) + len(future_items)
					remaining_byte_count = max(
						0,
						plan.bytes_total - completed_bytes - in_progress_bytes,
					)
					next_target = controller.target_workers(
						remaining_file_count,
						remaining_byte_count,
					)
					submit_until(next_target)
			except BaseException:
				for future in future_items:
					future.cancel()
				raise
	finally:
		client.download_observer = previous_observer
		client.bandwidth_limiter = previous_limiter


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 _catalog_entry(
	file: DepotFile,
	state: os.stat_result,
	*,
	verified: bool,
) -> DepotFileCatalogEntry:
	return DepotFileCatalogEntry(
		path=file.path,
		size=file.size,
		sha1=file.sha1.casefold(),
		device=state.st_dev,
		inode=state.st_ino,
		mtime_ns=state.st_mtime_ns,
		ctime_ns=state.st_ctime_ns,
		verified=verified,
	)


def _scan_game_tree(root: Path) -> dict[str, _ScannedGameEntry]:
	"""Inventory files without following symlinks or resolving every manifest path."""

	entries: dict[str, _ScannedGameEntry] = {}

	def scan(directory: Path, relative_parent: PurePosixPath) -> None:
		children = _scan_directory(directory)
		for child in children:
			relative = relative_parent / child.name
			key = relative.as_posix()
			try:
				state = child.stat(follow_symlinks=False)
			except OSError as exc:
				raise DepotError(f"cannot inspect managed game path {key}: {exc}") from exc
			path = Path(child.path)
			if stat.S_ISLNK(state.st_mode):
				entries[key] = _ScannedGameEntry(path, state, "symbolic link")
			elif stat.S_ISDIR(state.st_mode):
				scan(path, relative)
			elif stat.S_ISREG(state.st_mode):
				entries[key] = _ScannedGameEntry(path, state, "file")
			else:
				entries[key] = _ScannedGameEntry(path, state, "not a regular file")

	scan(root, PurePosixPath())
	return entries


def _scan_directory(directory: Path) -> list[os.DirEntry[str]]:
	try:
		with os.scandir(directory) as iterator:
			return list(iterator)
	except OSError as exc:
		raise DepotError(f"cannot scan managed game directory {directory}: {exc}") from exc


def _catalog_entry_from_cached(
	entry: DepotFileCatalogEntry,
	state: os.stat_result,
) -> DepotFileCatalogEntry:
	return DepotFileCatalogEntry(
		path=entry.path,
		size=entry.size,
		sha1=entry.sha1,
		device=state.st_dev,
		inode=state.st_ino,
		mtime_ns=state.st_mtime_ns,
		ctime_ns=state.st_ctime_ns,
		verified=True,
	)


def _catalog_identity_matches(
	entry: DepotFileCatalogEntry,
	state: os.stat_result,
) -> bool:
	return (
		entry.device == state.st_dev
		and entry.inode == state.st_ino
		and entry.size == state.st_size
		and entry.mtime_ns == state.st_mtime_ns
		and entry.ctime_ns == state.st_ctime_ns
	)


def _depot_file_catalog_payload(catalog: DepotFileCatalog) -> dict[str, object]:
	"""Use compact rows because this cache contains tens of thousands of files."""

	return {
		"schema": DEPOT_FILE_CATALOG_SCHEMA,
		"build": catalog.build,
		"created_at": catalog.created_at,
		"files": [
			[
				entry.path,
				entry.size,
				entry.sha1,
				entry.device,
				entry.inode,
				entry.mtime_ns,
				entry.ctime_ns,
				entry.verified,
			]
			for entry in catalog.files
		],
	}


def _parse_depot_file_catalog_entry(value: object) -> DepotFileCatalogEntry:
	if not isinstance(value, list) or len(value) != 8:
		raise DepotError("cached depot file entry must be an eight-value row")
	path = _normalise_relative_path(value[0], "cached depot file path")
	size = _non_negative_int(value[1], f"cached depot file size for {path}")
	sha1 = _string(value[2], f"cached depot file hash for {path}").casefold()
	if not SHA1_PATTERN.fullmatch(sha1):
		raise DepotError(f"invalid cached SHA-1 hash for {path}")
	identity: dict[str, int] = {}
	for field_name, field_value in zip(
		("device", "inode", "mtime_ns", "ctime_ns"),
		value[3:7],
		strict=True,
	):
		identity[field_name] = _non_negative_int(
			field_value,
			f"cached depot file {field_name} for {path}",
		)
	verified = value[7]
	if not isinstance(verified, bool):
		raise DepotError(f"cached depot verification flag is invalid for {path}")
	return DepotFileCatalogEntry(
		path=path,
		size=size,
		sha1=sha1,
		device=identity["device"],
		inode=identity["inode"],
		mtime_ns=identity["mtime_ns"],
		ctime_ns=identity["ctime_ns"],
		verified=verified,
	)


def _completed_depot_build(path: Path) -> int | None:
	try:
		value = json.loads(path.read_text(encoding="utf-8"))
	except (OSError, UnicodeError, json.JSONDecodeError):
		return None
	if not isinstance(value, dict) or value.get("dry_run") is not False:
		return None
	build = value.get("build")
	if isinstance(build, bool) or not isinstance(build, int) or build <= 0:
		return None
	return build


def _elapsed_milliseconds(started: float) -> int:
	return max(0, round((time.perf_counter() - started) * 1000))


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_planned=len(plan.downloads),
		bytes_planned=plan.bytes_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 _normalized_cpu_load() -> float | None:
	"""Return one-minute runnable load normalized by the available CPU count."""

	try:
		load = os.getloadavg()[0]
	except (AttributeError, OSError):
		return None
	processors = os.cpu_count()
	if processors is None or processors <= 0:
		return None
	return max(0.0, load / processors)


def _prune_transfer_samples(
	samples: deque[tuple[float, int]],
	now: float,
) -> None:
	cutoff = now - TRANSFER_RATE_WINDOW_SECONDS
	while len(samples) > 2 and samples[1][0] <= cutoff:
		samples.popleft()


def _sample_rate(samples: deque[tuple[float, int]]) -> float | None:
	if len(samples) < 2:
		return None
	first_time, first_bytes = samples[0]
	last_time, last_bytes = samples[-1]
	elapsed = last_time - first_time
	if elapsed <= 0 or last_bytes < first_bytes:
		return None
	return (last_bytes - first_bytes) / elapsed


def _server_display_name(server: str) -> str:
	parsed = urlparse(server if "://" in server else f"//{server}")
	return parsed.hostname or server.split("/", 1)[0]


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


def _write_json(path: Path, value: object, *, compact: bool = False) -> None:
	path.parent.mkdir(parents=True, exist_ok=True)
	temporary = path.with_name(f".{path.name}.{os.getpid()}.tmp")
	try:
		payload = (
			json.dumps(value, separators=(",", ":"), sort_keys=True)
			if compact
			else json.dumps(value, indent=2, sort_keys=True)
		)
		temporary.write_text(payload + "\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,
				bytes_remaining=max(0, bytes_total - bytes_completed),
			)
		)