prole/knoe/core/ops/k3d_registry.py

"""Registry ops — k3d mode (local dev cluster).

Uses k3d's built-in registry or falls back to a plain Docker registry container.
"""
from __future__ import annotations

from pathlib import Path

from ._services_common import (
    _LogFn,
    _docker,
    _k3d,
    _log,
    _to_bool,
)


def _k3d_registry_running(raw_list: str, name: str) -> bool:
    """Return True only if a line whose NAME ends with `name` has STATUS=running."""
    for line in raw_list.splitlines():
        parts = line.split()
        if not parts:
            continue
        # k3d may prefix the name with "k3d-"; match the suffix.
        if parts[0].endswith(name) and parts[-1].lower() == "running":
            return True
    return False


def _ensure_k3d_registry(*, env: dict | None = None, log: _LogFn | None = None) -> None:
    name = str((env or {}).get("K3D_REGISTRY_NAME") or "knoe-registry")
    port = str((env or {}).get("REGISTRY_PORT") or "5000")
    legacy_name = "prole-registry"

    listed = _k3d(["registry", "list"], env=env, timeout=60)
    raw = listed.stdout if listed.returncode == 0 else ""

    # Remove legacy prole-registry that may be squatting on the port.
    if legacy_name in raw:
        _log(log, f"[REGISTRY] Removing legacy k3d registry {legacy_name}")
        _k3d(["registry", "delete", legacy_name], env=env, timeout=60)
        listed = _k3d(["registry", "list"], env=env, timeout=60)
        raw = listed.stdout if listed.returncode == 0 else ""

    if name in raw:
        if _k3d_registry_running(raw, name):
            _log(log, f"[REGISTRY] k3d registry {name} already running")
            return
        # Exists but not running (e.g. "created" — port was unavailable).
        _log(log, f"[REGISTRY] k3d registry {name} exists but not running; recreating")
        _k3d(["registry", "delete", name], env=env, timeout=60)

    _log(log, f"[REGISTRY] Creating k3d registry {name} on port {port}")
    created = _k3d(["registry", "create", name, "--port", f"{port}:{port}"], env=env, timeout=180)
    if created.returncode == 0:
        return

    # fallback docker registry
    _log(log, "[REGISTRY] k3d registry creation failed; trying docker registry fallback")
    running = _docker(["ps", "--format", "{{.Names}}"], env=env, timeout=30)
    if running.returncode == 0 and name in running.stdout.splitlines():
        return
    _docker(["rm", "-f", name], env=env, timeout=30)
    _docker(
        [
            "run",
            "-d",
            "--restart=always",
            "-p",
            f"{port}:5000",
            "--name",
            name,
            "registry:2",
        ],
        env=env,
        timeout=120,
        check=True,
    )


def initialize(
    *,
    namespace: str | None = None,
    env: dict | None = None,
    project_root: str | Path = ".",
    log: _LogFn | None = None,
    mode: str | None = None,
) -> None:
    update(namespace=namespace, env=env, project_root=project_root, log=log, mode=mode)


def start(
    *,
    namespace: str | None = None,
    env: dict | None = None,
    project_root: str | Path = ".",
    log: _LogFn | None = None,
    mode: str | None = None,
) -> None:
    update(namespace=namespace, env=env, project_root=project_root, log=log, mode=mode)


def update(
    *,
    namespace: str | None = None,
    env: dict | None = None,
    project_root: str | Path = ".",
    log: _LogFn | None = None,
    mode: str | None = None,
) -> None:
    _ensure_k3d_registry(env=env, log=log)


def stop(
    *,
    namespace: str | None = None,
    env: dict | None = None,
    mode: str | None = None,
    log: _LogFn | None = None,
) -> None:
    name = str((env or {}).get("K3D_REGISTRY_NAME") or "knoe-registry")
    _log(log, f"[REGISTRY] Removing k3d registry {name}")
    _k3d(["registry", "delete", name], env=env, timeout=120)
    _docker(["rm", "-f", name], env=env, timeout=30)


def restart(
    *,
    namespace: str | None = None,
    env: dict | None = None,
    mode: str | None = None,
    log: _LogFn | None = None,
) -> None:
    update(namespace=namespace, env=env, mode=mode, log=log)


def status(
    *,
    namespace: str | None = None,
    env: dict | None = None,
    mode: str | None = None,
) -> bool:
    name = str((env or {}).get("K3D_REGISTRY_NAME") or "knoe-registry")
    reg = _k3d(["registry", "list"], env=env, timeout=30)
    return _to_bool(reg.returncode == 0 and name in reg.stdout)