archy/scripts/check-app-catalog-drift.py

#!/usr/bin/env python3
"""Report drift between app-catalog/catalog.json and apps/*/manifest.yml."""

from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path
from typing import Any

import yaml

INTERNAL_MANIFEST_IDS = {
    "aiui",
    "archy-btcpay-db",
    "archy-mempool-db",
    "archy-mempool-web",
    "archy-nbxplorer",
    "bitcoin-ui",
    "core-lightning",
    "did-wallet",
    "electrs-ui",
    "lightning-stack",
    "lnd-ui",
    "mempool-api",
    "morphos-server",
    "router",
    "strfry",
    "web5-dwn",
}

LEGACY_STACK_CATALOG_IDS = {
    "immich",
    "netbird",
    "saleor",
    "tailscale",
}


def load_catalog(path: Path) -> dict[str, dict[str, Any]]:
    with path.open("r", encoding="utf-8") as fh:
        data = json.load(fh)
    apps = data.get("apps", [])
    if not isinstance(apps, list):
        raise ValueError(f"{path}: expected .apps to be a list")
    return {str(app.get("id", "")): app for app in apps if isinstance(app, dict) and app.get("id")}


def load_manifests(apps_dir: Path) -> dict[str, dict[str, Any]]:
    manifests: dict[str, dict[str, Any]] = {}
    for path in sorted(apps_dir.glob("*/manifest.yml")):
        with path.open("r", encoding="utf-8") as fh:
            data = yaml.safe_load(fh)
        if not isinstance(data, dict) or not isinstance(data.get("app"), dict):
            continue
        app = data["app"]
        app_id = app.get("id")
        if app_id:
            manifests[str(app_id)] = {"path": str(path), "app": app}
    return manifests


def metadata(app: dict[str, Any]) -> dict[str, Any]:
    value = app.get("metadata")
    return value if isinstance(value, dict) else {}


def manifest_value(app: dict[str, Any], field: str) -> Any:
    meta = metadata(app)
    container = app.get("container") if isinstance(app.get("container"), dict) else {}
    match field:
        case "title":
            return app.get("name")
        case "version":
            return str(app.get("version", ""))
        case "description":
            return app.get("description")
        case "dockerImage":
            return container.get("image")
        case "category":
            return app.get("category") or meta.get("category")
        case "tier":
            return meta.get("tier")
        case "icon":
            return meta.get("icon")
        case "repoUrl":
            return meta.get("repo") or meta.get("repoUrl")
        case _:
            return None


def normalize(value: Any) -> str:
    if value is None:
        return ""
    return str(value).strip()


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--catalog", default="app-catalog/catalog.json")
    parser.add_argument("--apps-dir", default="apps")
    parser.add_argument(
        "--strict",
        action="store_true",
        help="exit non-zero when missing entries or metadata drift are found",
    )
    parser.add_argument(
        "--release",
        action="store_true",
        help="suppress known internal/legacy-stack entries so output is release-actionable",
    )
    args = parser.parse_args()

    catalog = load_catalog(Path(args.catalog))
    manifests = load_manifests(Path(args.apps_dir))

    catalog_ids = set(catalog)
    manifest_ids = set(manifests)
    missing_manifests = sorted(catalog_ids - manifest_ids)
    missing_catalog = sorted(manifest_ids - catalog_ids)
    if args.release:
        missing_manifests = [app_id for app_id in missing_manifests if app_id not in LEGACY_STACK_CATALOG_IDS]
        missing_catalog = [app_id for app_id in missing_catalog if app_id not in INTERNAL_MANIFEST_IDS]

    compared_fields = [
        "title",
        "version",
        "description",
        "dockerImage",
        "category",
        "tier",
        "icon",
        "repoUrl",
    ]
    drift: list[str] = []
    for app_id in sorted(catalog_ids & manifest_ids):
        catalog_app = catalog[app_id]
        manifest_app = manifests[app_id]["app"]
        for field in compared_fields:
            catalog_val = normalize(catalog_app.get(field))
            manifest_val = normalize(manifest_value(manifest_app, field))
            if catalog_val and manifest_val and catalog_val != manifest_val:
                drift.append(f"{app_id}: {field}: catalog={catalog_val!r} manifest={manifest_val!r}")

    print(
        json.dumps(
            {
                "catalog_apps": len(catalog),
                "manifest_apps": len(manifests),
                "missing_manifests": len(missing_manifests),
                "missing_catalog": len(missing_catalog),
                "metadata_drift": len(drift),
            },
            sort_keys=True,
        )
    )

    for app_id in missing_manifests:
        print(f"MISSING_MANIFEST {app_id}")
    for app_id in missing_catalog:
        print(f"MISSING_CATALOG {app_id}")
    for item in drift:
        print(f"DRIFT {item}")

    if args.strict and (missing_manifests or missing_catalog or drift):
        return 1
    return 0


if __name__ == "__main__":
    sys.exit(main())