headquarter/apps/api/src/services/git_operations.py

"""Git commands scoped to a workspace directory."""

import asyncio
import logging
from dataclasses import dataclass

from src.models.workspace import Workspace

logger = logging.getLogger(__name__)


@dataclass
class GitStatus:
    """Parsed git status output."""

    branch: str
    modified: list[str]
    added: list[str]
    deleted: list[str]
    untracked: list[str]
    ahead: int = 0
    behind: int = 0


@dataclass
class Commit:
    """A single git commit."""

    hash: str
    message: str
    author: str
    date: str


class GitOperations:
    """Run git commands within a workspace directory."""

    def __init__(self, workspace: Workspace) -> None:
        self.cwd = workspace.path
        self.branch = workspace.branch

    async def _run(self, *cmd: str) -> tuple[int, str, str]:
        """Run a git command and return (returncode, stdout, stderr)."""
        proc = await asyncio.create_subprocess_exec(
            *cmd,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
        )
        stdout, stderr = await proc.communicate()
        return proc.returncode or 0, stdout.decode(), stderr.decode()

    async def status(self) -> GitStatus:
        """Get git status for the workspace."""
        returncode, stdout, _ = await self._run(
            "git", "-C", self.cwd, "status", "--porcelain", "-b"
        )

        modified: list[str] = []
        added: list[str] = []
        deleted: list[str] = []
        untracked: list[str] = []
        branch = self.branch
        ahead = 0
        behind = 0

        for line in stdout.splitlines():
            if line.startswith("##"):
                # Branch info line
                branch_info = line[3:].strip()
                if "..." in branch_info:
                    branch = branch_info.split("...")[0]
                    if "[ahead " in branch_info:
                        ahead_str = branch_info.split("[ahead ")[1].split("]")[0]
                        ahead = int(ahead_str.split(",")[0])
                    if "[behind " in branch_info:
                        behind_str = branch_info.split("[behind ")[1].split("]")[0]
                        behind = int(behind_str.split(",")[0])
                else:
                    branch = branch_info
                continue

            if len(line) < 3:
                continue

            status_code = line[:2]
            file_path = line[3:]

            # XY format: X = index status, Y = working tree status
            if status_code == "??":
                untracked.append(file_path)
            elif status_code[1] == "D" or status_code[0] == "D":
                deleted.append(file_path)
            elif status_code[0] == "A" or status_code[1] == "A":
                added.append(file_path)
            else:
                modified.append(file_path)

        return GitStatus(
            branch=branch,
            modified=modified,
            added=added,
            deleted=deleted,
            untracked=untracked,
            ahead=ahead,
            behind=behind,
        )

    async def commit(self, message: str) -> None:
        """Stage all changes and commit."""
        rc, _, err = await self._run("git", "-C", self.cwd, "add", "-A")
        if rc != 0:
            raise RuntimeError(f"Git add failed: {err}")

        rc, _, err = await self._run("git", "-C", self.cwd, "commit", "-m", message)
        if rc != 0:
            raise RuntimeError(f"Git commit failed: {err}")

        logger.info("Committed in workspace: %s", self.cwd)

    async def push(self) -> None:
        """Push current branch to origin."""
        rc, _, err = await self._run(
            "git", "-C", self.cwd, "push", "origin", self.branch
        )
        if rc != 0:
            raise RuntimeError(f"Git push failed: {err}")

        logger.info("Pushed branch %s from workspace: %s", self.branch, self.cwd)

    async def pull(self) -> None:
        """Pull current branch from origin."""
        rc, _, err = await self._run(
            "git", "-C", self.cwd, "pull", "origin", self.branch
        )
        if rc != 0:
            raise RuntimeError(f"Git pull failed: {err}")

        logger.info("Pulled branch %s in workspace: %s", self.branch, self.cwd)

    async def fetch(self) -> None:
        """Fetch from origin."""
        rc, _, err = await self._run("git", "-C", self.cwd, "fetch", "origin")
        if rc != 0:
            raise RuntimeError(f"Git fetch failed: {err}")

        logger.info("Fetched origin for workspace: %s", self.cwd)

    async def checkout(self, branch: str) -> None:
        """Checkout a branch."""
        rc, _, err = await self._run("git", "-C", self.cwd, "checkout", branch)
        if rc != 0:
            raise RuntimeError(f"Git checkout failed: {err}")

        self.branch = branch
        logger.info("Checked out branch %s in workspace: %s", branch, self.cwd)

    async def history(self, path: str | None = None, limit: int = 50) -> list[Commit]:
        """Get commit history.

        Args:
            path: Optional file path to filter history.
            limit: Maximum number of commits.

        Returns:
            List of commits.
        """
        cmd = [
            "git",
            "-C",
            self.cwd,
            "log",
            f"--max-count={limit}",
            "--pretty=format:%H|%s|%an|%ad",
            "--date=iso",
        ]
        if path:
            cmd.extend(["--", path])

        rc, stdout, err = await self._run(*cmd)
        if rc != 0:
            raise RuntimeError(f"Git log failed: {err}")

        commits = []
        for line in stdout.strip().splitlines():
            parts = line.split("|", 3)
            if len(parts) >= 4:
                commits.append(
                    Commit(
                        hash=parts[0],
                        message=parts[1],
                        author=parts[2],
                        date=parts[3],
                    )
                )

        return commits

    async def branches(self) -> tuple[list[str], str]:
        """List all branches and current branch.

        Returns:
            Tuple of (all_branches, current_branch).
        """
        rc, stdout, err = await self._run(
            "git", "-C", self.cwd, "branch", "-a", "--format=%(refname:short)"
        )
        if rc != 0:
            raise RuntimeError(f"Git branch failed: {err}")

        branches = []
        current = self.branch
        for line in stdout.strip().splitlines():
            line = line.strip()
            if line.startswith("HEAD") or line.endswith("/HEAD"):
                continue
            if line.startswith("remotes/origin/"):
                branch_name = line.replace("remotes/origin/", "")
                if branch_name not in branches:
                    branches.append(branch_name)
            elif line and line not in branches:
                branches.append(line)

        return branches, current