grm/src/gitea_runner_manager/runner_manager.py

"""Core logic for managing Gitea runners."""

from __future__ import annotations

from pathlib import Path

from .exceptions import AnsibleError
from .executor import AnsibleExecutor
from .i18n import _
from .registry import RunnerRegistry
from .report import track_steps
from .ui import say


class RunnerManager:
    """Orchestrates runner installation, updates, and lifecycle."""

    def __init__(
        self,
        executor: AnsibleExecutor | None = None,
        registry: RunnerRegistry | None = None,
    ) -> None:
        self._executor = executor or AnsibleExecutor()
        self._registry = registry or RunnerRegistry()

    def install(
        self,
        host: str,
        user: str,
        key: str | None = None,
        name: str | None = None,
        token: str | None = None,
        gitea_url: str = "",
        admin_token: str | None = None,
        integration_retries: int = 3,
        ask_become_pass: bool = False,
        labels: str | None = None,
    ) -> None:
        """Install a runner on a remote host using Ansible."""
        if not name:
            name = host
        if not gitea_url:
            raise AnsibleError(_("GITEA_URL must be set (or pass --url)"))
        if not token:
            raise AnsibleError(_("GITEA_REGISTRATION_TOKEN must be set (or pass --token)"))

        extra_vars = (
            f"registration_token={token} runner_name={name} gitea_url={gitea_url}"
            f" gitea_runner_integration_retries={integration_retries}"
        )
        if admin_token:
            extra_vars += f" gitea_admin_token={admin_token}"
        if labels:
            extra_vars += f" runner_labels={labels}"

        with track_steps() as tracker:
            tracker.begin(_("Installing Gitea Runner on {host}", host=host))
            cmd = self._build_cmd("install-runner.yml", host, user, extra_vars, key, ask_become_pass)
            self._executor.run(cmd, description=_("Installing Gitea Runner on {host}", host=host))
            tracker.done()

            tracker.begin(_("Save runner '{name}' to local registry", name=name))
            self._registry.add(
                name=name,
                host=host,
                user=user,
                key=key,
                gitea_url=gitea_url,
                labels=labels,
            )
            tracker.done()

    def update(
        self,
        host: str,
        user: str,
        key: str | None = None,
        version: str | None = None,
        ask_become_pass: bool = False,
    ) -> None:
        """Update the gitea_runner binary on a remote host."""
        extra_vars = ""
        if version:
            extra_vars += f" gitea_runner_version={version}"

        with track_steps() as tracker:
            tracker.begin(_("Updating Gitea Runner on {host}", host=host))
            cmd = self._build_cmd("update-runner.yml", host, user, extra_vars, key, ask_become_pass)
            self._executor.run(cmd, description=_("Updating Gitea Runner on {host}", host=host))
            tracker.done()

    def _resolve_runner(
        self,
        name: str,
        host: str | None = None,
        user: str | None = None,
        key: str | None = None,
    ) -> tuple[str, str, str | None, str]:
        """Look up runner metadata from registry, applying CLI overrides.

        Returns ``(host, user, key, gitea_url)`` where *gitea_url* is
        taken from the registry when available, allowing ``disable`` and
        ``remove`` to reuse the value stored at install time.
        """
        info = self._registry.get(name)
        actual_gitea_url = info.get("gitea_url", "") if info else ""
        if host and user:
            # Explicit connection details — bypass registry
            actual_host = host
            actual_user = user
            actual_key = key
        elif info:
            actual_host = host or info["host"]
            actual_user = user or info["user"]
            actual_key = key if key is not None else info.get("key")
        else:
            raise AnsibleError(
                _(
                    "Runner '{name}' not found in registry. Use 'grm install' first or provide --host and --user.",
                    name=name,
                )
            )
        return actual_host, actual_user, actual_key, actual_gitea_url

    def start(
        self,
        name: str,
        host: str | None = None,
        user: str | None = None,
        key: str | None = None,
        ask_become_pass: bool = False,
    ) -> None:
        """Start a runner instance on a remote host."""
        actual_host, actual_user, actual_key, _gitea_url = self._resolve_runner(name, host, user, key)
        with track_steps() as tracker:
            tracker.begin(_("Starting Gitea Runner {name} on {host}", name=name, host=actual_host))
            extra_vars = f"runner_name={name}"
            cmd = self._build_cmd("start-runner.yml", actual_host, actual_user, extra_vars, actual_key, ask_become_pass)
            self._executor.run(
                cmd, description=_("Starting Gitea Runner {name} on {host}", name=name, host=actual_host)
            )
            tracker.done()

    def stop(
        self,
        name: str,
        host: str | None = None,
        user: str | None = None,
        key: str | None = None,
        ask_become_pass: bool = False,
    ) -> None:
        """Stop a runner instance on a remote host."""
        actual_host, actual_user, actual_key, _gitea_url = self._resolve_runner(name, host, user, key)
        with track_steps() as tracker:
            tracker.begin(_("Stopping Gitea Runner {name} on {host}", name=name, host=actual_host))
            extra_vars = f"runner_name={name}"
            cmd = self._build_cmd("stop-runner.yml", actual_host, actual_user, extra_vars, actual_key, ask_become_pass)
            self._executor.run(
                cmd, description=_("Stopping Gitea Runner {name} on {host}", name=name, host=actual_host)
            )
            tracker.done()

    def enable(
        self,
        name: str,
        host: str | None = None,
        user: str | None = None,
        key: str | None = None,
        ask_become_pass: bool = False,
    ) -> None:
        """Enable a runner instance to start on boot."""
        actual_host, actual_user, actual_key, _gitea_url = self._resolve_runner(name, host, user, key)
        with track_steps() as tracker:
            tracker.begin(_("Enabling Gitea Runner {name} on {host}", name=name, host=actual_host))
            extra_vars = f"runner_name={name}"
            cmd = self._build_cmd(
                "enable-runner.yml", actual_host, actual_user, extra_vars, actual_key, ask_become_pass
            )
            self._executor.run(
                cmd, description=_("Enabling Gitea Runner {name} on {host}", name=name, host=actual_host)
            )
            tracker.done()

    def disable(
        self,
        name: str,
        host: str | None = None,
        user: str | None = None,
        key: str | None = None,
        token: str | None = None,
        gitea_url: str = "",
        ask_become_pass: bool = False,
    ) -> None:
        """Disable and deregister a runner instance."""
        actual_host, actual_user, actual_key, registry_gitea_url = self._resolve_runner(name, host, user, key)
        resolved_gitea_url = gitea_url or registry_gitea_url
        if not resolved_gitea_url:
            raise AnsibleError(_("GITEA_URL must be set (or pass --url)"))
        if not token:
            raise AnsibleError(_("GITEA_REGISTRATION_TOKEN must be set (or pass --token)"))
        with track_steps() as tracker:
            tracker.begin(_("Disabling Gitea Runner {name} on {host}", name=name, host=actual_host))
            extra_vars = f"runner_name={name} registration_token={token} gitea_url={resolved_gitea_url}"
            cmd = self._build_cmd(
                "disable-runner.yml", actual_host, actual_user, extra_vars, actual_key, ask_become_pass
            )
            self._executor.run(
                cmd, description=_("Disabling Gitea Runner {name} on {host}", name=name, host=actual_host)
            )
            tracker.done()

    def status(
        self,
        name: str,
        host: str | None = None,
        user: str | None = None,
        key: str | None = None,
        ask_become_pass: bool = False,
    ) -> None:
        """Check the status of a runner instance."""
        actual_host, actual_user, actual_key, _gitea_url = self._resolve_runner(name, host, user, key)
        with track_steps() as tracker:
            tracker.begin(_("Checking status of Gitea Runner {name} on {host}", name=name, host=actual_host))
            extra_vars = f"runner_name={name}"
            cmd = self._build_cmd(
                "status-runner.yml", actual_host, actual_user, extra_vars, actual_key, ask_become_pass
            )
            self._executor.run(
                cmd, description=_("Checking status of Gitea Runner {name} on {host}", name=name, host=actual_host)
            )
            tracker.done()

    def remove(
        self,
        name: str,
        host: str | None = None,
        user: str | None = None,
        key: str | None = None,
        token: str | None = None,
        gitea_url: str = "",
        ask_become_pass: bool = False,
        force: bool = False,
    ) -> None:
        """Remove a runner instance completely.

        When *force* is ``True``, skip the remote Ansible playbook and
        only remove the local registry entry. Use this when the remote
        host is already gone or unreachable.
        """
        actual_host, actual_user, actual_key, registry_gitea_url = self._resolve_runner(name, host, user, key)
        resolved_gitea_url = gitea_url or registry_gitea_url
        resolved_token = token
        if not force:
            if not resolved_gitea_url:
                raise AnsibleError(_("GITEA_URL must be set (or pass --url)"))
            if not resolved_token:
                raise AnsibleError(_("GITEA_REGISTRATION_TOKEN must be set (or pass --token)"))
        with track_steps() as tracker:
            tracker.begin(_("Removing Gitea Runner {name} from {host}", name=name, host=actual_host))
            if not force:
                extra_vars = f"runner_name={name} registration_token={resolved_token} gitea_url={resolved_gitea_url}"
                cmd = self._build_cmd(
                    "remove-runner.yml", actual_host, actual_user, extra_vars, actual_key, ask_become_pass
                )
                self._executor.run(
                    cmd, description=_("Removing Gitea Runner {name} from {host}", name=name, host=actual_host)
                )
            tracker.done()

            tracker.begin(_("Remove runner '{name}' from local registry", name=name))
            self._registry.remove(name)
            tracker.done()

    def list_runners(self) -> list[dict[str, str]]:
        """Return a list of registered runners with live service status."""
        runners = self._registry.list()
        result: list[dict[str, str]] = []
        for name, info in runners.items():
            host = info["host"]
            user = info["user"]
            key = info.get("key")
            say(
                _(
                    "Checking status of Gitea Runner {name} on {host} as {user} (sudo required)",
                    name=name,
                    host=host,
                    user=user,
                )
            )

            service_status = "unknown"
            try:
                stdout = self._executor.run_ad_hoc(
                    host,
                    user,
                    key,
                    "shell",
                    f"sudo -u grm-{name} systemctl --user is-active gitea-runner 2>/dev/null",
                    become=True,
                    ask_become_pass=True,
                    check=False,
                )
                service_status = self._parse_status(stdout)
            except AnsibleError:
                service_status = "unknown"
            translated_status = _(
                service_status if service_status in {"active", "inactive", "failed", "unknown"} else "unknown"
            )
            result.append(
                {
                    "name": name,
                    "host": host,
                    "user": user,
                    "labels": info.get("labels", ""),
                    "status": translated_status,
                }
            )
        return result

    @staticmethod
    def _parse_status(stdout: str) -> str:
        ansible_noise = (" | CHANGED | ", " | FAILED | ", " | UNREACHABLE | ", "[WARNING]", "ssh:", ">>")
        lines = [ln for ln in stdout.splitlines() if ln.strip() and not any(p in ln for p in ansible_noise)]
        return lines[-1].strip() if lines else "unknown"

    def _build_cmd(
        self,
        playbook_name: str,
        host: str,
        user: str,
        extra_vars: str,
        key: str | None = None,
        ask_become_pass: bool = False,
    ) -> list[str]:
        """Build the ansible-playbook command."""
        playbook = Path(__file__).parent.parent.parent / "ansible" / playbook_name
        if not playbook.exists():
            raise AnsibleError(_("Playbook not found: {playbook}", playbook=playbook))

        cmd = [
            "ansible-playbook",
            str(playbook),
            "-i",
            f"{host},",
            "-u",
            user,
            "--extra-vars",
            extra_vars,
        ]
        if key:
            cmd.extend(["--private-key", key])
        if ask_become_pass:
            cmd.append("--ask-become-pass")
        return cmd