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

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

from __future__ import annotations

from collections import deque
from concurrent.futures import FIRST_COMPLETED, Future, ThreadPoolExecutor, wait
from dataclasses import asdict, dataclass
from datetime import datetime
import json
import os
from pathlib import Path
import re
import threading
import time
from typing import Callable, Iterable, Mapping
from urllib.parse import urlparse
from urllib.request import urlopen

from . import __version__
from .game.depot.catalogs import (
	DEPOT_FILE_CATALOG_SCHEMA,
	DEPOT_LAUNCH_CATALOG_SCHEMA,
	depot_file_catalog_payload as _depot_file_catalog_payload,
	launch_catalog_from_snapshot,
	launch_catalog_payload as _launch_catalog_payload,
	parse_depot_file_catalog as _parse_depot_file_catalog,
	parse_depot_file_catalog_entry as _parse_depot_file_catalog_entry,
	parse_launch_catalog as _parse_launch_catalog,
	select_depot_start_option,
)
from .game.depot.manifest import (
	MAX_CREATE_IF_MISSING_BYTES,
	SHA1_PATTERN,
	decode_create_if_missing as _decode_create_if_missing,
	non_negative_int as _non_negative_int,
	normalize_relative_path as _normalise_relative_path,
	parse_create_if_missing as _parse_create_if_missing,
	parse_deleted_paths as _parse_deleted_paths,
	parse_snapshot as _parse_snapshot,
	parse_start_options as _parse_start_options,
	positive_int as _positive_int,
	relative_parts as _relative_parts,
	required_string as _string,
	start_option_text as _start_option_text,
)
from .game.depot.integrity import (
	ScannedGameEntry as _ScannedGameEntry,
	build_depot_file_catalog as _build_depot_file_catalog,
	catalog_entry as _catalog_entry,
	catalog_entry_from_cached as _catalog_entry_from_cached,
	catalog_identity_matches as _catalog_identity_matches,
	inspect_managed_game as _inspect_managed_game,
	plan_update as _plan_update,
	prepare_game_root as _prepare_game_root,
	safe_target as _safe_target,
	scan_directory as _scan_directory,
	scan_game_tree as _scan_game_tree,
	sha1_file as _sha1_file,
)
from .game.depot.filesystem import (
	DEFAULT_MAX_LINE_ENDING_RECOVERY_BYTES,
	DEFAULT_UNMANAGED_EXECUTABLE_SUFFIXES,
	apply_create_if_missing as _filesystem_apply_create_if_missing,
	apply_deletions as _filesystem_apply_deletions,
	ensure_target_parent as _filesystem_ensure_target_parent,
	quarantine_unmanaged_files as _filesystem_quarantine_unmanaged_files,
	recover_crlf_manifest_file as _filesystem_recover_crlf_manifest_file,
	safe_imported_user_file as _filesystem_safe_imported_user_file,
	write_json as _filesystem_write_json,
)
from .game.depot.transfer import (
	AdaptiveConcurrencyController as _AdaptiveConcurrencyController,
	BandwidthLimiter,
	DepotHttpClient as _DepotHttpClient,
	DepotHttpPolicy as _DepotHttpPolicy,
	DownloadLimits as _DownloadLimits,
	DownloadObserver,
	DownloadTelemetry,
	HttpOpener,
	HttpResponse,
	build_manifest_url as _transfer_build_manifest_url,
	content_length as _transfer_content_length,
	content_range as _transfer_content_range,
	depot_manifest_url as _transfer_depot_manifest_url,
	download_partial_path as _transfer_download_partial_path,
	file_url as _transfer_file_url,
	normalize_server as _transfer_normalise_server,
	normalized_cpu_load as _transfer_normalized_cpu_load,
	ordered_download_servers as _transfer_ordered_download_servers,
	prune_transfer_samples as _transfer_prune_samples,
	request_headers as _transfer_request_headers,
	sample_rate as _transfer_sample_rate,
	server_candidates as _transfer_server_candidates,
)
from .game.depot import (
	DEFAULT_DEPOT_START_OPTION,
	CreateIfMissing,
	DepotDownloadMetrics,
	DepotError,
	DepotFile,
	DepotFileCatalog,
	DepotFileCatalogEntry,
	DepotIntegrityIssue,
	DepotLaunchCatalog,
	DepotPreflightResult,
	DepotSnapshot,
	DepotStartOption,
	DepotSyncResult,
	PlannedDownload,
	ProgressCallback,
	QuarantineResult,
	SyncProgress,
	UpdatePlan,
)
from .paths import AppPaths
from .telemetry import classify_network_failure, current_telemetry


# Keep historical imports available while implementation moves into the feature
# package. Referencing them here makes that compatibility surface explicit to
# static analysis without narrowing this module's existing wildcard behavior.
_COMPATIBILITY_EXPORTS = (
	DEFAULT_DEPOT_START_OPTION,
	DEPOT_FILE_CATALOG_SCHEMA,
	DEPOT_LAUNCH_CATALOG_SCHEMA,
	MAX_CREATE_IF_MISSING_BYTES,
	SHA1_PATTERN,
	BandwidthLimiter,
	DownloadTelemetry,
	HttpOpener,
	HttpResponse,
	DepotStartOption,
	DepotFileCatalogEntry,
	DepotIntegrityIssue,
	_depot_file_catalog_payload,
	_catalog_entry,
	_catalog_entry_from_cached,
	_catalog_identity_matches,
	_non_negative_int,
	_normalise_relative_path,
	_parse_create_if_missing,
	_parse_deleted_paths,
	_parse_depot_file_catalog_entry,
	_parse_start_options,
	_relative_parts,
	_start_option_text,
	_string,
	_ScannedGameEntry,
	_scan_directory,
	_scan_game_tree,
	launch_catalog_from_snapshot,
	select_depot_start_option,
)


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_LINE_ENDING_RECOVERY_BYTES = DEFAULT_MAX_LINE_ENDING_RECOVERY_BYTES
UNMANAGED_EXECUTABLE_SUFFIXES = DEFAULT_UNMANAGED_EXECUTABLE_SUFFIXES
ALWAYS_VERIFY_PATHS = frozenset(
	{"game.exe", "game.dll", "gamese.exe", "gamese.dll"}
)
CONTENT_RANGE_PATTERN = re.compile(r"^bytes\s+(\d+)-(\d+)/(\d+)$", re.IGNORECASE)
MAX_PREFLIGHT_ISSUES = 32


@dataclass(frozen=True)
class DownloadOptions(_DownloadLimits):
	"""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)


class AdaptiveConcurrencyController(_AdaptiveConcurrencyController):
	"""Compatibility facade configured from the historical module constants."""

	def __init__(
		self,
		options: DownloadOptions,
		*,
		clock: Callable[[], float] = time.monotonic,
		cpu_load: Callable[[], float | None] | None = None,
	) -> None:
		super().__init__(
			options,
			clock=clock,
			cpu_load=cpu_load or _normalized_cpu_load,
			initial_workers=DOWNLOAD_INITIAL_WORKERS,
			evaluation_interval_seconds=ADAPTIVE_EVALUATION_INTERVAL_SECONDS,
			retry_window_seconds=ADAPTIVE_RETRY_WINDOW_SECONDS,
			retry_recovery_seconds=ADAPTIVE_RETRY_RECOVERY_SECONDS,
			transfer_rate_window_seconds=TRANSFER_RATE_WINDOW_SECONDS,
		)


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

	_write_json(path, _launch_catalog_payload(snapshot))


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"))
		return _parse_launch_catalog(value)
	except (OSError, UnicodeError, json.JSONDecodeError, DepotError):
		return None


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

	catalog = _build_depot_file_catalog(
		game_directory,
		snapshot,
		verified_paths=verified_paths,
		previous=previous,
	)
	_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"))
		return _parse_depot_file_catalog(value)
	except (OSError, UnicodeError, json.JSONDecodeError, DepotError):
		return None


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
	result, updated_catalog = _inspect_managed_game(
		paths.source_game,
		catalog,
		hash_file=_sha1,
		max_issues=MAX_PREFLIGHT_ISSUES,
		clock=time.perf_counter,
	)
	if updated_catalog is not None:
		_write_json(
			paths.depot_file_catalog,
			_depot_file_catalog_payload(updated_catalog),
			compact=True,
		)
	return result


def _depot_http_policy() -> _DepotHttpPolicy:
	return _DepotHttpPolicy(
		configuration_url=lambda: CDN_CONFIGURATION_URL,
		server_candidates=lambda configuration, preferred_region: _server_candidates(
			configuration,
			preferred_region,
		),
		depot_manifest_url=lambda server: _depot_manifest_url(server),
		build_manifest_url=lambda server, build: _build_manifest_url(server, build),
		file_url=lambda server, file: _file_url(server, file),
		ordered_download_servers=lambda primary, candidates: _ordered_download_servers(
			primary,
			candidates,
		),
		download_partial_path=lambda destination: _download_partial_path(destination),
		recover_manifest_file=lambda temporary, destination, file: (
			_recover_crlf_manifest_file(temporary, destination, file)
		),
		parse_snapshot=lambda region, server, depot_manifest, build_manifest: (
			_parse_snapshot(region, server, depot_manifest, build_manifest)
		),
		positive_int=lambda value, field: _positive_int(value, field),
		content_length=lambda value: _content_length(value),
		content_range=lambda value: _content_range(value),
		headers=lambda: _headers(),
		read_size=lambda: READ_SIZE,
		hash_read_size=lambda: HASH_READ_SIZE,
		clock=lambda: time.monotonic(),
	)


class DepotClient(_DepotHttpClient):
	"""Stable facade for the publisher depot HTTP client."""

	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:
		super().__init__(
			policy=_depot_http_policy(),
			timeout_seconds=timeout_seconds,
			opener=opener,
			retry_delays=retry_delays,
			sleeper=sleeper,
			download_observer=download_observer,
		)


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

	return _plan_update(
		game_directory,
		snapshot,
		repair=repair,
		hash_file=_sha1,
		always_verify_paths=ALWAYS_VERIFY_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."""

	operation_started = time.perf_counter()
	operation = (
		"repair"
		if repair
		else "update"
		if paths.source_game.is_dir()
		else "install"
	)
	telemetry_enabled = not dry_run
	paths.create_base_directories()
	depot_client = client or DepotClient()
	previous_file_catalog = read_current_depot_file_catalog(paths)
	current_version = _completed_depot_version(paths.native_depot_state)
	manifest_started = time.perf_counter()
	if operation == "update":
		_send_depot_telemetry(
			"update_check_started",
			metadata={"current_version": current_version},
			context={"operation": operation, "stage": "manifest"},
			enabled=telemetry_enabled,
		)
	_emit_progress(progress, "manifest", 0, 0, 0, 0, None)
	try:
		snapshot = depot_client.fetch_snapshot()
	except BaseException as exc:
		_send_depot_telemetry(
			"download_failed",
			severity="error",
			message=str(exc),
			metadata={
				"resource_type": "manifest",
				"duration_ms": _elapsed_milliseconds(manifest_started),
				"failure_type": classify_network_failure(exc),
			},
			context={"operation": operation, "stage": "manifest"},
			enabled=telemetry_enabled,
		)
		raise
	manifest_duration_ms = _elapsed_milliseconds(manifest_started)
	scan_started = time.perf_counter()
	quarantine = QuarantineResult(0, None)
	try:
		if quarantine_unmanaged and not dry_run:
			quarantine = _quarantine_unmanaged_files(paths, snapshot)
		plan = plan_update(paths.source_game, snapshot, repair=repair)
	except BaseException as exc:
		_send_depot_telemetry(
			"file_validation_failed",
			severity="error",
			message=str(exc),
			metadata={
				"duration_ms": _elapsed_milliseconds(scan_started),
				"repair": repair,
			},
			context={"operation": operation, "stage": "scan"},
			enabled=telemetry_enabled,
		)
		raise
	scan_duration_ms = _elapsed_milliseconds(scan_started)
	_emit_progress(
		progress,
		"planned",
		0,
		len(plan.downloads),
		0,
		plan.bytes_total,
		None,
	)
	plan_metadata = {
		"current_version": current_version,
		"target_version": snapshot.version,
		"changed_file_count": len(plan.downloads) + len(plan.deletes),
		"files_checked": plan.files_total,
		"files_missing": sum(item.reason == "missing" for item in plan.downloads),
		"files_invalid": sum(item.reason != "missing" for item in plan.downloads),
		"files_replaced": len(plan.downloads),
		"download_bytes": plan.bytes_total,
		"manifest_duration_ms": manifest_duration_ms,
		"scan_duration_ms": scan_duration_ms,
	}
	if operation == "update":
		_send_depot_telemetry(
			"update_available" if plan.downloads or plan.deletes else "update_not_required",
			metadata=plan_metadata,
			context={"operation": operation, "stage": "scan"},
			enabled=telemetry_enabled,
		)
	elif operation == "repair":
		_send_depot_telemetry(
			"repair_scan_completed",
			metadata=plan_metadata,
			context={"operation": operation, "stage": "scan"},
			enabled=telemetry_enabled,
		)

	if dry_run:
		return _sync_result(
			snapshot,
			plan,
			0,
			0,
			0,
			repair,
			True,
			quarantine=quarantine,
			manifest_duration_ms=manifest_duration_ms,
			scan_duration_ms=scan_duration_ms,
			update_duration_ms=_elapsed_milliseconds(operation_started),
		)

	root = _prepare_game_root(paths.source_game, create=True)
	download_metrics = _download_plan(
		depot_client,
		snapshot,
		root,
		plan,
		progress,
		download_options or DownloadOptions(),
		operation=operation,
		telemetry_enabled=telemetry_enabled,
	)
	apply_started = time.perf_counter()
	if operation == "update":
		_send_depot_telemetry(
			"update_apply_started",
			metadata={
				"target_version": snapshot.version,
				"changed_file_count": len(plan.downloads) + len(plan.deletes),
			},
			context={"operation": operation, "stage": "apply"},
			enabled=telemetry_enabled,
		)
	try:
		deleted = _apply_deletions(root, plan.deletes)
		created = _apply_create_if_missing(root, snapshot.create_if_missing)
		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,
		)
		cache_depot_launch_catalog(paths.depot_launch_catalog_cache, snapshot)
		apply_duration_ms = _elapsed_milliseconds(apply_started)
		update_duration_ms = _elapsed_milliseconds(operation_started)
		result = _sync_result(
			snapshot,
			plan,
			len(plan.downloads),
			plan.bytes_total,
			deleted,
			repair,
			False,
			created=created,
			quarantine=quarantine,
			manifest_duration_ms=manifest_duration_ms,
			scan_duration_ms=scan_duration_ms,
			apply_duration_ms=apply_duration_ms,
			update_duration_ms=update_duration_ms,
			download_metrics=download_metrics,
		)
		_write_json(paths.native_depot_state, asdict(result))
	except BaseException as exc:
		_send_depot_telemetry(
			"filesystem_error",
			severity="error",
			message=str(exc),
			metadata={
				"duration_ms": _elapsed_milliseconds(apply_started),
				"target_version": snapshot.version,
			},
			context={"operation": operation, "stage": "apply"},
			enabled=telemetry_enabled,
		)
		raise
	if operation == "update":
		_send_depot_telemetry(
			"update_apply_completed",
			metadata={
				**plan_metadata,
				"apply_duration_ms": result.apply_duration_ms,
				"download_duration_ms": result.download_duration_ms,
				"update_duration_ms": result.update_duration_ms,
				"network_bytes_downloaded": result.network_bytes_downloaded,
				"average_download_mbps": result.download_average_mbps,
				"retry_count": result.download_retry_count,
				"resume_used": result.download_resume_used,
			},
			context={"operation": operation, "stage": "apply"},
			enabled=telemetry_enabled,
		)
	_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:
	return _filesystem_quarantine_unmanaged_files(
		paths.source_game,
		paths.game_import_quarantine,
		paths.data,
		snapshot,
		unmanaged_executable_suffixes=UNMANAGED_EXECUTABLE_SUFFIXES,
		prepare_root=lambda game_directory, *, create: _prepare_game_root(
			game_directory,
			create=create,
		),
		safe_user_file=lambda path, relative: _safe_imported_user_file(
			path,
			relative,
		),
		write_manifest=lambda path, value: _write_json(path, value),
		clock=lambda: datetime.now().astimezone(),
	)


def _safe_imported_user_file(path: Path, relative: str) -> bool:
	return _filesystem_safe_imported_user_file(
		path,
		relative,
		executable_suffixes=UNMANAGED_EXECUTABLE_SUFFIXES,
	)


def _download_plan(
	client: DepotClient,
	depot_snapshot: DepotSnapshot,
	root: Path,
	plan: UpdatePlan,
	progress: ProgressCallback | None,
	options: DownloadOptions,
	*,
	operation: str = "update",
	telemetry_enabled: bool = True,
) -> DepotDownloadMetrics:
	if not plan.downloads:
		return DepotDownloadMetrics(mirror_id=_server_display_name(depot_snapshot.server))

	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
	resume_offsets: dict[str, int] = {}
	mirrors: set[str] = set()
	last_http_status: int | None = None
	retry_events_sent = 0
	resume_events_sent = 0
	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: Callable[[DownloadTelemetry], None] | None = (
		client.download_observer
	)
	previous_limiter = client.bandwidth_limiter
	started_event, completed_event = _download_event_names(operation)
	_send_depot_telemetry(
		started_event,
		metadata={
			"resource_type": "repair_file" if operation == "repair" else "game",
			"download_bytes": plan.bytes_total,
			"file_count": len(plan.downloads),
			"mirror_id": _server_display_name(depot_snapshot.server),
			"cdn_region": depot_snapshot.region,
		},
		context={"operation": operation, "stage": "download"},
		enabled=telemetry_enabled,
	)

	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 last_http_status, resume_events_sent, retry_count
		nonlocal retry_events_sent, retry_reason, transferred_bytes
		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"
		emit_retry = False
		emit_resume = False
		if event.kind in {"request", "response", "retry", "resume", "chunk"}:
			with lock:
				current_servers[event.file] = _server_display_name(event.server)
				mirrors.add(_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 retry_events_sent < 5:
						retry_events_sent += 1
						emit_retry = True
				if event.kind == "resume" and event.resumed_bytes > 0:
					prior_offset = resume_offsets.get(event.file, 0)
					resume_offsets[event.file] = max(prior_offset, event.resumed_bytes)
					if prior_offset == 0 and resume_events_sent < 5:
						resume_events_sent += 1
						emit_resume = True
				if event.http_status is not None:
					last_http_status = event.http_status
		if previous_observer is not None:
			previous_observer(event)
		if emit_retry:
			_send_depot_telemetry(
				"download_retry",
				severity="warning",
				message=event.reason,
				metadata={
					"resource_type": (
						"repair_file" if operation == "repair" else "game"
					),
					"file": event.file,
					"attempt": event.attempt,
					"failure_type": classify_network_failure(event.reason),
					"http_status": event.http_status,
					"mirror_id": _server_display_name(event.server),
				},
				context={"operation": operation, "stage": "download_retry"},
				enabled=telemetry_enabled,
			)
		if emit_resume:
			_send_depot_telemetry(
				"download_resumed",
				metadata={
					"resource_type": (
						"repair_file" if operation == "repair" else "game"
					),
					"file": event.file,
					"resumed_bytes": event.resumed_bytes,
					"mirror_id": _server_display_name(event.server),
				},
				context={"operation": operation, "stage": "download_resume"},
				enabled=telemetry_enabled,
			)
		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,
		)

	def metrics_snapshot() -> DepotDownloadMetrics:
		with lock:
			duration_seconds = max(0.0, time.monotonic() - started_at)
			primary_mirror = _server_display_name(depot_snapshot.server)
			return DepotDownloadMetrics(
				duration_ms=round(duration_seconds * 1000),
				transferred_bytes=transferred_bytes,
				retry_count=retry_count,
				resume_used=bool(resume_offsets),
				resumed_bytes=sum(resume_offsets.values()),
				average_mbps=_average_megabits_per_second(
					transferred_bytes,
					duration_seconds,
				),
				mirror_id=primary_mirror,
				http_status=last_http_status,
			)

	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
	except BaseException as exc:
		metrics = metrics_snapshot()
		_send_depot_telemetry(
			"download_failed",
			severity="error",
			message=str(exc),
			metadata={
				"resource_type": "repair_file" if operation == "repair" else "game",
				"download_bytes": metrics.transferred_bytes,
				"content_bytes": plan.bytes_total,
				"duration_ms": metrics.duration_ms,
				"average_mbps": metrics.average_mbps,
				"retry_count": metrics.retry_count,
				"resume_used": metrics.resume_used,
				"resumed_bytes": metrics.resumed_bytes,
				"http_status": metrics.http_status,
				"failure_type": classify_network_failure(exc),
				"mirror_id": metrics.mirror_id,
				"mirrors_tried": sorted(mirrors),
			},
			context={"operation": operation, "stage": "download"},
			enabled=telemetry_enabled,
		)
		raise
	finally:
		client.download_observer = previous_observer
		client.bandwidth_limiter = previous_limiter

	metrics = metrics_snapshot()
	_send_depot_telemetry(
		completed_event,
		metadata={
			"resource_type": "repair_file" if operation == "repair" else "game",
			"download_bytes": metrics.transferred_bytes,
			"content_bytes": plan.bytes_total,
			"duration_ms": metrics.duration_ms,
			"average_mbps": metrics.average_mbps,
			"retry_count": metrics.retry_count,
			"resume_used": metrics.resume_used,
			"resumed_bytes": metrics.resumed_bytes,
			"http_status": metrics.http_status,
			"mirror_id": metrics.mirror_id,
			"mirrors_used": sorted(mirrors),
			"file_count": len(plan.downloads),
		},
		context={"operation": operation, "stage": "download"},
		enabled=telemetry_enabled,
	)
	return metrics


def _apply_deletions(root: Path, relative_paths: Iterable[str]) -> int:
	return _filesystem_apply_deletions(
		root,
		relative_paths,
		target_for=lambda game_root, relative: _safe_target(game_root, relative),
	)


def _apply_create_if_missing(
	root: Path,
	entries: Iterable[CreateIfMissing],
) -> int:
	return _filesystem_apply_create_if_missing(
		root,
		entries,
		target_for=lambda game_root, relative: _safe_target(game_root, relative),
		decode_entry=lambda entry: _decode_create_if_missing(entry),
		ensure_parent=lambda game_root, target: _ensure_target_parent(
			game_root,
			target,
		),
	)


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 _completed_depot_version(path: Path) -> str | 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
	version = value.get("version")
	return version[:32] if isinstance(version, str) and version else None


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


def _average_megabits_per_second(byte_count: int, duration_seconds: float) -> float | None:
	if byte_count <= 0 or duration_seconds <= 0:
		return None
	return round((byte_count * 8) / duration_seconds / 1_000_000, 3)


def _download_event_names(operation: str) -> tuple[str, str]:
	if operation == "install":
		return "install_download_started", "install_download_completed"
	if operation == "update":
		return "update_download_started", "update_download_completed"
	return "download_started", "download_completed"


def _send_depot_telemetry(
	event: str,
	*,
	severity: str = "info",
	message: str | None = None,
	metadata: Mapping[str, object] | None = None,
	context: Mapping[str, object] | None = None,
	enabled: bool = True,
) -> None:
	if not enabled:
		return
	telemetry = current_telemetry()
	if telemetry is None:
		return
	try:
		telemetry.send_event(
			event,
			severity=severity,
			message=message,
			metadata=metadata,
			context=context,
		)
	except Exception:
		# Telemetry must never alter verified depot behavior.
		pass


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),
	manifest_duration_ms: int = 0,
	scan_duration_ms: int = 0,
	apply_duration_ms: int = 0,
	update_duration_ms: int = 0,
	download_metrics: DepotDownloadMetrics | None = None,
) -> DepotSyncResult:
	metrics = download_metrics or DepotDownloadMetrics()
	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(),
		manifest_duration_ms=manifest_duration_ms,
		scan_duration_ms=scan_duration_ms,
		download_duration_ms=metrics.duration_ms,
		apply_duration_ms=apply_duration_ms,
		update_duration_ms=update_duration_ms,
		network_bytes_downloaded=metrics.transferred_bytes,
		download_retry_count=metrics.retry_count,
		download_resume_used=metrics.resume_used,
		download_average_mbps=metrics.average_mbps,
		mirror_id=metrics.mirror_id,
	)


def _server_candidates(
	configuration: Mapping[str, object] | None,
	preferred_region: str,
) -> tuple[tuple[str, str], ...]:
	return _transfer_server_candidates(
		configuration,
		preferred_region,
		fallback_servers=FALLBACK_SERVERS,
	)


def _normalise_server(value: str) -> str:
	return _transfer_normalise_server(value)


def _depot_manifest_url(server: str) -> str:
	return _transfer_depot_manifest_url(server, depot_id=DEPOT_ID)


def _build_manifest_url(server: str, build: int) -> str:
	return _transfer_build_manifest_url(server, build, depot_id=DEPOT_ID)


def _file_url(server: str, file: DepotFile) -> str:
	return _transfer_file_url(server, file, depot_id=DEPOT_ID)


def _ordered_download_servers(
	primary: str,
	candidates: Iterable[str],
) -> tuple[str, ...]:
	return _transfer_ordered_download_servers(primary, candidates)


def _download_partial_path(destination: Path) -> Path:
	return _transfer_download_partial_path(destination)


def _recover_crlf_manifest_file(
	temporary: Path,
	destination: Path,
	file: DepotFile,
) -> bool:
	return _filesystem_recover_crlf_manifest_file(
		temporary,
		destination,
		file,
		maximum_bytes=MAX_LINE_ENDING_RECOVERY_BYTES,
		process_id=lambda: os.getpid(),
	)


def _ensure_target_parent(root: Path, target: Path) -> None:
	_filesystem_ensure_target_parent(root, target)


def _sha1(path: Path) -> str:
	return _sha1_file(path, read_size=HASH_READ_SIZE)


def _content_length(value: str | None) -> int | None:
	return _transfer_content_length(value)


def _content_range(value: str | None) -> tuple[int, int, int]:
	return _transfer_content_range(value, pattern=CONTENT_RANGE_PATTERN)


def _normalized_cpu_load() -> float | None:
	return _transfer_normalized_cpu_load()


def _prune_transfer_samples(
	samples: deque[tuple[float, int]],
	now: float,
) -> None:
	_transfer_prune_samples(
		samples,
		now,
		window_seconds=TRANSFER_RATE_WINDOW_SECONDS,
	)


def _sample_rate(samples: deque[tuple[float, int]]) -> float | None:
	return _transfer_sample_rate(samples)


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 _transfer_request_headers(__version__)


def _write_json(path: Path, value: object, *, compact: bool = False) -> None:
	_filesystem_write_json(path, value, compact=compact)


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