headquarter/apps/api/src/services/config/resolver_service.py

"""Config profile resolver service functions."""

import logging
import os
import subprocess
import uuid
from typing import Any

from fastapi import HTTPException, status
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from src.models import ConfigProfile, SSHKey, UserConfig
from src.services.shared.ssh_keys import _get_fernet
from src.utils.git_url_parser import parse_git_url
from src.schemas.config import ValidateGitUrlResponse

logger = logging.getLogger(__name__)


async def resolve_default_profile(
    session: AsyncSession,
    user_id: uuid.UUID,
    project_id: uuid.UUID,
    tool_type_id: uuid.UUID,
) -> dict:
    """Resolve the default config profile for a project/tool combination."""
    query = (
        select(ConfigProfile)
        .where(ConfigProfile.user_id == user_id)
        .where(
            (ConfigProfile.project_id.is_(None) & ConfigProfile.tool_type_id.is_(None))
            | (ConfigProfile.project_id == project_id)
            | (ConfigProfile.tool_type_id == tool_type_id)
            | (
                (ConfigProfile.project_id == project_id)
                & (ConfigProfile.tool_type_id == tool_type_id)
            )
        )
        .order_by(ConfigProfile.created_at)
    )
    result = await session.execute(query)
    profiles = result.scalars().all()

    if not profiles:
        return {"profile_id": None, "profile_name": None}

    explicit_defaults = [p for p in profiles if p.is_default]

    for p in explicit_defaults:
        if p.project_id == project_id and p.tool_type_id == tool_type_id:
            return {"profile_id": str(p.id), "profile_name": p.name}

    for p in explicit_defaults:
        if p.project_id == project_id and p.tool_type_id is None:
            return {"profile_id": str(p.id), "profile_name": p.name}

    for p in explicit_defaults:
        if p.project_id is None and p.tool_type_id == tool_type_id:
            return {"profile_id": str(p.id), "profile_name": p.name}

    for p in explicit_defaults:
        if p.project_id is None and p.tool_type_id is None:
            return {"profile_id": str(p.id), "profile_name": p.name}

    first = profiles[0]
    return {"profile_id": str(first.id), "profile_name": first.name}


async def validate_git_url(
    session: AsyncSession,
    current_user_id: uuid.UUID,
    url: str,
    ssh_key_id: str | None,
) -> ValidateGitUrlResponse:
    """Validate a git remote URL and list available branches."""
    parse_result = parse_git_url(url)
    original_url = url.strip()
    url_to_check = parse_result.get("base_url") or original_url

    if not url_to_check:
        return ValidateGitUrlResponse(
            valid=False,
            error=parse_result.get("message", "Invalid URL"),
            error_code=parse_result.get("error_code", "INVALID_URL"),
        )

    if parse_result.get("needs_parsing") and url_to_check != original_url:
        return ValidateGitUrlResponse(
            valid=False,
            suggested_url=url_to_check,
            error=parse_result.get("message"),
            error_code=parse_result.get("error_code", "URL_NEEDS_PARSING"),
        )

    env = None
    key_path = None
    if ssh_key_id:
        try:
            ssh_key_uuid = uuid.UUID(ssh_key_id)
        except ValueError:
            return ValidateGitUrlResponse(
                valid=False,
                error="Invalid SSH key ID format",
                error_code="INVALID_SSH_KEY",
            )

        ssh_key = await session.get(SSHKey, ssh_key_uuid)
        if ssh_key is None or ssh_key.user_id != current_user_id:
            return ValidateGitUrlResponse(
                valid=False,
                error="SSH key not found or not authorized",
                error_code="SSH_KEY_NOT_FOUND",
            )

        import tempfile

        fernet = _get_fernet()
        private_key = fernet.decrypt(ssh_key.private_key_encrypted.encode()).decode()
        fd, key_path = tempfile.mkstemp(prefix="ssh_key_")
        try:
            os.write(fd, private_key.encode())
        finally:
            os.close(fd)
        os.chmod(key_path, 0o600)
        env = {
            "GIT_SSH_COMMAND": f"ssh -i {key_path} -o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null"
        }

    try:
        result = subprocess.run(
            ["git", "ls-remote", "--heads", url_to_check],
            capture_output=True,
            text=True,
            timeout=30,
            env={**os.environ, **env} if env else None,
        )
    except subprocess.TimeoutExpired:
        if key_path and os.path.exists(key_path):
            os.unlink(key_path)
        return ValidateGitUrlResponse(
            valid=False,
            error="Remote repository check timed out",
            error_code="TIMEOUT",
        )
    except FileNotFoundError:
        if key_path and os.path.exists(key_path):
            os.unlink(key_path)
        return ValidateGitUrlResponse(
            valid=False,
            error="git command not found on server",
            error_code="GIT_NOT_FOUND",
        )
    finally:
        if key_path and os.path.exists(key_path):
            os.unlink(key_path)

    if result.returncode != 0:
        stderr = result.stderr.strip()
        if (
            "could not resolve" in stderr.lower()
            or "unable to access" in stderr.lower()
        ):
            error_msg = "Could not reach repository. Check the URL and network access."
            error_code = "UNREACHABLE"
        elif (
            "authentication" in stderr.lower() or "permission denied" in stderr.lower()
        ):
            error_msg = (
                "Authentication failed. Provide an SSH key for private repositories."
            )
            error_code = "AUTH_FAILED"
        else:
            error_msg = f"Repository not accessible: {stderr[:200]}"
            error_code = "REMOTE_ERROR"
        return ValidateGitUrlResponse(
            valid=False,
            error=error_msg,
            error_code=error_code,
        )

    branches: list[str] = []
    default_branch = "main"
    for line in result.stdout.strip().split("\n"):
        if not line.strip():
            continue
        parts = line.split()
        if len(parts) == 2:
            ref = parts[1]
            if ref.startswith("refs/heads/"):
                branch_name = ref[len("refs/heads/") :]
                branches.append(branch_name)
                if branch_name in ("main", "master"):
                    default_branch = branch_name

    if not branches:
        return ValidateGitUrlResponse(
            valid=False,
            error="No branches found in remote repository",
            error_code="NO_BRANCHES",
        )

    return ValidateGitUrlResponse(
        valid=True,
        suggested_url=url_to_check if url_to_check != original_url else None,
        branches=branches,
        default_branch=default_branch,
    )