from __future__ import annotations import asyncio import logging import shutil import time import uuid from collections.abc import Callable from datetime import datetime, timedelta, timezone from typing import Any from .config import Settings from .database import Database from .steam_api import SteamWorkshopClient, WorkshopDetails from .workshop import SteamCmdRunner, directory_size, parse_about_xml, utc_now logger = logging.getLogger("uvicorn.error") ProgressCallback = Callable[[dict[str, Any]], None] class InitialDownloadTimeout(RuntimeError): pass def _batches(values: list[str], size: int) -> list[list[str]]: return [values[index : index + size] for index in range(0, len(values), size)] class RefreshService: def __init__( self, settings: Settings, database: Database, steam: SteamWorkshopClient | None = None, runner: SteamCmdRunner | None = None, ): self.settings = settings self.database = database self.steam = steam or SteamWorkshopClient() self.runner = runner or SteamCmdRunner( settings.steamcmd_path, settings.workshop_content_dir, settings.steamcmd_timeout_seconds, ) def needs_initial_download(self) -> bool: rows = self.database.current_mods() if not rows: return True for row in rows: if row.get("status") == "skipped": continue if row.get("status") != "ready" or not ( self.settings.workshop_content_dir / row["workshop_id"] ).is_dir(): return True return False def _store_metadata( self, detail: WorkshopDetails, previous: dict[str, Any] | None, stale: bool, skip_reason: str | None = None, ) -> dict[str, Any]: invalid_reason = skip_reason if invalid_reason is None and detail.result != 1: invalid_reason = f"Steam returned result {detail.result}" elif invalid_reason is None and detail.consumer_app_id not in (0, 294100): invalid_reason = f"Workshop item belongs to app {detail.consumer_app_id}" return { "workshop_id": detail.workshop_id, "title": detail.title, "package_id": None, "steam_app_id": None, "display_name": None, "time_updated": detail.time_updated, "metadata_file_size": detail.file_size, "disk_size": None, "local_path": None, "status": "skipped" if invalid_reason else ("pending" if stale else (previous or {}).get("status", "pending")), "error": invalid_reason, "last_checked_at": utc_now(), "last_downloaded_at": None, } def _inspection_values( self, workshop_id: str, downloaded: bool, forced_error: str | None = None, ) -> tuple[bool, dict[str, Any]]: mod_path = self.settings.workshop_content_dir / workshop_id if forced_error: return False, { "workshop_id": workshop_id, "status": "failed", "error": forced_error, "disk_size": directory_size(mod_path), "local_path": str(mod_path) if mod_path.exists() else None, "last_checked_at": utc_now(), "last_downloaded_at": utc_now() if downloaded else None, } try: parsed = parse_about_xml(mod_path) return True, { "workshop_id": workshop_id, "package_id": parsed.package_id, "steam_app_id": parsed.declared_steam_app_id, "display_name": parsed.display_name, "disk_size": directory_size(mod_path), "local_path": str(mod_path), "status": "ready", "error": "", "last_checked_at": utc_now(), "last_downloaded_at": utc_now() if downloaded else None, } except (OSError, ValueError) as error: return False, { "workshop_id": workshop_id, "status": "failed", "error": str(error), "disk_size": directory_size(mod_path), "local_path": str(mod_path) if mod_path.exists() else None, "last_checked_at": utc_now(), "last_downloaded_at": utc_now() if downloaded else None, } def refresh( self, run_id: str, trigger: str, *, initial_download: bool = False, overall_timeout_seconds: int | None = None, progress: ProgressCallback | None = None, ) -> None: started_at = utc_now() started = time.monotonic() deadline = ( started + overall_timeout_seconds if initial_download and overall_timeout_seconds is not None else None ) self.database.create_run(run_id, trigger, started_at) metrics: dict[str, Any] = { "collection_items": 0, "unique_items": 0, "ready_items": 0, "downloads_requested": 0, "downloads_succeeded": 0, "downloads_failed": 0, "bytes_before": directory_size(self.settings.workshop_content_dir), "bytes_after": 0, "bytes_downloaded": 0, "cache_bytes": 0, "status": "failed", "error": None, } def report(**values: Any) -> None: if progress: progress(values) def remaining_time() -> float | None: if deadline is None: return None remaining = deadline - time.monotonic() if remaining <= 0: raise InitialDownloadTimeout( "The initial Workshop download reached its " f"{overall_timeout_seconds}-second limit" ) return remaining try: report(stage="collections", completed_mods=0, total_mods=0, percent=0.0) remaining_time() collection_ids = list(self.settings.collections.values()) response, collection_reference_ids = self.steam.get_collections(collection_ids) named_collections: dict[str, tuple[str, list[str]]] = {} all_ids: list[str] = [] for name, collection_id in self.settings.collections.items(): workshop_ids = response.get(collection_id) if workshop_ids is None: raise RuntimeError( f"Steam did not return collection {collection_id}" ) named_collections[name] = (collection_id, workshop_ids) all_ids.extend(workshop_ids) self.database.replace_collection_members(named_collections) unique_ids = sorted(set(all_ids), key=int) metrics["collection_items"] = len(all_ids) metrics["unique_items"] = len(unique_ids) details = self.steam.get_item_details(unique_ids) previous = self.database.mod_map(unique_ids) targets: list[str] = [] metadata_rows: list[dict[str, Any]] = [] for workshop_id in unique_ids: detail = details.get(workshop_id) if not detail: self.database.upsert_mod( { "workshop_id": workshop_id, "status": "skipped", "error": "Steam returned no item metadata", "last_checked_at": utc_now(), } ) continue old = previous.get(workshop_id) local_path = self.settings.workshop_content_dir / workshop_id stale = ( old is None or int(old.get("time_updated") or 0) != detail.time_updated or old.get("status") != "ready" or not local_path.is_dir() ) skip_reason = ( "Workshop entry is a collection reference" if workshop_id in collection_reference_ids else None ) metadata_rows.append( self._store_metadata(detail, old, stale, skip_reason=skip_reason) ) if ( detail.result == 1 and detail.consumer_app_id in (0, 294100) and workshop_id not in collection_reference_ids and stale ): targets.append(workshop_id) self.database.upsert_mods(metadata_rows) metrics["downloads_requested"] = len(targets) total_targets = len(targets) completed_targets = 0 current_after_metadata = self.database.mod_map(unique_ids) resolved_before = sum( 1 for row in current_after_metadata.values() if row.get("status") in {"ready", "skipped"} ) progress_total = len(unique_ids) if initial_download else total_targets progress_completed = resolved_before if initial_download else 0 title_by_id = { workshop_id: details[workshop_id].title for workshop_id in targets if workshop_id in details } report( stage="downloading" if targets else "inspecting", completed_mods=progress_completed, total_mods=progress_total, percent=round( (progress_completed / max(progress_total, 1)) * 100, 1 ) if progress_total else 100.0, ) if initial_download: logger.info( "Initial Workshop download started: %d mods need download, " "%d/%d collection entries already resolved, %.1f-hour hard limit", total_targets, resolved_before, len(unique_ids), (overall_timeout_seconds or 0) / 3600, ) if targets: remaining_bytes = sum( max(int(details[workshop_id].file_size or 0), 0) for workshop_id in targets if workshop_id in details ) workshop_root = self.settings.workshop_content_dir.parent.parent workshop_root.mkdir(parents=True, exist_ok=True) free_bytes = shutil.disk_usage(workshop_root).free required_bytes = int(remaining_bytes * 1.1) logger.info( "Workshop capacity check: %.2f GiB expected, %.2f GiB free", remaining_bytes / (1024**3), free_bytes / (1024**3), ) if required_bytes and free_bytes < required_bytes: raise RuntimeError( "Insufficient Workshop volume space: " f"{free_bytes / (1024**3):.2f} GiB free, " f"approximately {required_bytes / (1024**3):.2f} GiB required" ) if targets and not self.runner.available(): reason = ( f"SteamCMD was not found at '{self.settings.steamcmd_path}'. " "Set STEAMCMD_PATH to enable anonymous downloads." ) inspection_rows: list[dict[str, Any]] = [] for workshop_id in targets: mod_path = self.settings.workshop_content_dir / workshop_id success, values = self._inspection_values( workshop_id, downloaded=False, forced_error=None if mod_path.is_dir() else reason, ) inspection_rows.append(values) if not success: metrics["downloads_failed"] += 1 self.database.upsert_mods(inspection_rows) else: download_queue = list(targets) ready_target_ids: set[str] = set() batch_number = 0 pass_number = 1 while download_queue: # SteamCMD stops processing the rest of a command when one # Workshop item times out. Keep the first pass efficient, # then isolate every remaining item so one bad mod cannot # permanently block the items that followed it in a batch. pass_batch_size = ( self.settings.steamcmd_batch_size if pass_number == 1 else 1 ) batches = _batches(download_queue, pass_batch_size) if pass_number > 1: logger.info( "Initial Workshop download retry pass %d is " "isolating %d remaining mods individually", pass_number, len(download_queue), ) for pass_batch_number, batch in enumerate(batches, start=1): remaining = remaining_time() batch_number += 1 logger.info( "%s pass %d batch %d/%d starting (%d mods): %s", "Initial Workshop download" if initial_download else "Workshop refresh", pass_number, pass_batch_number, len(batches), len(batch), ", ".join( f"{title_by_id.get(workshop_id, 'Unknown mod')} " f"[{workshop_id}]" for workshop_id in batch ), ) result = self.runner.run_batch( batch, timeout_seconds=remaining ) if result.status != "completed": logger.warning( "SteamCMD pass %d batch %d/%d finished with " "status=%s exit_code=%s after %.1fs. Output:\n%s", pass_number, pass_batch_number, len(batches), result.status, result.exit_code, result.elapsed_seconds, result.output_tail or "(no SteamCMD output)", ) if result.has_io_failure: recover = getattr(self.runner, "recover_io_failure", None) moved = recover() if callable(recover) else [] if moved: logger.warning( "SteamCMD I/O recovery quarantined transient " "Workshop state: %s", ", ".join(str(path) for path in moved), ) self.database.add_download_attempt( { "run_id": run_id, "batch_number": batch_number, "item_count": len(batch), "workshop_ids": batch, "started_at": result.started_at, "finished_at": result.finished_at, "elapsed_seconds": result.elapsed_seconds, "exit_code": result.exit_code, "status": result.status, "output_tail": result.output_tail, } ) inspection_rows = [] for workshop_id in batch: failure = ( ( "SteamCMD did not complete this item; " "queued for retry" if initial_download else "SteamCMD reported a download failure" ) if workshop_id in result.failed_ids else None ) success, values = self._inspection_values( workshop_id, downloaded=True, forced_error=failure ) inspection_rows.append(values) if success: metrics["downloads_succeeded"] += 1 ready_target_ids.add(workshop_id) else: metrics["downloads_failed"] += 1 completed_targets = len(ready_target_ids) progress_completed = ( resolved_before + completed_targets if initial_download else completed_targets ) percent = round( (progress_completed / max(progress_total, 1)) * 100, 1, ) title = title_by_id.get(workshop_id, "Unknown mod") outcome = ( "ready" if success else ("retrying" if initial_download else "failed") ) report( stage="downloading", completed_mods=progress_completed, total_mods=progress_total, percent=percent, current_workshop_id=workshop_id, current_title=title, current_status=outcome, ) logger.info( "%s %d/%d (%.1f%%) %s: %s [%s]", "Initial mod download" if initial_download else "Mod refresh", progress_completed, progress_total, percent, outcome, title, workshop_id, ) self.database.upsert_mods(inspection_rows) remaining_time() if not initial_download: break download_queue = [ workshop_id for workshop_id in download_queue if workshop_id not in ready_target_ids ] if download_queue: remaining = remaining_time() retry_delay = min(30.0, remaining or 30.0) logger.warning( "Initial Workshop download pass %d left %d mods; " "retrying in %.0f seconds", pass_number, len(download_queue), retry_delay, ) time.sleep(retry_delay) remaining_time() pass_number += 1 # Re-inspect ready cached items whose metadata has never been parsed by this backend. current = self.database.mod_map(unique_ids) pending_rows = [] for workshop_id in unique_ids: row = current.get(workshop_id) if row and row.get("status") == "pending": _, values = self._inspection_values(workshop_id, downloaded=False) pending_rows.append(values) if pending_rows: self.database.upsert_mods(pending_rows) metrics["bytes_after"] = directory_size(self.settings.workshop_content_dir) metrics["cache_bytes"] = metrics["bytes_after"] metrics["bytes_downloaded"] = max( metrics["bytes_after"] - metrics["bytes_before"], 0 ) final_rows = self.database.mod_map(unique_ids) metrics["ready_items"] = sum( 1 for row in final_rows.values() if row.get("status") == "ready" ) resolved_items = sum( 1 for row in final_rows.values() if row.get("status") in {"ready", "skipped"} ) metrics["status"] = ( "completed" if resolved_items == metrics["unique_items"] else "partial" ) if metrics["status"] == "partial" and not metrics["error"]: metrics["error"] = ( f"{metrics['unique_items'] - resolved_items} items are not resolved" ) report( stage=metrics["status"], completed_mods=resolved_items, total_mods=metrics["unique_items"], percent=round( (resolved_items / max(metrics["unique_items"], 1)) * 100, 1 ) if metrics["unique_items"] else 100.0, ) except Exception as error: # noqa: BLE001 - the run must always persist a terminal state metrics["status"] = "failed" metrics["error"] = str(error) metrics["bytes_after"] = directory_size(self.settings.workshop_content_dir) metrics["cache_bytes"] = metrics["bytes_after"] metrics["bytes_downloaded"] = max( metrics["bytes_after"] - metrics["bytes_before"], 0 ) report(stage="failed", error=str(error)) logger.error( "%s failed: %s", "Initial Workshop download" if initial_download else "Workshop refresh", error, ) finally: if metrics["unique_items"]: metrics["ready_items"] = sum( 1 for row in self.database.current_mods() if row.get("status") == "ready" ) elapsed = max(time.monotonic() - started, 0.001) metrics["elapsed_seconds"] = round(elapsed, 3) metrics["average_mbps"] = round( (metrics["bytes_downloaded"] * 8 / 1_000_000) / elapsed, 3 ) metrics["finished_at"] = utc_now() self.database.finish_run(run_id, metrics) logger.info( "%s finished: status=%s, ready=%d/%d, elapsed=%.1fs, " "cache=%d bytes", "Initial Workshop download" if initial_download else "Workshop refresh", metrics["status"], metrics["ready_items"], metrics["unique_items"], elapsed, metrics["cache_bytes"], ) class RefreshCoordinator: def __init__( self, service: RefreshService, interval_seconds: int, refresh_on_startup: bool, initial_download_timeout_seconds: int = 43200, ): self.service = service self.interval_seconds = interval_seconds self.refresh_on_startup = refresh_on_startup self.initial_download_timeout_seconds = initial_download_timeout_seconds self.current_run_id: str | None = None self.next_refresh_at: str | None = None self._refresh_task: asyncio.Task[None] | None = None self._scheduler_task: asyncio.Task[None] | None = None self._stop = asyncio.Event() self._active_initial_download = False self._progress: dict[str, Any] | None = None async def start(self) -> None: if self._scheduler_task: return self._scheduler_task = asyncio.create_task(self._scheduler_loop()) async def stop(self) -> None: self._stop.set() if self._scheduler_task: self._scheduler_task.cancel() await asyncio.gather(self._scheduler_task, return_exceptions=True) async def trigger(self, trigger: str) -> tuple[str, bool]: if self._refresh_task and not self._refresh_task.done(): return self.current_run_id or "", False run_id = uuid.uuid4().hex self.current_run_id = run_id initial_download = self.service.needs_initial_download() self._active_initial_download = initial_download self._progress = { "stage": "starting", "completed_mods": 0, "total_mods": 0, "percent": 0.0, } def update_progress(values: dict[str, Any]) -> None: self._progress = {**(self._progress or {}), **values} async def run() -> None: try: await asyncio.to_thread( self.service.refresh, run_id, trigger, initial_download=initial_download, overall_timeout_seconds=( self.initial_download_timeout_seconds if initial_download else None ), progress=update_progress, ) finally: self.current_run_id = None self._active_initial_download = False self._refresh_task = asyncio.create_task(run()) return run_id, True async def _wait_for_initial_download(self, started: bool) -> None: if started and self._active_initial_download and self._refresh_task: self.next_refresh_at = None await asyncio.shield(self._refresh_task) async def _scheduler_loop(self) -> None: if self.refresh_on_startup: _, started = await self.trigger("startup") await self._wait_for_initial_download(started) while not self._stop.is_set(): next_time = datetime.now(timezone.utc) + timedelta( seconds=self.interval_seconds ) self.next_refresh_at = next_time.isoformat() try: await asyncio.wait_for(self._stop.wait(), timeout=self.interval_seconds) except asyncio.TimeoutError: _, started = await self.trigger("scheduled") await self._wait_for_initial_download(started) def state(self) -> dict[str, Any]: return { "running": bool(self._refresh_task and not self._refresh_task.done()), "current_run_id": self.current_run_id, "next_refresh_at": self.next_refresh_at, "interval_seconds": self.interval_seconds, "initial_download": bool( self._active_initial_download and self._refresh_task and not self._refresh_task.done() ), "initial_download_timeout_seconds": self.initial_download_timeout_seconds, "progress": dict(self._progress) if self._progress else None, }