furtka/furtka/cli.py

import argparse
import json
import sys
from pathlib import Path

from furtka import deps, dockerops, installer, reconciler
from furtka.paths import apps_dir
from furtka.scanner import scan


def _cmd_app_list(args: argparse.Namespace) -> int:
    results = scan(apps_dir())
    if args.json:
        out = [
            {
                "path": str(r.path),
                "name": r.manifest.name if r.manifest else None,
                "ok": r.ok,
                "error": r.error,
                "manifest": {
                    "name": r.manifest.name,
                    "display_name": r.manifest.display_name,
                    "version": r.manifest.version,
                    "description": r.manifest.description,
                    "description_long": r.manifest.description_long,
                    "volumes": list(r.manifest.volumes),
                    "ports": list(r.manifest.ports),
                    "icon": r.manifest.icon,
                    "open_url": r.manifest.open_url,
                    "settings": [
                        {
                            "name": s.name,
                            "label": s.label,
                            "description": s.description,
                            "type": s.type,
                            "required": s.required,
                            "default": s.default,
                        }
                        for s in r.manifest.settings
                    ],
                    "requires": [
                        {
                            "app": req.app,
                            "on_install": req.on_install,
                            "on_start": req.on_start,
                        }
                        for req in r.manifest.requires
                    ],
                }
                if r.manifest
                else None,
            }
            for r in results
        ]
        print(json.dumps(out, indent=2))
        return 0
    if not results:
        print("(no apps installed)")
        return 0
    for r in results:
        if r.ok:
            m = r.manifest
            print(f"{m.name:20s}  {m.version:10s}  {m.display_name}")
        else:
            print(f"{r.path.name:20s}  ERROR       {r.error}")
    return 0


def _cmd_app_install(args: argparse.Namespace) -> int:
    # If the user passed a path (or a path-ish thing), bypass dep resolution —
    # local paths are dev/test workflows where the caller knows what they want.
    # These just copy + reconcile; no `requires`/hook handling.
    src_path = Path(args.source)
    is_path = src_path.is_dir() or "/" in args.source or args.source.startswith(".")
    if is_path:
        try:
            src = installer.resolve_source(args.source)
            target = installer.install_from(src)
        except installer.InstallError as e:
            print(f"error: {e}", file=sys.stderr)
            return 2
        print(f"installed {target.name} to {target}")
        actions = reconciler.reconcile(apps_dir())
        for a in actions:
            print(f"  {a.describe()}")
        return 1 if reconciler.has_errors(actions) else 0

    # Catalog/bundled name install: resolve transitive `requires`, copy every
    # app folder (the synchronous phase the API runs inline), then drive the
    # exact same docker phase the API dispatches via systemd-run — so providers
    # come up before consumers and `on_install` hooks fire to provision them.
    # Going through run_install (not a bare reconcile) is what makes a CLI
    # install of a consumer actually provision against its provider.
    from furtka import install_runner

    try:
        plan = deps.plan_install(args.source)
    except deps.DependencyError as e:
        print(f"error: {e}", file=sys.stderr)
        return 2
    try:
        if plan.to_install:
            installer.install_plan(plan)
            to_install = list(plan.to_install)
        else:
            # Target already installed — reinstall to refresh files, then still
            # run the docker phase so hooks re-fire (matches API reinstall).
            installer.install_from(installer.resolve_source(args.source))
            to_install = [args.source]
    except installer.InstallError as e:
        print(f"error: {e}", file=sys.stderr)
        return 2

    # Stage the plan file run_install consumes (it removes it after reading).
    install_runner.plan_path().parent.mkdir(parents=True, exist_ok=True)
    install_runner.plan_path().write_text(
        json.dumps({"target": args.source, "to_install": to_install})
    )
    try:
        install_runner.run_install(args.source)
    except Exception as e:
        # run_install already wrote state="error"; surface it to the caller.
        print(f"error: {e}", file=sys.stderr)
        return 1
    for name in to_install:
        print(f"installed {name}")
    return 0


def _cmd_app_install_bg(args: argparse.Namespace) -> int:
    """Docker-facing phases of an install — called by the API via systemd-run.

    Internal subcommand; normal CLI users want `app install` (synchronous).
    This exists to separate the slow docker pull/up from the synchronous
    validation the API does inline, so the UI can poll a state file.
    """
    from furtka import install_runner

    try:
        install_runner.run_install(args.name)
    except Exception as e:
        # run_install already wrote state="error"; echo for journald.
        print(f"install-bg failed: {e}", file=sys.stderr)
        return 1
    return 0


def _cmd_app_remove(args: argparse.Namespace) -> int:
    target = apps_dir() / args.name
    if not target.exists():
        print(f"error: {args.name!r} is not installed", file=sys.stderr)
        return 1
    dependents = deps.dependents_of(args.name)
    if dependents:
        print(
            f"error: {args.name!r} is required by: {', '.join(dependents)}. Remove those first.",
            file=sys.stderr,
        )
        return 2
    try:
        dockerops.compose_down(target, args.name)
    except dockerops.DockerError as e:
        # Container may already be down (or never came up). Don't block removal.
        print(f"warning: compose down failed, removing folder anyway: {e}", file=sys.stderr)
    try:
        installer.remove(args.name)
    except installer.InstallError as e:
        print(f"error: {e}", file=sys.stderr)
        return 1
    print(f"removed {args.name} (volumes preserved)")
    return 0


def _cmd_serve(args: argparse.Namespace) -> int:
    # Imported lazily so `furtka` startup stays cheap when the user only runs
    # `app list` or `reconcile` (the common case during tests + scripts).
    from furtka import api

    api.serve(args.host, args.port)
    return 0


def _cmd_reconcile(args: argparse.Namespace) -> int:
    actions = reconciler.reconcile(apps_dir(), dry_run=args.dry_run)
    print(f"Scanned {apps_dir()}: {len(actions)} actions")
    for a in actions:
        print(f"  {a.describe()}")
    if args.dry_run:
        print("(dry-run — nothing changed)")
    # Exit non-zero on any per-app failure so systemd marks furtka-reconcile
    # red — but only AFTER all apps were attempted, so a broken app doesn't
    # hide healthy ones.
    return 1 if reconciler.has_errors(actions) else 0


def _cmd_update(args: argparse.Namespace) -> int:
    from furtka import updater

    if args.check:
        try:
            check = updater.check_update()
        except updater.UpdateError as e:
            print(f"error: {e}", file=sys.stderr)
            return 2
        if args.json:
            print(
                json.dumps(
                    {
                        "current": check.current,
                        "latest": check.latest,
                        "update_available": check.update_available,
                    },
                    indent=2,
                )
            )
        elif check.update_available:
            print(f"Update available: {check.current} → {check.latest}")
        else:
            print(f"Already up to date ({check.current})")
        return 0

    try:
        check = updater.run_update()
    except updater.UpdateError as e:
        print(f"error: {e}", file=sys.stderr)
        return 2
    if not check.update_available:
        print(f"Already up to date ({check.current})")
    else:
        print(f"Updated {check.current} → {check.latest}")
    return 0


def _cmd_rollback(args: argparse.Namespace) -> int:
    from furtka import updater

    try:
        restored = updater.rollback()
    except updater.UpdateError as e:
        print(f"error: {e}", file=sys.stderr)
        return 2
    print(f"Rolled back to {restored}")
    return 0


def _cmd_catalog_sync(args: argparse.Namespace) -> int:
    from furtka import catalog

    if args.check:
        try:
            check = catalog.check_catalog()
        except catalog.CatalogError as e:
            print(f"error: {e}", file=sys.stderr)
            return 2
        if args.json:
            print(
                json.dumps(
                    {
                        "current": check.current,
                        "latest": check.latest,
                        "update_available": check.update_available,
                    },
                    indent=2,
                )
            )
        elif check.update_available:
            print(f"Catalog update available: {check.current or '(none)'} → {check.latest}")
        else:
            print(f"Catalog already up to date ({check.current or check.latest})")
        return 0

    try:
        check = catalog.sync_catalog()
    except catalog.CatalogError as e:
        print(f"error: {e}", file=sys.stderr)
        return 2
    if not check.update_available:
        print(f"Catalog already up to date ({check.current or check.latest})")
    else:
        print(f"Synced catalog {check.current or '(none)'} → {check.latest}")
    return 0


def _cmd_catalog_status(args: argparse.Namespace) -> int:
    from furtka import catalog

    current = catalog.read_current_catalog_version()
    state = catalog.read_state()
    if args.json:
        print(json.dumps({"current": current, "state": state}, indent=2))
        return 0
    print(f"Catalog version: {current or '(none — run `furtka catalog sync`)'}")
    if state:
        print(f"Last sync stage: {state.get('stage', '?')} at {state.get('updated_at', '?')}")
    else:
        print("Last sync stage: (never)")
    return 0


def build_parser() -> argparse.ArgumentParser:
    p = argparse.ArgumentParser(prog="furtka", description="Furtka resource manager")
    sub = p.add_subparsers(dest="command", required=True)

    app = sub.add_parser("app", help="Manage installed apps")
    app_sub = app.add_subparsers(dest="subcommand", required=True)

    app_list = app_sub.add_parser("list", help="List installed apps")
    app_list.add_argument("--json", action="store_true", help="Emit JSON instead of a table")
    app_list.set_defaults(func=_cmd_app_list)

    app_install = app_sub.add_parser(
        "install",
        help="Install an app from a local folder or a bundled-app name",
    )
    app_install.add_argument(
        "source",
        help="Path to an app folder, or the name of a bundled app under /opt/furtka/apps/",
    )
    app_install.set_defaults(func=_cmd_app_install)

    # Internal — called by the HTTP API via systemd-run. Deliberately omitted
    # from the help listing; regular CLI users want `app install` above.
    app_install_bg = app_sub.add_parser(
        "install-bg",
        help=argparse.SUPPRESS,
    )
    app_install_bg.add_argument("name", help="Installed app folder name")
    app_install_bg.set_defaults(func=_cmd_app_install_bg)

    app_remove = app_sub.add_parser("remove", help="Stop and uninstall an app (keeps volumes)")
    app_remove.add_argument("name", help="App name (folder name under /var/lib/furtka/apps/)")
    app_remove.set_defaults(func=_cmd_app_remove)

    reconcile = sub.add_parser(
        "reconcile",
        help="Bring docker state in line with /var/lib/furtka/apps",
    )
    reconcile.add_argument(
        "--dry-run",
        action="store_true",
        help="Show what would be done without changing anything",
    )
    reconcile.set_defaults(func=_cmd_reconcile)

    serve = sub.add_parser("serve", help="Run the resource-manager HTTP API + UI")
    serve.add_argument("--host", default="127.0.0.1", help="Bind address (default 127.0.0.1)")
    serve.add_argument("--port", type=int, default=7000, help="Bind port (default 7000)")
    serve.set_defaults(func=_cmd_serve)

    update = sub.add_parser(
        "update",
        help="Check for or apply a Furtka release (Phase 2 self-update)",
    )
    update.add_argument(
        "--check",
        action="store_true",
        help="Only check whether an update is available; don't apply",
    )
    update.add_argument(
        "--json",
        action="store_true",
        help="Emit machine-readable JSON (only honoured with --check)",
    )
    update.set_defaults(func=_cmd_update)

    rollback = sub.add_parser(
        "rollback",
        help="Flip /opt/furtka/current back to the previous version slot",
    )
    rollback.set_defaults(func=_cmd_rollback)

    catalog = sub.add_parser("catalog", help="Manage the apps catalog (daniel/furtka-apps)")
    catalog_sub = catalog.add_subparsers(dest="subcommand", required=True)

    catalog_sync = catalog_sub.add_parser(
        "sync",
        help="Download and install the latest apps catalog from Forgejo",
    )
    catalog_sync.add_argument(
        "--check",
        action="store_true",
        help="Only check whether a catalog update is available; don't apply",
    )
    catalog_sync.add_argument(
        "--json",
        action="store_true",
        help="Emit machine-readable JSON (only honoured with --check)",
    )
    catalog_sync.set_defaults(func=_cmd_catalog_sync)

    catalog_status = catalog_sub.add_parser(
        "status",
        help="Print the currently-installed catalog version and last-sync stage",
    )
    catalog_status.add_argument(
        "--json",
        action="store_true",
        help="Emit machine-readable JSON",
    )
    catalog_status.set_defaults(func=_cmd_catalog_status)

    return p


def main(argv: list[str] | None = None) -> int:
    args = build_parser().parse_args(argv)
    return args.func(args)


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