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 return any( row.get("status") != "ready" or not (self.settings.workshop_content_dir / row["workshop_id"]).is_dir() for row in rows ) def _store_metadata( self, detail: WorkshopDetails, previous: dict[str, Any] | None, stale: bool, ) -> dict[str, Any]: invalid_reason = None if detail.result != 1: invalid_reason = f"Steam returned result {detail.result}" elif 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": "failed" 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 = 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": "failed", "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() ) metadata_rows.append(self._store_metadata(detail, old, stale)) if ( detail.result == 1 and detail.consumer_app_id in (0, 294100) and stale ): targets.append(workshop_id) self.database.upsert_mods(metadata_rows) metrics["downloads_requested"] = len(targets) total_targets = len(targets) completed_targets = 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=0, total_mods=total_targets, percent=100.0 if not targets else 0.0, ) if initial_download: logger.info( "Initial Workshop download started: %d mods, %.1f-hour hard limit", total_targets, (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: batches = _batches( download_queue, self.settings.steamcmd_batch_size ) 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 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) percent = round( (completed_targets / max(total_targets, 1)) * 100, 1, ) title = title_by_id.get(workshop_id, "Unknown mod") outcome = "ready" if success else "failed" report( stage="downloading", completed_mods=completed_targets, total_mods=total_targets, 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", completed_targets, total_targets, 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" ) metrics["status"] = ( "completed" if metrics["ready_items"] == metrics["unique_items"] else "partial" ) if metrics["status"] == "partial" and not metrics["error"]: metrics["error"] = ( f"{metrics['unique_items'] - metrics['ready_items']} items are not ready" ) report( stage=metrics["status"], completed_mods=completed_targets, total_mods=total_targets, percent=round( (completed_targets / max(total_targets, 1)) * 100, 1 ) if total_targets 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, }