From 811e3748d4cf218ae44b5fb17256aa51782a5e60 Mon Sep 17 00:00:00 2001 From: LeRatierBretonnier Date: Wed, 24 Jun 2026 13:33:45 +0200 Subject: [PATCH] refactor: bypass Gitea pre-receive via update-ref + manual post-receive --- svn_mirror/gitea.py | 117 ++++++++++++++++++++++++++++++++------------ 1 file changed, 87 insertions(+), 30 deletions(-) diff --git a/svn_mirror/gitea.py b/svn_mirror/gitea.py index e3ffe4d..a8869e4 100644 --- a/svn_mirror/gitea.py +++ b/svn_mirror/gitea.py @@ -4,7 +4,7 @@ import logging import os import subprocess from pathlib import Path -from typing import Optional +from typing import Dict, Optional from .mirror import Mirror @@ -52,9 +52,12 @@ def _git(*args: str, timeout: int = 300, check: bool = True) -> str: def push_to_gitea(mirror: Mirror) -> bool: - """Push all heads and tags from the canonical repo to Gitea's bare repo. + """Sync canonical refs to Gitea's bare repo via direct ref updates. - Returns True if anything was pushed. + Uses ``git update-ref`` to bypass Gitea's pre-receive hook, then + fires Gitea's ``post-receive`` hook manually so the UI refreshes. + + Returns True if anything changed. """ gitea_path = _get_gitea_repo_path(mirror) if not gitea_path.exists(): @@ -63,41 +66,95 @@ def push_to_gitea(mirror: Mirror) -> bool: canonical = str(mirror.canonical_dir) - # Check if there's anything to push by comparing refs + # Collect canonical 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 + canonical_refs: Dict[str, str] = {} + for line in (out or "").splitlines(): + line = line.strip() + if not line: + continue + obj_hash, ref = line.split(None, 1) + canonical_refs[ref] = obj_hash - 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") + changed: list = [] + for ref, obj_hash in canonical_refs.items(): + gitea_hash = _git("--git-dir", str(gitea_path), "rev-parse", + "--verify", "--quiet", ref, + timeout=15, check=False) + if obj_hash != gitea_hash: + if gitea_hash: + _git("--git-dir", str(gitea_path), "update-ref", ref, + obj_hash, gitea_hash, timeout=15) + else: + _git("--git-dir", str(gitea_path), "update-ref", ref, + obj_hash, timeout=15) + changed.append((gitea_hash or "0" * 40, obj_hash, ref)) + logger.debug("Updated %s in Gitea", ref) + + # Prune refs in Gitea not in canonical + gitea_out = _git("--git-dir", str(gitea_path), "for-each-ref", + "--format=%(objectname) %(refname)", + "refs/heads/", "refs/tags/", + timeout=30, check=False) + for line in (gitea_out or "").splitlines(): + line = line.strip() + if not line: + continue + obj_hash, ref = line.split(None, 1) + if ref not in canonical_refs: + _git("--git-dir", str(gitea_path), "update-ref", "-d", ref, + obj_hash, timeout=15) + changed.append((obj_hash, "0" * 40, ref)) + logger.debug("Pruned %s from Gitea", ref) + + if not changed: + logger.debug("No new commits to push to Gitea") + return False + + logger.info("Synced %d ref(s) to Gitea", len(changed)) + _trigger_gitea_post_receive(gitea_path, mirror.config.gitea) return True +def _trigger_gitea_post_receive(gitea_path: Path, gitea_cfg): + """Fire Gitea's post-receive hook so the UI refreshes immediately.""" + # Find the gitea hook binary + candidates = ("/usr/local/bin/gitea", "/usr/bin/gitea", "gitea") + hook_bin = None + for c in candidates: + if subprocess.run(["which", c], capture_output=True, check=False).returncode == 0: + hook_bin = c + break + if not hook_bin: + logger.debug("gitea binary not found – skipping post-receive hook") + return + + try: + config_path = os.environ.get("GITEA_CONFIG", "/etc/gitea/app.ini") + # Build stdin with the ref updates that would normally arrive + # Format: " \n" + data = _git("--git-dir", str(gitea_path), "for-each-ref", + "--format=%(objectname) %(objectname) %(refname)", + "refs/heads/", "refs/tags/", + timeout=15, check=False) + r = subprocess.run( + [hook_bin, "hook", "--config", config_path, "post-receive"], + input=(data or "").encode(), + capture_output=True, + timeout=60, + ) + if r.returncode != 0: + err = (r.stderr or b"").decode(errors="replace").strip() + logger.warning("Gitea post-receive hook failed: %s", err) + else: + logger.debug("Gitea post-receive hook fired") + except Exception as e: + logger.warning("Could not fire Gitea post-receive hook: %s", e) + + # ─── Fetch Gitea → canonical ──────────────────────────────────