non-interactive update check feature

Author: Redbot

src/redfetch/config_firstrun.py

@@ -147,6 +147,17 @@ def is_first_run(default_config_dir):
     return not os.path.exists(first_run_flag)
 
 
+def is_configured(default_config_dir: str | None = None) -> bool:
+    """Can initialize_config() proceed without interactive prompts?"""
+    default_config_dir = default_config_dir or user_config_dir("redfetch", "RedGuides")
+    try:
+        with open(os.path.join(default_config_dir, "first_run_complete")) as f:
+            config_dir = f.read().strip()
+    except OSError:
+        return False
+    return os.path.exists(os.path.join(config_dir, ".env"))
+
+
 def get_or_create_table(doc: TOMLDocument, table_path: str):
     """Navigate or create nested tables in the TOML document."""
     current_section = doc

src/redfetch/main.py

@@ -3,7 +3,7 @@
 import os
 from enum import Enum
 from pathlib import Path
-from typing import Optional
+from typing import NoReturn, Optional
 import asyncio
 
 # third-party imports
@@ -39,6 +39,12 @@ class Env(str, Enum):
     EMU = "EMU"
 
 
+EXIT_CHECK_ERROR = 1
+EXIT_CALLER_UPDATE = 2
+EXIT_AUTH_REQUIRED = 3
+EXIT_NOT_CONFIGURED = 4
+
+
 def parse_resource_id_or_fail(value: str) -> str:
     """Accept either an integer ID or a URL that includes a recognizable ID."""
     value_stripped = value.strip()
@@ -259,6 +265,83 @@ def download(
     asyncio.run(download_command_async(db_name=db_name, db_path=db_path, id_or_url=id_or_url, force=force))
 
 
+def _check_exit(real_stdout, exit_code: int, stdout_line: str = "") -> NoReturn:
+    """Write one line to real stdout and exit (never returns)."""
+    if stdout_line:
+        real_stdout.write(stdout_line + "\n")
+    real_stdout.flush()
+    raise typer.Exit(exit_code)
+
+
+def _has_auth_credentials() -> bool:
+    """Peek at env / keyring for stored credentials (no network, no init)."""
+    if os.environ.get("REDGUIDES_API_KEY"):
+        return True
+    try:
+        import keyring
+        token = keyring.get_password(auth.KEYRING_SERVICE_NAME, "access_token")
+        return token is not None
+    except Exception:
+        return False
+
+
+@app.command(
+    "check",
+    help="Machine-readable check for available updates. Stdout: update count. Exit code: 0=ok, 2=caller needs update, 3=auth required, 4=not configured.",
+    rich_help_panel="📦 Resource Management",
+)
+def check_command(
+    caller_resource_id: str | None = typer.Option(
+        None, "--caller-resource-id",
+        help="Resource ID of the calling program (e.g. 1974 for Very Vanilla MQ Live).",
+    ),
+):
+    real_stdout = sys.stdout
+    sys.stdout = sys.stderr
+
+    try:
+        # Phase 1 — pre-flight (no init, instant)
+        from redfetch.config_firstrun import is_configured
+
+        if not is_configured():
+            _check_exit(real_stdout, EXIT_NOT_CONFIGURED)
+
+        if not _has_auth_credentials():
+            _check_exit(real_stdout, EXIT_AUTH_REQUIRED)
+
+        # Phase 2 — init + check
+        config.initialize_config()
+        auth.initialize_keyring()
+        db_name = f"{config.settings.ENV}_resources.db"
+        store.initialize_db(db_name)
+        db_path = store.get_db_path(db_name)
+
+        result = asyncio.run(_check_command_async(db_path, caller_resource_id))
+
+        sys.stdout = real_stdout
+
+        if result is None:
+            _check_exit(real_stdout, EXIT_AUTH_REQUIRED)
+
+        exit_code = EXIT_CALLER_UPDATE if result.caller_update_available is True else 0
+        _check_exit(real_stdout, exit_code, str(result.updates_available))
+
+    except typer.Exit:
+        raise
+    except Exception:
+        sys.stdout = real_stdout
+        _check_exit(real_stdout, EXIT_CHECK_ERROR)
+
+
+async def _check_command_async(db_path: str, caller_resource_id: str | None):
+    try:
+        headers = await auth.get_api_headers()
+    except RuntimeError:
+        return None
+    from redfetch.update_check import check_for_updates
+    return await check_for_updates(db_path, headers, caller_resource_id)
+
+
 @app.command(
     "ui",
     help="Launch the [italic]Terminal User Interface[/italic].",

src/redfetch/sync_types.py

@@ -336,3 +336,11 @@ def has_errors(self) -> bool:
         return self.was_cancelled or any(
             item.outcome == "error" for item in self.items.values()
         )
+
+
+class UpdateCheckResult(SyncModel):
+    """Result of a lightweight update check (no downloads)."""
+
+    updates_available: int = 0
+    caller_update_available: bool | None = None
+    caller_resource_id: str | None = None

src/redfetch/update_check.py

@@ -0,0 +1,69 @@
+"""Lightweight non-interactive update check, no downloads"""
+
+from __future__ import annotations
+
+import httpx
+
+from redfetch import net
+from redfetch import store
+from redfetch.sync_discovery import payload_version_id
+from redfetch.sync_types import LocalSnapshot, UpdateCheckResult
+
+
+def _count_outdated(
+    local_snapshot: LocalSnapshot,
+    manifest: dict,
+    caller_resource_id: str | None = None,
+) -> UpdateCheckResult:
+    manifest_resources = manifest.get("resources") or {}
+
+    updates_available = 0
+    caller_found = False
+    caller_outdated = False
+
+    for local_state in local_snapshot.install_targets.values():
+        manifest_entry = manifest_resources.get(local_state.resource_id)
+        remote_version = payload_version_id(manifest_entry)
+        if local_state.version_local is None or remote_version is None:
+            continue
+        is_caller_root = (
+            caller_resource_id
+            and local_state.resource_id == caller_resource_id
+            and local_state.target_kind == "root"
+        )
+        if local_state.version_local != remote_version:
+            updates_available += 1
+            if is_caller_root:
+                caller_outdated = True
+        if is_caller_root:
+            caller_found = True
+
+    if caller_resource_id and caller_found:
+        caller_update_available = caller_outdated
+    else:
+        caller_update_available = None
+
+    return UpdateCheckResult(
+        updates_available=updates_available,
+        caller_update_available=caller_update_available,
+        caller_resource_id=caller_resource_id,
+    )
+
+
+async def check_for_updates(
+    db_path: str,
+    headers: dict,
+    caller_resource_id: str | None = None,
+) -> UpdateCheckResult:
+    """Compare local DB versions against the manifest"""
+    local_snapshot = await store.load_local_snapshot(db_path)
+
+    async with httpx.AsyncClient(
+        headers=headers,
+        http2=True,
+        timeout=30.0,
+        limits=httpx.Limits(max_connections=10, max_keepalive_connections=5),
+    ) as client:
+        manifest = await net.fetch_manifest_cached(client)
+
+    return _count_outdated(local_snapshot, manifest, caller_resource_id)

tests/test_check.py

@@ -0,0 +1,61 @@
+"""Tests the non-interactive update check logic."""
+import pytest
+
+from redfetch.config_firstrun import is_configured
+from redfetch.sync_types import LocalInstallState, LocalSnapshot
+from redfetch.update_check import _count_outdated
+
+
+@pytest.fixture
+def make_state():
+    def _make(
+        resource_id,
+        *,
+        version_local=10,
+        target_kind="root",
+        parent_id=None,
+        root_resource_id=None,
+    ):
+        if target_kind == "root":
+            target_key = f"/{resource_id}/"
+        else:
+            target_key = f"/{parent_id}/{resource_id}/"
+        return LocalInstallState(
+            target_key=target_key,
+            resource_id=resource_id,
+            parent_id=parent_id,
+            parent_target_key=f"/{parent_id}/" if parent_id else None,
+            root_resource_id=root_resource_id or resource_id,
+            target_kind=target_kind,
+            version_local=version_local,
+        )
+    return _make
+
+
+def test_counts_outdated_and_skips_none_version(make_state):
+    states = [
+        make_state("100", version_local=10),
+        make_state("200", version_local=20),
+        make_state("300", version_local=None),
+    ]
+    snapshot = LocalSnapshot(install_targets={s.target_key: s for s in states})
+    manifest = {"resources": {"100": {"version_id": 11}, "200": {"version_id": 20}, "300": {"version_id": 99}}}
+    result = _count_outdated(snapshot, manifest)
+    assert result.updates_available == 1
+
+
+def test_caller_as_dependency_not_tracked(make_state):
+    dep = make_state("1974", version_local=5, target_kind="dependency",
+                     parent_id="100", root_resource_id="100")
+    snapshot = LocalSnapshot(install_targets={dep.target_key: dep})
+    manifest = {"resources": {"1974": {"version_id": 10}}}
+    result = _count_outdated(snapshot, manifest, caller_resource_id="1974")
+    assert result.updates_available == 1
+    assert result.caller_update_available is None
+
+
+def test_is_configured_false_when_flag_but_no_env(tmp_path):
+    config_dir = tmp_path / "config"
+    config_dir.mkdir()
+    (tmp_path / "first_run_complete").write_text(str(config_dir))
+    assert is_configured(str(tmp_path)) is False