headquarter/apps/api/src/services/git/operations.py

"""Git repository operations service."""

import logging
import os
import subprocess
import uuid

from fastapi import HTTPException, status

from src.config import Settings
from src.models import SSHKey
from src.services.shared.ssh_keys import _get_fernet

logger = logging.getLogger(__name__)


def get_repo_path(user_id: uuid.UUID, project_id: uuid.UUID, name: str) -> str:
    """Generate the filesystem path for a repository."""
    base = Settings().repo_base_path or "/data/repos"
    return os.path.join(base, str(user_id), str(project_id), f"{name}.git")


def build_provider_clone_url(owner: str, repo: str) -> str:
    """Build the SSH clone URL for the fixed git provider."""
    return f"git@git.commumedia.org:{owner}/{repo}.git"


def prepare_ssh_env(ssh_key: SSHKey | None) -> tuple[dict, str] | None:
    """Prepare environment variables for git commands with SSH authentication."""
    if ssh_key is None:
        return None

    import tempfile

    fernet = _get_fernet()
    private_key = fernet.decrypt(ssh_key.private_key_encrypted.encode()).decode()

    fd, key_path = tempfile.mkstemp(prefix="ssh_key_")
    try:
        os.write(fd, private_key.encode())
    finally:
        os.close(fd)
    os.chmod(key_path, 0o600)

    env = {
        "GIT_SSH_COMMAND": f"ssh -i {key_path} -o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null"
    }
    return env, key_path


def preflight_remote_repository(remote_url: str, ssh_key: SSHKey | None = None) -> None:
    """Verify a remote repository is reachable before cloning."""
    env = None
    key_path = None

    if ssh_key is not None:
        ssh_result = prepare_ssh_env(ssh_key)
        if ssh_result:
            env, key_path = ssh_result

    try:
        result = subprocess.run(
            ["git", "ls-remote", remote_url],
            capture_output=True,
            text=True,
            timeout=60,
            env={**os.environ, **env} if env else None,
        )
    except subprocess.TimeoutExpired:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="remote repository check timed out",
        )
    except FileNotFoundError:
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail="git command not found",
        )
    finally:
        if key_path and os.path.exists(key_path):
            os.unlink(key_path)

    if result.returncode != 0:
        logger.error(
            "Preflight check failed for %s: stderr=%s", remote_url, result.stderr
        )
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=f"repository not found or inaccessible: {result.stderr}",
        )


def clone_working_repository(remote_url: str, repo_path: str, ssh_key: SSHKey | None = None) -> None:
    """Clone a remote repository to a local path."""
    env = None
    key_path = None

    if ssh_key is not None:
        ssh_result = prepare_ssh_env(ssh_key)
        if ssh_result:
            env, key_path = ssh_result

    try:
        result = subprocess.run(
            ["git", "clone", remote_url, repo_path],
            capture_output=True,
            text=True,
            timeout=300,
            env={**os.environ, **env} if env else None,
        )
    except subprocess.TimeoutExpired:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST, detail="clone operation timed out"
        )
    except FileNotFoundError:
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail="git command not found",
        )
    finally:
        if key_path and os.path.exists(key_path):
            os.unlink(key_path)

    if result.returncode != 0:
        logger.error("Clone failed for %s: stderr=%s", remote_url, result.stderr)
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=f"failed to clone repository: {result.stderr}",
        )


def init_working_repository(repo_path: str) -> None:
    """Initialize a new git repository at the given path."""
    try:
        result = subprocess.run(
            ["git", "init", "-b", "main", repo_path],
            capture_output=True,
            text=True,
        )
    except FileNotFoundError:
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail="git command not found",
        )

    if result.returncode == 0:
        return

    fallback = subprocess.run(
        ["git", "init", repo_path],
        capture_output=True,
        text=True,
    )
    if fallback.returncode != 0:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=f"failed to initialize repository: {fallback.stderr}",
        )

    ref_result = subprocess.run(
        ["git", "-C", repo_path, "symbolic-ref", "HEAD", "refs/heads/main"],
        capture_output=True,
        text=True,
    )
    if ref_result.returncode != 0:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=f"failed to set initial branch: {ref_result.stderr}",
        )


def list_remote_branches(remote_url: str, ssh_key: SSHKey | None = None) -> tuple[list[str], str]:
    """List branches from a remote repository via ls-remote.

    Returns:
        Tuple of (branch_names, default_branch).
    """
    ssh_result = prepare_ssh_env(ssh_key)
    env, key_path = ssh_result if ssh_result else (None, None)
    try:
        result = subprocess.run(
            ["git", "ls-remote", "--heads", remote_url],
            capture_output=True,
            text=True,
            timeout=30,
            env={**os.environ, **env} if env else None,
        )
        if result.returncode != 0:
            logger.warning("ls-remote returned %d: %s", result.returncode, result.stderr)
            raise RuntimeError(f"ls-remote failed: {result.stderr}")
        branches = []
        default_branch = "main"
        for line in result.stdout.strip().split("\n"):
            if not line:
                continue
            parts = line.split("\t")
            if len(parts) == 2:
                ref = parts[1]
                if ref.startswith("refs/heads/"):
                    branch_name = ref[len("refs/heads/"):]
                    branches.append(branch_name)
                    if branch_name in ("main", "master"):
                        default_branch = branch_name
        return branches, default_branch
    except subprocess.TimeoutExpired:
        logger.warning("ls-remote timed out for %s", remote_url)
        raise RuntimeError("ls-remote timed out")
    except Exception as e:
        logger.warning("ls-remote failed for %s: %s", remote_url, str(e))
        raise RuntimeError(f"ls-remote failed: {e}")
    finally:
        if key_path and os.path.exists(key_path):
            os.unlink(key_path)