from __future__ import annotations import asyncio import datetime import logging import os import subprocess import threading import urllib.parse from pathlib import Path from typing import Callable, Optional from app.config import ( AUTO_GIT_SYNC, GIT_AUTHOR_EMAIL, GIT_AUTHOR_NAME, GIT_BRANCH, GIT_PASSWORD, GIT_REPO_URL, GIT_SYNC_INTERVAL_SECONDS, GIT_USERNAME, WIKI_DIR, ) logger = logging.getLogger("duckdb_backend.git_sync") class GitSyncManager: def __init__(self, wiki_dir: Optional[Path] = None, on_data_pulled: Optional[Callable[[], None]] = None): self.wiki_dir = wiki_dir or WIKI_DIR self.on_data_pulled = on_data_pulled self._lock = threading.Lock() self.status = "ok" self.is_conflict = False self.last_sync_time: Optional[str] = None self.last_commit_hash: Optional[str] = None self.last_error: Optional[str] = None self._bg_task: Optional[asyncio.Task] = None self._stop_event = asyncio.Event() def _configure_git_client(self): """Configures git client with safe.directory and author identity.""" try: subprocess.run(["git", "config", "--global", "http.sslVerify", "false"], check=False) subprocess.run(["git", "config", "--global", "--add", "safe.directory", "*"], check=False) if (self.wiki_dir / ".git").exists(): subprocess.run(["git", "-C", str(self.wiki_dir), "config", "user.name", GIT_AUTHOR_NAME], check=False) subprocess.run(["git", "-C", str(self.wiki_dir), "config", "user.email", GIT_AUTHOR_EMAIL], check=False) except Exception as e: logger.warning(f"Failed to configure git client options: {e}") def get_auth_repo_url(self) -> str: if GIT_USERNAME and GIT_PASSWORD and "://" in GIT_REPO_URL: proto, rest = GIT_REPO_URL.split("://", 1) # Remove existing credentials in URL if any if "@" in rest: rest = rest.split("@", 1)[1] user_enc = urllib.parse.quote(GIT_USERNAME, safe="") pass_enc = urllib.parse.quote(GIT_PASSWORD, safe="") return f"{proto}://{user_enc}:{pass_enc}@{rest}" return GIT_REPO_URL def ensure_repo(self) -> bool: """Ensures repository exists and is initialized.""" with self._lock: self._configure_git_client() if (self.wiki_dir / ".git").exists(): try: res = subprocess.run( ["git", "-C", str(self.wiki_dir), "rev-parse", "--short", "HEAD"], capture_output=True, text=True, check=False, ) if res.returncode == 0: self.last_commit_hash = res.stdout.strip() return True except Exception as e: logger.warning(f"Failed to check git rev: {e}") return True if not GIT_REPO_URL: logger.info(f"Wiki directory {self.wiki_dir} is not a git repo, but no GIT_REPO_URL configured.") return False logger.info(f"Initializing repository {GIT_REPO_URL} in {self.wiki_dir}...") self.wiki_dir.mkdir(parents=True, exist_ok=True) auth_url = self.get_auth_repo_url() try: # Use git init + remote + fetch + checkout to safely support non-empty directories (e.g. lost+found on PVC) subprocess.run(["git", "-C", str(self.wiki_dir), "init"], capture_output=True, check=False) self._configure_git_client() subprocess.run( ["git", "-C", str(self.wiki_dir), "remote", "add", "origin", auth_url], capture_output=True, check=False, ) subprocess.run( ["git", "-C", str(self.wiki_dir), "remote", "set-url", "origin", auth_url], capture_output=True, check=False, ) res = subprocess.run( ["git", "-C", str(self.wiki_dir), "fetch", "origin", GIT_BRANCH], capture_output=True, text=True, check=False, timeout=60, ) if res.returncode == 0: if (self.wiki_dir / ".git" / "shallow").exists(): subprocess.run( ["git", "-C", str(self.wiki_dir), "fetch", "--unshallow", "origin", GIT_BRANCH], capture_output=True, check=False, ) checkout_res = subprocess.run( ["git", "-C", str(self.wiki_dir), "checkout", "-f", "-B", GIT_BRANCH, f"origin/{GIT_BRANCH}"], capture_output=True, text=True, check=False, ) if checkout_res.returncode == 0: logger.info("Successfully initialized and checked out wiki repository.") self.status = "ok" self.is_conflict = False self.last_sync_time = datetime.datetime.now(datetime.timezone.utc).isoformat() return True else: self.status = "error" self.last_error = f"Checkout failed: {checkout_res.stderr.strip()}" logger.error(self.last_error) return False else: self.status = "error" self.last_error = f"Fetch failed: {res.stderr.strip()}" logger.error(self.last_error) return False except Exception as e: self.status = "error" self.last_error = f"Clone exception: {e}" logger.error(self.last_error) return False def pull_rebase(self) -> bool: """Pulls latest changes via rebase. Handles conflicts cleanly.""" if not (self.wiki_dir / ".git").exists(): return False with self._lock: try: # Update remote url with credentials if provided auth_url = self.get_auth_repo_url() subprocess.run( ["git", "-C", str(self.wiki_dir), "remote", "set-url", "origin", auth_url], capture_output=True, check=False, ) # Fetch and rebase with autostash res = subprocess.run( ["git", "-C", str(self.wiki_dir), "pull", "--rebase", "--autostash", "origin", GIT_BRANCH], capture_output=True, text=True, check=False, timeout=30, ) if res.returncode == 0: self.status = "ok" self.is_conflict = False self.last_error = None self.last_sync_time = datetime.datetime.now(datetime.timezone.utc).isoformat() # Update commit hash rev = subprocess.run( ["git", "-C", str(self.wiki_dir), "rev-parse", "--short", "HEAD"], capture_output=True, text=True, check=False, ) if rev.returncode == 0: self.last_commit_hash = rev.stdout.strip() if self.on_data_pulled: try: self.on_data_pulled() except Exception as ex: logger.error(f"Error in on_data_pulled callback: {ex}") return True else: # Check if conflict occurred stderr = res.stderr or res.stdout logger.warning(f"Git pull --rebase failed: {stderr}") # Abort any in-progress rebase to keep working tree stable subprocess.run( ["git", "-C", str(self.wiki_dir), "rebase", "--abort"], capture_output=True, check=False, ) self.is_conflict = True self.status = "conflict" self.last_error = f"Rebase conflict: {stderr.strip()}" return False except Exception as e: self.status = "error" self.last_error = f"Pull error: {e}" logger.error(self.last_error) return False def commit_and_push(self, commit_msg: str) -> bool: """Stages data/, commits with descriptive message, rebases and pushes.""" if not (self.wiki_dir / ".git").exists(): return False with self._lock: try: self._configure_git_client() data_rel = "data" subprocess.run( ["git", "-C", str(self.wiki_dir), "add", data_rel], capture_output=True, check=False, ) commit_res = subprocess.run( ["git", "-C", str(self.wiki_dir), "commit", "-m", f"{commit_msg} [skip ci]"], capture_output=True, text=True, check=False, ) if commit_res.returncode != 0: # Nothing to commit or already committed logger.debug(f"Commit returncode {commit_res.returncode}: {commit_res.stdout.strip()}") return True # Pull rebase with autostash before push auth_url = self.get_auth_repo_url() subprocess.run( ["git", "-C", str(self.wiki_dir), "remote", "set-url", "origin", auth_url], capture_output=True, check=False, ) rebase_res = subprocess.run( ["git", "-C", str(self.wiki_dir), "pull", "--rebase", "--autostash", "origin", GIT_BRANCH], capture_output=True, text=True, check=False, timeout=30, ) if rebase_res.returncode != 0: stderr = rebase_res.stderr or rebase_res.stdout logger.warning(f"Rebase failed during commit_and_push: {stderr}") subprocess.run( ["git", "-C", str(self.wiki_dir), "rebase", "--abort"], capture_output=True, check=False, ) self.is_conflict = True self.status = "conflict" self.last_error = f"Rebase conflict on push: {stderr.strip()}" return False # Push to remote push_res = subprocess.run( ["git", "-C", str(self.wiki_dir), "push", "origin", GIT_BRANCH], capture_output=True, text=True, check=False, timeout=30, ) if push_res.returncode == 0: self.status = "ok" self.is_conflict = False self.last_error = None self.last_sync_time = datetime.datetime.now(datetime.timezone.utc).isoformat() rev = subprocess.run( ["git", "-C", str(self.wiki_dir), "rev-parse", "--short", "HEAD"], capture_output=True, text=True, check=False, ) if rev.returncode == 0: self.last_commit_hash = rev.stdout.strip() return True else: self.status = "error" self.last_error = f"Push error: {push_res.stderr.strip()}" logger.error(self.last_error) return False except Exception as e: self.status = "error" self.last_error = f"Commit/push error: {e}" logger.error(self.last_error) return False async def start_periodic_sync(self): """Starts background periodic git pull loop.""" if not AUTO_GIT_SYNC: logger.info("Automatic git sync is disabled by configuration.") return self._stop_event.clear() async def _loop(): logger.info(f"Starting periodic git sync every {GIT_SYNC_INTERVAL_SECONDS}s") while not self._stop_event.is_set(): try: await asyncio.to_thread(self.pull_rebase) except Exception as ex: logger.error(f"Error during periodic git pull: {ex}") try: await asyncio.wait_for(self._stop_event.wait(), timeout=GIT_SYNC_INTERVAL_SECONDS) except asyncio.TimeoutError: pass self._bg_task = asyncio.create_task(_loop()) async def stop_periodic_sync(self): """Stops background sync loop.""" self._stop_event.set() if self._bg_task: self._bg_task.cancel() try: await self._bg_task except asyncio.CancelledError: pass self._bg_task = None