src/core/tools/git_server_client.py

#!/usr/bin/env python3
"""
Git Server Client for connecting to external Gitea instance.

This module provides a lightweight client for interacting with a shared
Gitea service, optimized for task-based isolation where multiple environment
instances share the same Gitea server but have isolated workspaces.
"""

import json
import os
import shutil
import subprocess
import time
from dataclasses import dataclass
from pathlib import Path
from urllib.parse import urlparse


@dataclass
class RepoInfo:
    """Information about a repository."""

    name: str
    url: str
    commit: str
    clone_url: str


class GitServerClient:
    """
    Client for connecting to an external Gitea server.

    This client is optimized for task-based isolation where:
    - Multiple tasks share the same Gitea instance
    - Each task has its own isolated workspace
    - Fast reset() via git operations (no server restart)
    - Repos are pre-migrated to Gitea once

    Args:
        gitea_url: URL of the Gitea server (e.g., "http://gitea:3000")
        username: Gitea username for authentication
        password: Gitea password for authentication
        workspace_dir: Local workspace directory for cloning repos

    Example:
        >>> # Connect to shared Gitea (credentials from environment)
        >>> import os
        >>> client = GitServerClient(
        ...     gitea_url=os.getenv("GITEA_URL"),
        ...     username=os.getenv("GITEA_USERNAME"),
        ...     password=os.getenv("GITEA_PASSWORD")
        ... )
        >>> client.wait_for_ready()
        >>> # Clone repo to workspace
        >>> path = client.clone_to_workspace("my-repo", commit="abc123")
        >>> # Fast reset to base state
        >>> client.reset_workspace("my-repo", commit="abc123")
    """

    def __init__(
        self,
        gitea_url: str,
        username: str,
        password: str,
        workspace_dir: str = "/workspace",
    ):
        """Initialize Git Server Client."""
        self.gitea_url = gitea_url.rstrip("/")
        self.username = username
        self.password = password
        self.workspace_dir = Path(workspace_dir)
        self.is_ready = False

        # Parse Gitea URL
        parsed = urlparse(self.gitea_url)
        self.domain = parsed.hostname or "localhost"
        self.port = parsed.port or 3000

        # Ensure workspace exists
        os.makedirs(self.workspace_dir, exist_ok=True)

        # Configure git credentials
        self._configure_git()

    def _configure_git(self):
        """Configure git credentials for automatic authentication."""
        home_dir = Path.home()

        # Git config
        git_config = f"""[user]
    name = {self.username}
    email = {self.username}@local.env
[init]
    defaultBranch = main
[credential]
    helper = store
"""
        gitconfig_path = home_dir / ".gitconfig"
        gitconfig_path.write_text(git_config)

        # Git credentials
        git_credentials = f"http://{self.username}:{self.password}@{self.domain}:{self.port}\n"
        gitcreds_path = home_dir / ".git-credentials"
        gitcreds_path.write_text(git_credentials)
        gitcreds_path.chmod(0o600)

    def wait_for_ready(self, timeout: int = 30) -> bool:
        """
        Wait for Gitea server to be ready.

        Args:
            timeout: Maximum seconds to wait

        Returns:
            True if server is ready, False otherwise
        """
        start_time = time.time()
        while time.time() - start_time < timeout:
            try:
                result = subprocess.run(
                    ["curl", "-sf", f"{self.gitea_url}/"],
                    capture_output=True,
                    timeout=5,
                )
                if result.returncode == 0:
                    self.is_ready = True
                    return True
            except subprocess.TimeoutExpired:
                pass
            except Exception:
                pass

            time.sleep(1)

        return False

    def list_repositories(self) -> list[dict[str, str]]:
        """
        List all repositories in Gitea.

        Returns:
            List of repository information dictionaries
        """
        if not self.is_ready:
            raise RuntimeError("Gitea server is not ready")

        result = subprocess.run(
            [
                "curl",
                "-s",
                f"{self.gitea_url}/api/v1/user/repos",
                "-u",
                f"{self.username}:{self.password}",
            ],
            capture_output=True,
            text=True,
        )

        if result.returncode != 0:
            return []

        try:
            repos = json.loads(result.stdout)
            return [
                {
                    "name": repo["name"],
                    "full_name": repo["full_name"],
                    "clone_url": repo["clone_url"],
                    "description": repo.get("description", ""),
                }
                for repo in repos
            ]
        except (json.JSONDecodeError, KeyError):
            return []

    def clone_to_workspace(
        self, repo_name: str, target_dir: str | None = None, commit: str = "main"
    ) -> str:
        """
        Clone a repository to the workspace at a specific commit.

        This creates a fresh clone optimized for task isolation.

        Args:
            repo_name: Name of repository to clone
            target_dir: Target directory name (defaults to repo_name)
            commit: Commit hash or branch to checkout

        Returns:
            Path to cloned repository

        Raises:
            RuntimeError: If clone fails
        """
        if not self.is_ready:
            raise RuntimeError("Gitea server is not ready")

        target_dir = target_dir or repo_name
        target_path = self.workspace_dir / target_dir

        # Remove existing directory if present
        if target_path.exists():
            shutil.rmtree(target_path)

        clone_url = f"{self.gitea_url}/{self.username}/{repo_name}.git"

        # Clone repository
        result = subprocess.run(
            ["git", "clone", clone_url, str(target_path)],
            capture_output=True,
            text=True,
        )

        if result.returncode != 0:
            raise RuntimeError(f"Clone failed: {result.stderr}")

        # Checkout specific commit
        if commit != "main":
            result = subprocess.run(
                ["git", "checkout", commit],
                cwd=str(target_path),
                capture_output=True,
                text=True,
            )

            if result.returncode != 0:
                raise RuntimeError(f"Checkout failed: {result.stderr}")

        return str(target_path)

    def reset_workspace(self, repo_name: str, commit: str = "main") -> bool:
        """
        Fast reset of workspace to base state (optimized for task resets).

        This is much faster than re-cloning. It:
        1. Checks out the target commit
        2. Resets to that commit (hard)
        3. Cleans untracked files

        Args:
            repo_name: Name of repository (directory in workspace)
            commit: Commit hash or branch to reset to

        Returns:
            True if reset successful

        Raises:
            RuntimeError: If reset fails
        """
        repo_path = self.workspace_dir / repo_name

        if not repo_path.exists():
            raise RuntimeError(f"Repository not found in workspace: {repo_name}")

        # Fetch latest (in case commit is new)
        subprocess.run(
            ["git", "fetch", "--all"],
            cwd=str(repo_path),
            capture_output=True,
        )

        # Checkout and hard reset to commit
        result = subprocess.run(
            ["git", "checkout", commit],
            cwd=str(repo_path),
            capture_output=True,
            text=True,
        )

        if result.returncode != 0:
            raise RuntimeError(f"Checkout failed: {result.stderr}")

        result = subprocess.run(
            ["git", "reset", "--hard", f"origin/{commit}" if commit != "main" else commit],
            cwd=str(repo_path),
            capture_output=True,
            text=True,
        )

        if result.returncode != 0:
            # Try without origin/ prefix
            result = subprocess.run(
                ["git", "reset", "--hard", commit],
                cwd=str(repo_path),
                capture_output=True,
                text=True,
            )
            if result.returncode != 0:
                raise RuntimeError(f"Reset failed: {result.stderr}")

        # Clean untracked files and directories
        subprocess.run(
            ["git", "clean", "-fdx"],
            cwd=str(repo_path),
            capture_output=True,
        )

        return True

    def execute_git_command(
        self, command: str, working_dir: str = ""
    ) -> tuple[int, str, str]:
        """
        Execute a git command in the workspace.

        Args:
            command: Git command to execute (without 'git' prefix)
            working_dir: Working directory relative to workspace

        Returns:
            Tuple of (exit_code, stdout, stderr)
        """
        work_path = (
            self.workspace_dir / working_dir if working_dir else self.workspace_dir
        )

        if not work_path.exists():
            return (1, "", f"Working directory does not exist: {work_path}")

        # Split command safely
        cmd_parts = ["git"] + command.split()

        result = subprocess.run(
            cmd_parts,
            cwd=str(work_path),
            capture_output=True,
            text=True,
        )

        return (result.returncode, result.stdout, result.stderr)

    def get_current_commit(self, repo_name: str) -> str:
        """
        Get current commit hash of a workspace repository.

        Args:
            repo_name: Name of repository in workspace

        Returns:
            Commit hash
        """
        repo_path = self.workspace_dir / repo_name

        if not repo_path.exists():
            raise RuntimeError(f"Repository not found: {repo_name}")

        result = subprocess.run(
            ["git", "rev-parse", "HEAD"],
            cwd=str(repo_path),
            capture_output=True,
            text=True,
        )

        if result.returncode != 0:
            raise RuntimeError(f"Failed to get commit: {result.stderr}")

        return result.stdout.strip()

    def workspace_exists(self, repo_name: str) -> bool:
        """Check if a repository exists in workspace."""
        return (self.workspace_dir / repo_name).exists()