ss-tools/backend/src/services/git/_sync.py

# #region GitServiceSyncMixin [C:4] [TYPE Module] [SEMANTICS git, sync, push, pull, remote, lock]
# @defgroup Services Module group.
# @LAYER Infrastructure
# @BRIEF Push and pull operations for GitService with origin host auto-alignment (concurrent-safe).
# @RELATION CALLED_BY -> [GitService]
# @RELATION DEPENDS_ON -> [GitServiceUrlMixin]

import os
import re

from fastapi import HTTPException
from git.exc import GitCommandError

from src.core.database import SessionLocal
from src.core.logger import belief_scope, logger
from src.models.git import GitRepository, GitServerConfig


# #region GitServiceSyncMixin [C:3] [TYPE Class]
# @defgroup Services Module group.
# @BRIEF Mixin providing push and pull operations with origin host alignment.
class GitServiceSyncMixin:
    @staticmethod
    def _redact_pat_from_message(message: str, pat: str | None = None) -> str:
        redacted = str(message or "")
        if pat:
            redacted = redacted.replace(pat, "***")
        return re.sub(r"(https?://[^\s:/@]+:)[^\s@]+(@)", r"\1***\2", redacted)

    # region push_changes [C:4] [TYPE Function] [SEMANTICS git,push,lock]
    # @PURPOSE: Push local commits to remote (concurrent-safe).
    # @PRE  Repository exists and has an 'origin' remote.
    # @POST  Local branch commits are pushed to origin.
    # region _embed_pat_in_origin_url [C:2] [TYPE Function] [SEMANTICS git,auth,pat,url]
    # @BRIEF Temporarily embed a personal access token into the origin remote URL.
    # @POST  Origin URL is updated with embedded PAT; returns the original URL for restoration.
    @staticmethod
    def _embed_pat_in_origin_url(repo, pat: str) -> str | None:
        from urllib.parse import urlparse, urlunparse, quote

        try:
            origin = repo.remote(name="origin")
            original_url = list(origin.urls)[0]
        except Exception:
            return None
        parsed = urlparse(original_url)
        if parsed.scheme not in ("https", "http"):
            return None
        username = parsed.username or "git-user"
        encoded_pat = quote(pat, safe="")
        auth_part = f"{quote(username, safe='')}:{encoded_pat}@"
        host_part = parsed.hostname or ""
        if parsed.port:
            host_part = f"{host_part}:{parsed.port}"
        new_url = urlunparse(
            (parsed.scheme, f"{auth_part}{host_part}", parsed.path, parsed.params, parsed.query, parsed.fragment)
        )
        try:
            origin.set_url(new_url)
            return original_url
        except Exception:
            return None
    # endregion _embed_pat_in_origin_url

    async def push_changes(self, dashboard_id: int, pat: str | None = None):
        with self._locked(dashboard_id):
            with belief_scope("GitService.push_changes"):
                repo = await self.get_repo(dashboard_id)
            if not repo.heads:
                logger.explore("No local branches to push", extra={"src": "push_changes"}, payload={"dashboard_id": dashboard_id})
                return
            try:
                origin = repo.remote(name='origin')
            except ValueError:
                logger.explore("Remote 'origin' not found", extra={"src": "push_changes"}, payload={"dashboard_id": dashboard_id})
                raise HTTPException(status_code=400, detail="Remote 'origin' not configured")
            try:
                origin_urls = list(origin.urls)
            except Exception:
                origin_urls = []
            binding_remote_url = None
            binding_config_id = None
            binding_config_url = None
            try:
                session = SessionLocal()
                try:
                    db_repo = (
                        session.query(GitRepository)
                        .filter(GitRepository.dashboard_id == int(dashboard_id))
                        .first()
                    )
                    if db_repo:
                        binding_remote_url = db_repo.remote_url
                        binding_config_id = db_repo.config_id
                        db_config = (
                            session.query(GitServerConfig)
                            .filter(GitServerConfig.id == db_repo.config_id)
                            .first()
                        )
                        if db_config:
                            binding_config_url = db_config.url
                finally:
                    session.close()
            except Exception as diag_error:
                logger.reason(
                    "Failed to load repository binding diagnostics",
                    extra={"src": "push_changes", "dashboard_id": dashboard_id, "error": str(diag_error)},
                )
            realigned_origin_url = self._align_origin_host_with_config(
                dashboard_id=dashboard_id,
                origin=origin,
                config_url=binding_config_url,
                current_origin_url=(origin_urls[0] if origin_urls else None),
                binding_remote_url=binding_remote_url,
            )
            try:
                origin_urls = list(origin.urls)
            except Exception:
                origin_urls = []
            logger.reason(
                f"Push diagnostics dashboard={dashboard_id} config_id={binding_config_id} config_url={binding_config_url} binding_remote_url={binding_remote_url} origin_urls={origin_urls} origin_realigned={bool(realigned_origin_url)}",
                extra={"src": "push_changes"},
            )
            # ── Embed PAT credential if provided ──────────────────────────────
            _original_push_url: str | None = None
            if pat:
                _original_push_url = self._embed_pat_in_origin_url(repo, pat)
            # ──────────────────────────────────────────────────────────────────
            try:
                try:
                    current_branch = repo.active_branch
                    logger.reason(f"Pushing branch {current_branch.name} to origin", extra={"src": "push_changes"})
                    tracking_branch = None
                    try:
                        tracking_branch = current_branch.tracking_branch()
                    except Exception:
                        tracking_branch = None
                    if tracking_branch is None:
                        repo.git.push("--set-upstream", "origin", f"{current_branch.name}:{current_branch.name}")
                    else:
                        push_info = origin.push(refspec=f'{current_branch.name}:{current_branch.name}')
                        for info in push_info:
                            if info.flags & info.ERROR:
                                safe_summary = self._redact_pat_from_message(str(info.summary), pat)
                                logger.explore("Error pushing ref", extra={"src": "push_changes"}, payload={"ref": info.remote_ref_string}, error=safe_summary)
                                raise Exception(f"Git push error for {info.remote_ref_string}: {safe_summary}")
                except GitCommandError as e:
                    details = self._redact_pat_from_message(str(e), pat)
                    lowered = details.lower()
                    if "non-fast-forward" in lowered or "rejected" in lowered:
                        raise HTTPException(
                            status_code=409,
                            detail="Push rejected: remote branch contains newer commits. Run Pull first, resolve conflicts if any, then push again.",
                        )
                    logger.explore("Failed to push changes", extra={"src": "push_changes"}, error=details)
                    raise HTTPException(status_code=500, detail=f"Git push failed: {details}")
                except Exception as e:
                    details = self._redact_pat_from_message(str(e), pat)
                    logger.explore("Failed to push changes", extra={"src": "push_changes"}, error=details)
                    raise HTTPException(status_code=500, detail=f"Git push failed: {details}")
            finally:
                if _original_push_url is not None:
                    try:
                        origin.set_url(_original_push_url)
                    except Exception:
                        pass
    # endregion push_changes

    # region pull_changes [C:4] [TYPE Function] [SEMANTICS git,pull,lock]
    # @PURPOSE: Pull changes from remote (concurrent-safe).
    # @PRE  Repository exists and has an 'origin' remote.
    # @POST  Changes from origin are pulled and merged into the active branch.
    async def pull_changes(self, dashboard_id: int, pat: str | None = None):
        with self._locked(dashboard_id):
            with belief_scope("GitService.pull_changes"):
                repo = await self.get_repo(dashboard_id)
            merge_head_path = os.path.join(repo.git_dir, "MERGE_HEAD")
            if os.path.exists(merge_head_path):
                payload = self._build_unfinished_merge_payload(repo)
                logger.reason(
                    f"Unfinished merge detected for dashboard {dashboard_id} (repo_path={payload['repository_path']} git_dir={payload['git_dir']} branch={payload['current_branch']} merge_head={payload['merge_head']} merge_msg={payload['merge_message_preview']})",
                    extra={"src": "pull_changes"},
                )
                raise HTTPException(status_code=409, detail=payload)
            # ── Embed PAT credential if provided ──────────────────────────────
            _original_pull_url: str | None = None
            if pat:
                _original_pull_url = self._embed_pat_in_origin_url(repo, pat)
            # ──────────────────────────────────────────────────────────────────
            origin = None
            try:
                try:
                    origin = repo.remote(name='origin')
                    current_branch = repo.active_branch.name
                    try:
                        origin_urls = [self._redact_pat_from_message(url, pat) for url in origin.urls]
                    except Exception:
                        origin_urls = []
                    logger.reason(
                        f"Pull diagnostics dashboard={dashboard_id} repo_path={repo.working_tree_dir} branch={current_branch} origin_urls={origin_urls}",
                        extra={"src": "pull_changes"},
                    )
                    origin.fetch(prune=True)
                    remote_ref = f"origin/{current_branch}"
                    has_remote_branch = any(ref.name == remote_ref for ref in repo.refs)
                    logger.reason(
                        f"Pull remote branch check dashboard={dashboard_id} branch={current_branch} remote_ref={remote_ref} exists={has_remote_branch}",
                        extra={"src": "pull_changes"},
                    )
                    if not has_remote_branch:
                        raise HTTPException(
                            status_code=409,
                            detail=f"Remote branch '{current_branch}' does not exist yet. Push this branch first.",
                        )
                    logger.reason(f"Pulling changes from origin/{current_branch}", extra={"src": "pull_changes"})
                    repo.git.pull("--no-rebase", "origin", current_branch)
                except ValueError:
                    logger.explore("Remote 'origin' not found", extra={"src": "pull_changes"}, payload={"dashboard_id": dashboard_id})
                    raise HTTPException(status_code=400, detail="Remote 'origin' not configured")
                except GitCommandError as e:
                    details = self._redact_pat_from_message(str(e), pat)
                    lowered = details.lower()
                    if "conflict" in lowered or "not possible to fast-forward" in lowered:
                        raise HTTPException(
                            status_code=409,
                            detail="Pull requires conflict resolution. Resolve conflicts in repository and repeat operation.",
                        )
                    logger.explore("Failed to pull changes", extra={"src": "pull_changes"}, error=details)
                    raise HTTPException(status_code=500, detail=f"Git pull failed: {details}")
                except HTTPException:
                    raise
                except Exception as e:
                    details = self._redact_pat_from_message(str(e), pat)
                    logger.explore("Failed to pull changes", extra={"src": "pull_changes"}, error=details)
                    raise HTTPException(status_code=500, detail=f"Git pull failed: {details}")
            finally:
                if _original_pull_url is not None and origin is not None:
                    try:
                        origin.set_url(_original_pull_url)
                    except Exception:
                        pass
    # endregion pull_changes
# #endregion GitServiceSyncMixin
# #endregion GitServiceSyncMixin