prole/knoe/core/ops/_services_common.py

from __future__ import annotations

import os
import subprocess
from pathlib import Path
from typing import Callable

_LogFn = Callable[[str], None]


def _log(log: _LogFn | None, msg: str) -> None:
    if log:
        log(msg)


def _merge_env(env: dict | None) -> dict:
    merged = dict(os.environ)
    if env:
        merged.update(env)
    return merged


def _run(
    args: list[str],
    *,
    env: dict | None = None,
    input_text: str | None = None,
    timeout: int = 300,
    check: bool = False,
) -> subprocess.CompletedProcess:
    res = subprocess.run(
        args,
        input=input_text,
        capture_output=True,
        text=True,
        env=_merge_env(env),
        timeout=timeout,
    )
    if check and res.returncode != 0:
        raise RuntimeError(
            f"Command failed ({res.returncode}): {' '.join(args)}\n"
            f"stdout: {res.stdout}\n"
            f"stderr: {res.stderr}"
        )
    return res


def _kubectl(
    args: list[str],
    *,
    env: dict | None = None,
    input_text: str | None = None,
    timeout: int = 300,
    check: bool = False,
) -> subprocess.CompletedProcess:
    return _run(
        ["kubectl", *args],
        env=env,
        input_text=input_text,
        timeout=timeout,
        check=check,
    )


def _helm(
    args: list[str],
    *,
    env: dict | None = None,
    timeout: int = 300,
    check: bool = False,
) -> subprocess.CompletedProcess:
    return _run(["helm", *args], env=env, timeout=timeout, check=check)


def _k3d(
    args: list[str],
    *,
    env: dict | None = None,
    timeout: int = 120,
    check: bool = False,
) -> subprocess.CompletedProcess:
    return _run(["k3d", *args], env=env, timeout=timeout, check=check)


def _docker(
    args: list[str],
    *,
    env: dict | None = None,
    timeout: int = 120,
    check: bool = False,
) -> subprocess.CompletedProcess:
    return _run(["docker", *args], env=env, timeout=timeout, check=check)


def _detect_mode(mode: str | None, env: dict | None) -> str:
    if mode:
        return str(mode).strip()
    if env:
        for key in ("PROLE_MODE", "DEPLOYMENT_MODE"):
            value = str(env.get(key) or "").strip()
            if value:
                return value
    return "k3s"


def _namespace(namespace: str | None, env: dict | None, default: str = "default") -> str:
    raw = (
        (namespace or "").strip()
        or str((env or {}).get("SERVICE_NAMESPACE") or "").strip()
        or str((env or {}).get("NAMESPACE") or "").strip()
        or default
    )
    return raw


def _registry_namespace(namespace: str | None, env: dict | None) -> str:
    raw = (
        (namespace or "").strip()
        or str((env or {}).get("REGISTRY_NAMESPACE") or "").strip()
        or str((env or {}).get("NAMESPACE") or "").strip()
        or "default"
    )
    return raw


def _ensure_namespace(namespace: str, env: dict | None = None) -> None:
    res = _kubectl(["get", "namespace", namespace], env=env, timeout=30)
    if res.returncode == 0:
        return
    _kubectl(["create", "namespace", namespace], env=env, timeout=60, check=True)


def _wait_rollout(
    kind: str,
    name: str,
    namespace: str,
    *,
    env: dict | None = None,
    timeout: str = "300s",
) -> None:
    _kubectl(
        [
            "-n",
            namespace,
            "rollout",
            "status",
            f"{kind}/{name}",
            f"--timeout={timeout}",
        ],
        env=env,
        timeout=360,
        check=True,
    )


def _manifest_path(project_root: str | Path, *segments: str) -> Path:
    return Path(project_root).joinpath(*segments)


def _exists(kind: str, name: str, namespace: str, env: dict | None = None) -> bool:
    res = _kubectl(["-n", namespace, "get", kind, name], env=env, timeout=20)
    return res.returncode == 0


def _to_bool(value: str | bool | None, default: bool = False) -> bool:
    if isinstance(value, bool):
        return value
    if value is None:
        return default
    return str(value).strip().lower() in {"1", "true", "yes", "on"}