refactor: bypass Gitea pre-receive via update-ref + manual post-receive

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:
+    canonical_refs: Dict[str, str] = {}
-        has_new = False
+    for line in (out or "").splitlines():
-        for line in out.splitlines():
+        line = line.strip()
-            line = line.strip()
+        if not line:
-            if not line:
+            continue
-                continue
+        obj_hash, ref = line.split(None, 1)
-            obj_hash, ref = line.split(None, 1)
+        canonical_refs[ref] = obj_hash
-            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)
+    changed: list = []
-    # Use `git push --prune` with explicit refspecs to mirror branches and tags
+    for ref, obj_hash in canonical_refs.items():
-    _git(
+        gitea_hash = _git("--git-dir", str(gitea_path), "rev-parse",
-        "--git-dir", canonical, "push", "--prune",
+                          "--verify", "--quiet", ref,
-        str(gitea_path),
+                          timeout=15, check=False)
-        "+refs/heads/*:refs/heads/*",
+        if obj_hash != gitea_hash:
-        "+refs/tags/*:refs/tags/*",
+            if gitea_hash:
-        timeout=300,
+                _git("--git-dir", str(gitea_path), "update-ref", ref,
-    )
+                     obj_hash, gitea_hash, timeout=15)
-    logger.info("Push to Gitea complete")
+            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: "<old-hash> <new-hash> <refname>\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 ──────────────────────────────────