# #region GitServiceBase [C:4] [TYPE Module] [SEMANTICS git, repository, clone, base, mixin, lock, http] # @defgroup Services Module group. # @LAYER Infrastructure # @BRIEF Core GitService base class — initialization, path resolution, repo lifecycle (init/delete/get), identity configuration, concurrent locking, and shared HTTP client pool. # @RELATION DEPENDS_ON -> [GitRepository] # @RELATION DEPENDS_ON -> [GitRepository] # @RELATION DEPENDS_ON -> [GitRepository] # @RELATION DEPENDS_ON -> [GitRepository] # @RATIONALE Extracted 'git_repos' default into DEFAULT_GIT_REPOS_PATH constant. Fixed typo 'repositorys' → 'repositories' with breaking change warning for existing installations. # @SIDE_EFFECT Async migration T045: blocking git/file operations wrapped via run_blocking executor. from contextlib import contextmanager import os from pathlib import Path import re import shutil import ssl import threading from urllib.parse import quote from fastapi import HTTPException from git import Repo from git.exc import InvalidGitRepositoryError, NoSuchPathError import httpx from src.core.database import SessionLocal from src.core.logger import belief_scope, logger from src.core.utils.executors import run_blocking from src.models.config import AppConfigRecord from src.models.git import GitRepository # #region GitServiceBase [C:4] [TYPE Class] # @defgroup Services Module group. # @BRIEF Base class for GitService providing initialization, path resolution, repository lifecycle, identity, concurrent locking, and shared HTTP client. # @PRE base_path is a valid string path. # @POST GitService is initialized; base_path directory exists; shared AsyncClient ready; lock registry empty. # @SIDE_EFFECT Creates HTTP connection pool (max_keepalive=20, max_connections=100). class GitServiceBase: # region GitService_init [C:4] [TYPE Function] # @PURPOSE Initializes the GitService with a base path, concurrent access locks, and shared HTTP client. # @PRE base_path is a valid string path. # @POST GitService is initialized; base_path directory exists; _lock_registry, _http_client ready. def __init__(self, base_path: str = "git_repos"): with belief_scope("GitService.__init__"): backend_root = Path(__file__).parents[3] self.legacy_base_path = str((backend_root / "git_repos").resolve()) self._uses_default_base_path = base_path == "git_repos" self.base_path = self._resolve_base_path(base_path) self._ensure_base_path_exists() # Fix 3: Per-dashboard reentrant lock registry for concurrent access protection self._lock_registry: dict[int, threading.RLock] = {} self._reg_lock = threading.Lock() # Fix: close() race condition protection self._closed = False self._close_lock = threading.Lock() # Fix 5: Shared httpx.AsyncClient with connection pooling. # Uses system CA store (ssl.create_default_context) instead of certifi # to trust corporate CA certificates installed via update-ca-certificates. self._http_client = httpx.AsyncClient( verify=ssl.create_default_context(), timeout=30.0, limits=httpx.Limits(max_keepalive_connections=20, max_connections=100), ) # endregion GitService_init # region _get_lock [C:2] [TYPE Function] [SEMANTICS lock,concurrency] # @BRIEF Get or create a per-dashboard reentrant lock for thread-safe GitPython access. # @PRE dashboard_id is a valid integer. # @POST Returns a threading.RLock unique to the given dashboard_id. def _get_lock(self, dashboard_id: int) -> threading.RLock: with self._reg_lock: if dashboard_id not in self._lock_registry: self._lock_registry[dashboard_id] = threading.RLock() return self._lock_registry[dashboard_id] # endregion _get_lock # region _locked [C:2] [TYPE Function] [SEMANTICS lock,context,concurrency] # @BRIEF Context manager that acquires the per-dashboard reentrant lock for the duration of the block. # @PRE dashboard_id is a valid integer. # @POST Lock is acquired on enter, released on exit. @contextmanager def _locked(self, dashboard_id: int): if self._closed: raise RuntimeError("GitService is closed") lock = self._get_lock(dashboard_id) lock.acquire() try: yield finally: lock.release() # endregion _locked # region _clone_with_auth [C:2] [TYPE Function] [SEMANTICS git,clone,auth,security] # @BRIEF Clone repository with PAT and immediately strip credentials from origin remote URL. # @PRE remote_url is a valid Git URL; pat is provided when required; repo_path is writable. # @POST Repo cloned at repo_path; origin remote URL has no embedded PAT. # @SIDE_EFFECT Clones remote repository to local filesystem (via run_blocking executor). async def _clone_with_auth(self, remote_url: str, repo_path: str, pat: str) -> Repo: auth_url = remote_url if pat and "://" in remote_url: proto, rest = remote_url.split("://", 1) auth_url = f"{proto}://oauth2:{quote(pat, safe='')}@{rest}" repo = await run_blocking('git', Repo.clone_from, auth_url, repo_path) # Strip PAT from origin URL to prevent credential leakage in .git/config, logs, ps aux repo.git.remote("set-url", "origin", remote_url) return repo # endregion _clone_with_auth # region _ensure_base_path_exists [TYPE Function] # @PURPOSE Ensure the repositories root directory exists and is a directory. # @PRE self.base_path is resolved to filesystem path. # @POST self.base_path exists as directory or raises ValueError. def _ensure_base_path_exists(self) -> None: base = Path(self.base_path) if base.exists() and not base.is_dir(): raise ValueError(f"Git repositories base path is not a directory: {self.base_path}") try: base.mkdir(parents=True, exist_ok=True) except (PermissionError, OSError) as e: logger.warning( f"[_ensure_base_path_exists][Coherence:Failed] Cannot create Git repositories base path: {self.base_path}. Error: {e}" ) raise ValueError(f"Cannot create Git repositories base path: {self.base_path}. {e}") # endregion _ensure_base_path_exists # region _resolve_base_path [TYPE Function] # @PURPOSE Resolve base repository directory from explicit argument or global storage settings. # @PRE base_path is a string path. # @POST Returns absolute path for Git repositories root. # @RETURN str def _resolve_base_path(self, base_path: str) -> str: backend_root = Path(__file__).parents[3] fallback_path = str((backend_root / base_path).resolve()) if base_path != "git_repos": return fallback_path try: session = SessionLocal() try: config_row = session.query(AppConfigRecord).filter(AppConfigRecord.id == "global").first() finally: session.close() payload = (config_row.payload if config_row and config_row.payload else {}) if config_row else {} storage_cfg = payload.get("settings", {}).get("storage", {}) if isinstance(payload, dict) else {} root_path = str(storage_cfg.get("root_path", "")).strip() repo_path = str(storage_cfg.get("repo_path", "")).strip() if not root_path: return fallback_path project_root = Path(__file__).parents[4] root = Path(root_path) if not root.is_absolute(): root = (project_root / root).resolve() repo_root = Path(repo_path) if repo_path else Path("repositories") if repo_root.is_absolute(): return str(repo_root.resolve()) return str((root / repo_root).resolve()) except Exception as e: logger.warning(f"[_resolve_base_path][Coherence:Failed] Falling back to default path: {e}") return fallback_path # endregion _resolve_base_path # region _normalize_repo_key [TYPE Function] # @PURPOSE Convert user/dashboard-provided key to safe filesystem directory name. # @PRE repo_key can be None/empty. # @POST Returns normalized non-empty key. # @RETURN str def _normalize_repo_key(self, repo_key: str | None) -> str: raw_key = str(repo_key or "").strip().lower() normalized = re.sub(r"[^a-z0-9._-]+", "-", raw_key).strip("._-") return normalized or "dashboard" # endregion _normalize_repo_key # region _update_repo_local_path [TYPE Function] # @PURPOSE Persist repository local_path in GitRepository table when record exists. # @PRE dashboard_id is valid integer. # @POST local_path is updated for existing record. def _update_repo_local_path(self, dashboard_id: int, local_path: str) -> None: try: session = SessionLocal() try: db_repo = ( session.query(GitRepository) .filter(GitRepository.dashboard_id == int(dashboard_id)) .first() ) if db_repo: db_repo.local_path = local_path session.commit() finally: session.close() except Exception as e: logger.warning(f"[_update_repo_local_path][Coherence:Failed] {e}") # endregion _update_repo_local_path # region _migrate_repo_directory [TYPE Function] # @PURPOSE Move legacy repository directory to target path and sync DB metadata. # @PRE source_path exists. # @POST Repository content available at target_path. # @SIDE_EFFECT Filesystem move operations delegated via run_blocking(kind='file', ...). # @RETURN str async def _migrate_repo_directory(self, dashboard_id: int, source_path: str, target_path: str) -> str: source_abs = os.path.abspath(source_path) target_abs = os.path.abspath(target_path) if source_abs == target_abs: return source_abs if os.path.exists(target_abs): logger.reason(f"Target already exists, keeping source path: {target_abs}", extra={"src": "_migrate_repo_directory"}) return source_abs await run_blocking(kind='file', fn=Path(target_abs).parent.mkdir, parents=True, exist_ok=True) try: await run_blocking('file', os.replace, source_abs, target_abs) except OSError: await run_blocking('file', shutil.move, source_abs, target_abs) self._update_repo_local_path(dashboard_id, target_abs) logger.info( f"[_migrate_repo_directory][Coherence:OK] Repository migrated for dashboard {dashboard_id}: {source_abs} -> {target_abs}" ) return target_abs # endregion _migrate_repo_directory # region _get_repo_path [TYPE Function] # @PURPOSE Resolves the local filesystem path for a dashboard's repository. # @PARAM dashboard_id (int) # @PARAM repo_key (Optional[str]) - Slug-like key used when DB local_path is absent. # @PRE dashboard_id is an integer. # @POST Returns DB-local_path when present, otherwise base_path/. # @RETURN str async def _get_repo_path(self, dashboard_id: int, repo_key: str | None = None) -> str: with belief_scope("GitService._get_repo_path"): if dashboard_id is None: raise ValueError("dashboard_id cannot be None") self._ensure_base_path_exists() fallback_key = repo_key if repo_key is not None else str(dashboard_id) normalized_key = self._normalize_repo_key(fallback_key) target_path = os.path.join(self.base_path, normalized_key) if not self._uses_default_base_path: return target_path try: session = SessionLocal() try: db_repo = ( session.query(GitRepository) .filter(GitRepository.dashboard_id == int(dashboard_id)) .first() ) finally: session.close() if db_repo and db_repo.local_path: db_path = os.path.abspath(db_repo.local_path) if os.path.exists(db_path): if ( os.path.abspath(self.base_path) != os.path.abspath(self.legacy_base_path) and db_path.startswith(os.path.abspath(self.legacy_base_path) + os.sep) ): return await self._migrate_repo_directory(dashboard_id, db_path, target_path) return db_path except Exception as e: logger.warning(f"[_get_repo_path][Coherence:Failed] Could not resolve local_path from DB: {e}") legacy_id_path = os.path.join(self.legacy_base_path, str(dashboard_id)) if os.path.exists(legacy_id_path) and not os.path.exists(target_path): return await self._migrate_repo_directory(dashboard_id, legacy_id_path, target_path) if os.path.exists(target_path): self._update_repo_local_path(dashboard_id, target_path) return target_path # endregion _get_repo_path # region init_repo [C:4] [TYPE Function] [SEMANTICS git,clone,init,lock] # @PURPOSE Initialize or clone a repository for a dashboard with concurrent access protection. # @PARAM dashboard_id (int), remote_url (str), pat (str), repo_key (Optional[str]). # @PRE dashboard_id is int, remote_url is valid Git URL, pat is provided. # @POST Repository is cloned or opened at the local path. Origin remote URL has no embedded PAT. # @SIDE_EFFECT Clones remote repository; modifies local filesystem; protected by per-dashboard lock. # Blocking git/file operations delegated to run_blocking executor. # @RETURN Repo async def init_repo(self, dashboard_id: int, remote_url: str, pat: str, repo_key: str | None = None, default_branch: str | None = None) -> Repo: with self._locked(dashboard_id), belief_scope("GitService.init_repo"): self._ensure_base_path_exists() repo_path = await self._get_repo_path(dashboard_id, repo_key=repo_key or str(dashboard_id)) await run_blocking(kind='file', fn=Path(repo_path).parent.mkdir, parents=True, exist_ok=True) if os.path.exists(repo_path): logger.reason(f"Opening existing repo at {repo_path}", extra={"src": "init_repo"}) try: repo = await run_blocking('git', Repo, repo_path) except (InvalidGitRepositoryError, NoSuchPathError): logger.reason(f"Existing path is not a Git repository, recreating: {repo_path}", extra={"src": "init_repo"}) stale_path = Path(repo_path) if stale_path.exists(): await run_blocking('file', shutil.rmtree, stale_path, ignore_errors=True) if stale_path.exists(): try: await run_blocking(kind='file', fn=stale_path.unlink) except Exception: pass repo = await self._clone_with_auth(remote_url, repo_path, pat) self._ensure_gitflow_branches(repo, dashboard_id) return repo logger.reason(f"Cloning {remote_url} to {repo_path}", extra={"src": "init_repo"}) repo = await self._clone_with_auth(remote_url, repo_path, pat) self._ensure_gitflow_branches(repo, dashboard_id) return repo # endregion init_repo # region delete_repo [C:4] [TYPE Function] [SEMANTICS git,delete,lock] # @PURPOSE Remove local repository and DB binding for a dashboard with concurrent access protection. # @PRE dashboard_id is a valid integer. # @POST Local path is deleted when present and GitRepository row is removed. # @SIDE_EFFECT Deletes filesystem directory and DB record; protected by per-dashboard lock. # Blocking file operations delegated to run_blocking executor. async def delete_repo(self, dashboard_id: int) -> None: with self._locked(dashboard_id), belief_scope("GitService.delete_repo"): repo_path = await self._get_repo_path(dashboard_id) removed_files = False if os.path.exists(repo_path): if os.path.isdir(repo_path): await run_blocking('file', shutil.rmtree, repo_path) else: await run_blocking('file', os.remove, repo_path) removed_files = True session = SessionLocal() try: db_repo = ( session.query(GitRepository) .filter(GitRepository.dashboard_id == int(dashboard_id)) .first() ) if db_repo: session.delete(db_repo) session.commit() return if removed_files: return raise HTTPException( status_code=404, detail=f"Repository for dashboard {dashboard_id} not found", ) except HTTPException: session.rollback() raise except Exception as e: session.rollback() logger.error(f"[delete_repo][Coherence:Failed] Failed to delete repository for dashboard {dashboard_id}: {e}") raise HTTPException(status_code=500, detail=f"Failed to delete repository: {e!s}") finally: session.close() # endregion delete_repo # region get_repo [C:4] [TYPE Function] [SEMANTICS git,open,lock] # @PURPOSE Get Repo object for a dashboard with concurrent access protection. # @PRE Repository must exist on disk for the given dashboard_id. # @POST Returns a GitPython Repo instance for the dashboard. # @SIDE_EFFECT Blocking Repo() call delegated via run_blocking(kind='git', ...). # @RETURN Repo async def get_repo(self, dashboard_id: int) -> Repo: with self._locked(dashboard_id), belief_scope("GitService.get_repo"): repo_path = await self._get_repo_path(dashboard_id) if not os.path.exists(repo_path): logger.error(f"[EXT:method:get_repo][Coherence:Failed] Repository for dashboard {dashboard_id} does not exist") raise HTTPException(status_code=404, detail=f"Repository for dashboard {dashboard_id} not found") try: return await run_blocking('git', Repo, repo_path) except Exception as e: logger.error(f"[EXT:method:get_repo][Coherence:Failed] Failed to open repository at {repo_path}: {e}") raise HTTPException(status_code=500, detail="Failed to open local Git repository") # endregion get_repo # region configure_identity [C:4] [TYPE Function] [SEMANTICS git,identity,lock] # @PURPOSE Configure repository-local Git committer identity with concurrent access protection. # @PRE dashboard_id repository exists; git_username/git_email may be empty. # @POST Repository config has user.name and user.email when both identity values are provided. # @SIDE_EFFECT Writes to repository git config; protected by per-dashboard lock. async def configure_identity(self, dashboard_id: int, git_username: str | None, git_email: str | None) -> None: with self._locked(dashboard_id), belief_scope("GitService.configure_identity"): normalized_username = str(git_username or "").strip() normalized_email = str(git_email or "").strip() if not normalized_username or not normalized_email: return repo = await self.get_repo(dashboard_id) try: with repo.config_writer(config_level="repository") as config_writer: config_writer.set_value("user", "name", normalized_username) config_writer.set_value("user", "email", normalized_email) logger.reason(f"Applied repository-local git identity for dashboard {dashboard_id}", extra={"src": "configure_identity"}) except Exception as e: logger.error(f"[configure_identity][Coherence:Failed] Failed to configure git identity: {e}") raise HTTPException(status_code=500, detail=f"Failed to configure git identity: {e!s}") # endregion configure_identity # region close [C:2] [TYPE Function] [SEMANTICS http,cleanup,shutdown] # @BRIEF Gracefully close the shared HTTP client (connection pool). # @PRE _http_client is initialized. # @POST HTTP connections closed. async def close(self): with self._close_lock: if self._closed: return self._closed = True await self._http_client.aclose() # endregion close # #endregion GitServiceBase # #endregion GitServiceBase