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

"""Git operations for workspace management."""

import asyncio
import logging
import os
import subprocess
import tempfile

logger = logging.getLogger(__name__)


class GitService:
    """Low-level git operations for creating and syncing workspaces."""

    @staticmethod
    def _prepare_ssh_env(
        ssh_key: str | None,
    ) -> tuple[dict[str, str] | None, str | None]:
        """Prepare environment for git commands with SSH authentication.

        Returns a tuple of (env_dict, temp_key_path). Caller must clean up key_path.
        """
        if not ssh_key:
            return None, None

        fd, key_path = tempfile.mkstemp(prefix="ssh_key_")
        try:
            os.write(fd, ssh_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

    @staticmethod
    async def clone(
        remote_url: str, branch: str, path: str, ssh_key: str | None = None
    ) -> None:
        """Clone a repository to the given path.

        Args:
            remote_url: The git remote URL.
            branch: The branch to clone.
            path: The destination path for the clone.
            ssh_key: Optional decrypted SSH private key for authentication.

        Raises:
            RuntimeError: If the clone fails.
        """
        cmd = [
            "git",
            "clone",
            "--branch",
            branch,
            "--single-branch",
            remote_url,
            path,
        ]

        env, key_path = GitService._prepare_ssh_env(ssh_key)
        try:
            proc = await asyncio.create_subprocess_exec(
                *cmd,
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.PIPE,
                env={**os.environ, **env} if env else None,
            )
            stdout, stderr = await proc.communicate()
            if proc.returncode != 0:
                error_msg = stderr.decode().strip() if stderr else "unknown error"
                logger.error("Git clone failed: %s", error_msg)
                raise RuntimeError(f"Git clone failed: {error_msg}")
            logger.debug("Cloned %s (branch: %s) to %s", remote_url, branch, path)
        finally:
            if key_path and os.path.exists(key_path):
                os.unlink(key_path)

    @staticmethod
    async def fetch(path: str, ssh_key: str | None = None) -> None:
        """Fetch from origin.

        Args:
            path: The path to the local git repository.
            ssh_key: Optional decrypted SSH private key for authentication.

        Raises:
            RuntimeError: If fetch fails.
        """
        env, key_path = GitService._prepare_ssh_env(ssh_key)
        try:
            proc = await asyncio.create_subprocess_exec(
                "git",
                "-C",
                path,
                "fetch",
                "origin",
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.PIPE,
                env={**os.environ, **env} if env else None,
            )
            stdout, stderr = await proc.communicate()
            if proc.returncode != 0:
                error_msg = stderr.decode().strip() if stderr else "unknown error"
                logger.error("Git fetch failed: %s", error_msg)
                raise RuntimeError(f"Git fetch failed: {error_msg}")
            logger.debug("Fetched origin for %s", path)
        finally:
            if key_path and os.path.exists(key_path):
                os.unlink(key_path)

    @staticmethod
    async def pull(path: str, branch: str, ssh_key: str | None = None) -> None:
        """Pull latest changes from origin.

        Args:
            path: The path to the local git repository.
            branch: The branch to pull.
            ssh_key: Optional decrypted SSH private key for authentication.

        Raises:
            RuntimeError: If pull fails.
        """
        env, key_path = GitService._prepare_ssh_env(ssh_key)
        try:
            proc = await asyncio.create_subprocess_exec(
                "git",
                "-C",
                path,
                "pull",
                "origin",
                branch,
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.PIPE,
                env={**os.environ, **env} if env else None,
            )
            stdout, stderr = await proc.communicate()
            if proc.returncode != 0:
                error_msg = stderr.decode().strip() if stderr else "unknown error"
                logger.error("Git pull failed: %s", error_msg)
                raise RuntimeError(f"Git pull failed: {error_msg}")
            logger.debug("Pulled origin/%s for %s", branch, path)
        finally:
            if key_path and os.path.exists(key_path):
                os.unlink(key_path)

    @staticmethod
    def branch_exists_remotely(
        path: str, branch: str, ssh_key: str | None = None
    ) -> bool:
        """Check if a branch exists on the remote.

        Args:
            path: The path to the local git repository.
            branch: The branch name to check.
            ssh_key: Optional decrypted SSH private key for authentication.

        Returns:
            True if the branch exists on origin, False otherwise.
        """
        env, key_path = GitService._prepare_ssh_env(ssh_key)
        try:
            result = subprocess.run(
                ["git", "-C", path, "ls-remote", "--heads", "origin", branch],
                capture_output=True,
                text=True,
                env={**os.environ, **env} if env else None,
            )
            exists = result.returncode == 0 and result.stdout.strip() != ""
            logger.debug("Branch %s exists on remote: %s", branch, exists)
            return exists
        finally:
            if key_path and os.path.exists(key_path):
                os.unlink(key_path)