/headroom/lean_ctx/installer.py

"""Download and install lean-ctx binary from GitHub releases."""

from __future__ import annotations

import io
import logging
import os
import platform
import stat
import subprocess
import tarfile
import zipfile
from pathlib import Path
from urllib.request import urlopen

from headroom._subprocess import run

from . import LEAN_CTX_BIN_DIR, LEAN_CTX_VERSION

logger = logging.getLogger(__name__)

GITHUB_RELEASE_URL = "https://github.com/yvgude/lean-ctx/releases/download"


def _detect_runtime_target_triple() -> str:
    """Detect platform and return the lean-ctx release target triple."""
    system = platform.system()
    machine = platform.machine()

    if system == "Darwin":
        arch = "aarch64" if machine == "arm64" else "x86_64"
        return f"{arch}-apple-darwin"
    if system == "Linux":
        arch = "aarch64" if machine == "aarch64" else "x86_64"
        suffix = "unknown-linux-musl" if _is_musl() else "unknown-linux-gnu"
        return f"{arch}-{suffix}"
    if system == "Windows":
        return "x86_64-pc-windows-msvc"

    raise RuntimeError(f"Unsupported platform: {system} {machine}")


def _is_musl() -> bool:
    try:
        result = run(
            ["ldd", "--version"],
            capture_output=True,
            text=True,
            timeout=2,
            check=False,
        )
        return "musl" in (result.stdout + result.stderr).lower()
    except Exception:
        return False


def _get_target_triple() -> str:
    """Return the requested lean-ctx target triple, honoring explicit overrides."""
    return _get_explicit_target_triple() or _detect_runtime_target_triple()


def _get_explicit_target_triple() -> str:
    """Return the explicitly requested lean-ctx target triple, if any."""
    return (
        os.environ.get("HEADROOM_LEAN_CTX_TARGET", "").strip()
        or os.environ.get("LEAN_CTX_TARGET", "").strip()
    )


def _binary_name_for_target(target: str) -> str:
    """Return the expected binary name for a target triple."""
    return "lean-ctx.exe" if "windows" in target else "lean-ctx"


def _should_verify_target(target: str) -> bool:
    """Verify runtime-detected targets; explicit overrides may be cross-target."""
    if _get_explicit_target_triple():
        return False
    return target == _detect_runtime_target_triple()


def _get_download_url(version: str) -> tuple[str, str]:
    """Get download URL and extension for this platform."""
    target = _get_target_triple()
    ext = "zip" if "windows" in target else "tar.gz"
    url = f"{GITHUB_RELEASE_URL}/{version}/lean-ctx-{target}.{ext}"
    return url, ext


def download_lean_ctx(version: str | None = None) -> Path:
    """Download lean-ctx binary from GitHub releases."""
    version = version or LEAN_CTX_VERSION
    target = _get_target_triple()
    url, ext = _get_download_url(version)
    target_path = LEAN_CTX_BIN_DIR / _binary_name_for_target(target)

    LEAN_CTX_BIN_DIR.mkdir(parents=True, exist_ok=True)

    logger.info("Downloading lean-ctx %s from %s ...", version, url)

    try:
        if not url.startswith(("http://", "https://")):
            raise ValueError(f"Invalid URL scheme in {url}")
        try:
            with urlopen(url, timeout=30) as response:
                data = response.read()
        except Exception as download_err:
            if "CERTIFICATE_VERIFY_FAILED" in str(download_err):
                raise RuntimeError(
                    "TLS verification failed downloading lean-ctx; "
                    "fix the local trust store and retry."
                ) from download_err
            raise
    except Exception as e:
        raise RuntimeError(f"Failed to download lean-ctx from {url}: {e}") from e

    try:
        if ext == "tar.gz":
            with tarfile.open(fileobj=io.BytesIO(data), mode="r:gz") as tar:
                for member in tar.getmembers():
                    if member.name.endswith("/lean-ctx") or member.name == "lean-ctx":
                        member.name = target_path.name
                        tar.extract(member, LEAN_CTX_BIN_DIR)
                        break
                else:
                    raise RuntimeError("lean-ctx binary not found in archive")
        elif ext == "zip":
            with zipfile.ZipFile(io.BytesIO(data)) as zf:
                for name in zf.namelist():
                    if name.endswith("lean-ctx.exe") or name.endswith("/lean-ctx"):
                        with zf.open(name) as src, open(target_path, "wb") as dst:
                            dst.write(src.read())
                        break
                else:
                    raise RuntimeError("lean-ctx binary not found in archive")
    except (tarfile.TarError, zipfile.BadZipFile) as e:
        raise RuntimeError(f"Failed to extract lean-ctx archive: {e}") from e

    if "windows" not in target:
        target_path.chmod(target_path.stat().st_mode | stat.S_IEXEC | stat.S_IXGRP | stat.S_IXOTH)

    if _should_verify_target(target):
        try:
            result = run(
                [str(target_path), "--version"],
                capture_output=True,
                text=True,
                timeout=5,
            )
            if result.returncode != 0:
                raise RuntimeError(f"lean-ctx verification failed: {result.stderr}")
            logger.info("lean-ctx installed: %s", result.stdout.strip())
        except FileNotFoundError as e:
            raise RuntimeError("lean-ctx binary not found after extraction") from e
        except subprocess.TimeoutExpired as e:
            raise RuntimeError("lean-ctx verification timed out") from e
    else:
        logger.info(
            "lean-ctx installed for target %s at %s (verification skipped)",
            target,
            target_path,
        )

    return target_path


def ensure_lean_ctx(version: str | None = None) -> Path | None:
    """Ensure lean-ctx is installed — download if needed."""
    from . import get_lean_ctx_path

    existing = get_lean_ctx_path()
    if existing:
        return existing

    try:
        return download_lean_ctx(version)
    except RuntimeError as e:
        logger.warning("Could not install lean-ctx: %s", e)
        return None