svn-git-server/svn_mirror/gitea.py

"""Gitea integration – push/pull between canonical repo and Gitea's on-disk repo."""

import logging
import os
import subprocess
from pathlib import Path
from typing import Optional

from .mirror import Mirror

logger = logging.getLogger(__name__)


class GiteaError(Exception):
    pass


def _get_gitea_repo_path(mirror: Mirror) -> Path:
    """Return the filesystem path to Gitea's bare repo for this mirror."""
    try:
        return Path(mirror.config.gitea.repo_dir)
    except Exception as e:
        raise GiteaError(str(e))


# ─── Git subprocess helper ─────────────────────────────────────


def _git(*args: str, timeout: int = 300, check: bool = True) -> str:
    """Run a git command (uses CWD for repo discovery)."""
    try:
        r = subprocess.run(
            ["git"] + list(args),
            capture_output=True,
            text=True,
            timeout=timeout,
            env={**os.environ, "LC_ALL": "C", "GITEA_PUSH_VERIFIED": "1"},
        )
    except FileNotFoundError:
        raise GiteaError("git not found on PATH")
    except subprocess.TimeoutExpired:
        raise GiteaError(f"git timed out: {' '.join(args)}")
    if check and r.returncode != 0:
        raise GiteaError(
            f"git failed (exit {r.returncode}): "
            f"git {' '.join(args)}\n{r.stderr.strip()}"
        )
    return r.stdout.strip() if r.returncode == 0 else ""


# ─── Push canonical → Gitea ───────────────────────────────────


def push_to_gitea(mirror: Mirror) -> bool:
    """Push all heads and tags from the canonical repo to Gitea's bare repo.

    Returns True if anything was pushed.
    """
    gitea_path = _get_gitea_repo_path(mirror)
    if not gitea_path.exists():
        logger.warning("Gitea repo not found at %s – skipping push", gitea_path)
        return False

    canonical = str(mirror.canonical_dir)

    # Check if there's anything to push by comparing refs
    out = _git("--git-dir", canonical, "for-each-ref",
               "--format=%(objectname) %(refname)",
               "refs/heads/", "refs/tags/",
               timeout=30, check=False)
    if out:
        has_new = False
        for line in out.splitlines():
            line = line.strip()
            if not line:
                continue
            obj_hash, ref = line.split(None, 1)
            gitea_hash = _git("--git-dir", str(gitea_path), "rev-parse",
                              "--verify", "--quiet", ref,
                              timeout=30, check=False)
            if obj_hash != gitea_hash:
                has_new = True
                break
        if not has_new:
            logger.debug("No new commits to push to Gitea")
            return False

    logger.info("Pushing canonical → Gitea (%s) …", gitea_path)
    # Use `git push --prune` with explicit refspecs to mirror branches and tags
    _git(
        "--git-dir", canonical, "push", "--prune",
        str(gitea_path),
        "+refs/heads/*:refs/heads/*",
        "+refs/tags/*:refs/tags/*",
        timeout=300,
    )
    logger.info("Push to Gitea complete")
    return True


# ─── Fetch Gitea → canonical ──────────────────────────────────


def fetch_from_gitea(mirror: Mirror, ref: Optional[str] = None) -> bool:
    """Fetch refs from Gitea's bare repo into the canonical repo.

    If *ref* is given (e.g. ``refs/heads/master``), only that ref is
    fetched.  Otherwise all heads are fetched.

    Returns True if anything new was fetched.
    """
    gitea_path = _get_gitea_repo_path(mirror)
    if not gitea_path.exists():
        raise GiteaError(f"Gitea repo not found at {gitea_path}")

    canonical = str(mirror.canonical_dir)

    if ref:
        refspec = f"+{ref}:refs/remotes/gitea/{ref.removeprefix('refs/heads/')}"
    else:
        refspec = "+refs/heads/*:refs/remotes/gitea/*"

    logger.info("Fetching %s → canonical …", gitea_path)
    _git(
        "--git-dir", canonical, "fetch", "--prune",
        str(gitea_path), refspec,
        timeout=300,
    )

    logger.debug("Fetch from Gitea complete")

    # Fast-forward local branches from remote-tracking refs
    if not ref:
        _ff_local_branches(canonical)

    return True


def _ff_local_branches(git_dir: str):
    """Fast-forward ``refs/heads/*`` from ``refs/remotes/gitea/*``."""
    out = _git("--git-dir", git_dir, "for-each-ref",
               "--format=%(refname)", "refs/remotes/gitea/",
               timeout=30, check=False)
    for rt_ref in out.splitlines():
        rt_ref = rt_ref.strip()
        if not rt_ref:
            continue
        branch = rt_ref.removeprefix("refs/remotes/gitea/")
        local_ref = f"refs/heads/{branch}"

        # Check if local branch exists
        local_hash = _git("--git-dir", git_dir, "rev-parse", "--verify",
                          "--quiet", local_ref, timeout=15, check=False)
        if not local_hash:
            _git("--git-dir", git_dir, "update-ref", local_ref, rt_ref,
                 timeout=15)
            logger.debug("Created new branch %s from Gitea", local_ref)
            continue

        # Fast-forward if possible
        rt_hash = _git("--git-dir", git_dir, "rev-parse", "--verify",
                       rt_ref, timeout=15, check=False)
        if not rt_hash or rt_hash == local_hash:
            continue

        base = _git("--git-dir", git_dir, "merge-base", local_hash, rt_ref,
                     timeout=15, check=False)
        if base == local_hash:
            _git("--git-dir", git_dir, "update-ref", local_ref, rt_ref,
                 timeout=15)
            logger.debug("Fast-forward %s from Gitea", local_ref)