smart-org/apps/api/app/infrastructure/adapters/gitea.py

"""Gitea adapter for :class:`GitProviderPort`.

Talked to over the HTTP API with an admin token. The token is held by this
adapter only and is never placed on an object that a response could serialise.

Design decisions
----------------

**The adapter is stateless.** It is handed ``owner`` / ``repo_name`` by the
application layer instead of keeping a ``project_id`` map. A map would be lost on
restart and could drift from ``ProjectRepository``, so the database stays the one
source of truth for repository identity.

**Repositories live in one namespace, credentials are per project.** All
repositories are created under a single owner (the configured admin user) so
tenant repos cannot collide, and every project gets its *own* Gitea user that is
added as a collaborator on that one repository. A leaked clone credential
therefore reaches exactly one repository — least privilege rather than a shared
platform token.

**Names are re-validated here.** The application layer already produces safe
slugs, but the adapter must not depend on a caller behaving well: an unvalidated
name is the classic path to a path-traversal or an unintended API path segment.

Reference: ``docs/current-infra-notes.md`` §4.
"""

from __future__ import annotations

import base64
import re
import secrets
from urllib.parse import quote

import httpx

from app.application.ports import (
    BranchInfo,
    CommitInfo,
    GitCredentials,
    MergeResult,
    PullRequestInfo,
    RepoInfo,
    RepoStatus,
)
from app.core.errors import ProviderUnavailableError, ValidationError

#: Gitea accepts ``[a-zA-Z0-9._-]`` for repository names; we are stricter and
#: keep the platform's lowercase-slug convention.
_REPO_NAME_RE = re.compile(r"^[a-z0-9][a-z0-9._-]{0,99}$")
#: Gitea usernames: alphanumeric plus ``-``/``_``, at most 40 characters, and
#: they may not start or end with a separator.
_USERNAME_RE = re.compile(r"^[a-z0-9]([a-z0-9_-]{1,38}[a-z0-9])$")

DEFAULT_TIMEOUT_SECONDS = 20.0
#: Cloning/committing can be slower than a metadata read.
WRITE_TIMEOUT_SECONDS = 60.0


class GiteaGitAdapter:
    """Git provider backed by a Gitea instance."""

    def __init__(
        self,
        *,
        base_url: str,
        admin_user: str,
        admin_token: str,
        owner: str | None = None,
        timeout_seconds: float = DEFAULT_TIMEOUT_SECONDS,
    ) -> None:
        if not base_url:
            raise ProviderUnavailableError(
                "GIT_PROVIDER=gitea نیازمند تنظیم GITEA_INTERNAL_URL یا GITEA_URL است."
            )
        if not admin_token:
            raise ProviderUnavailableError(
                "GIT_PROVIDER=gitea نیازمند تنظیم GITEA_ADMIN_TOKEN است."
            )
        self.base_url = base_url.rstrip("/")
        self.admin_user = admin_user
        self._admin_token = admin_token
        #: Namespace that owns every project repository.
        self.owner = owner or admin_user
        self.timeout_seconds = timeout_seconds

    # -- plumbing ------------------------------------------------------------

    def _client(self, *, write: bool = False) -> httpx.AsyncClient:
        return httpx.AsyncClient(
            base_url=self.base_url,
            headers={
                "Authorization": f"token {self._admin_token}",
                "Accept": "application/json",
            },
            timeout=WRITE_TIMEOUT_SECONDS if write else self.timeout_seconds,
        )

    @staticmethod
    def _sanitize_repo_name(repo_name: str) -> str:
        clean = (repo_name or "").strip().lower()
        if not _REPO_NAME_RE.match(clean):
            raise ValidationError(
                "نام مخزن نامعتبر است؛ فقط حروف کوچک، عدد، نقطه، خط تیره و زیرخط مجاز است."
            )
        return clean

    @staticmethod
    def _sanitize_username(username: str) -> str:
        clean = (username or "").strip().lower()
        if not _USERNAME_RE.match(clean):
            raise ValidationError("نام کاربری مخزن نامعتبر است.")
        return clean

    @staticmethod
    def _seg(value: str) -> str:
        """Percent-encode a path segment so it cannot break out of the path."""
        return quote(value, safe="")

    async def _send(
        self,
        method: str,
        path: str,
        *,
        write: bool = False,
        expect: tuple[int, ...] = (200, 201),
        **kwargs,
    ) -> httpx.Response:
        """Perform a request and translate transport failures.

        A network problem must surface as ``ProviderUnavailableError`` (a 503),
        never as a raw ``httpx`` exception leaking through the API.
        """
        try:
            async with self._client(write=write) as client:
                response = await client.request(method, path, **kwargs)
        except httpx.TimeoutException as exc:
            raise ProviderUnavailableError("ارتباط با Gitea زمان‌بر شد.") from exc
        except httpx.HTTPError as exc:
            raise ProviderUnavailableError("ارتباط با Gitea برقرار نشد.") from exc

        if response.status_code not in expect:
            raise ProviderUnavailableError(
                f"Gitea پاسخ ناموفق داد ({response.status_code}) برای {method} {path}."
            )
        return response

    async def _get_json(self, path: str, *, expect: tuple[int, ...] = (200,)) -> dict | list | None:
        """GET that tolerates 404 by returning ``None``."""
        try:
            async with self._client() as client:
                response = await client.get(path)
        except httpx.HTTPError as exc:
            raise ProviderUnavailableError("ارتباط با Gitea برقرار نشد.") from exc
        if response.status_code == 404:
            return None
        if response.status_code not in expect:
            raise ProviderUnavailableError(
                f"Gitea پاسخ ناموفق داد ({response.status_code}) برای {path}."
            )
        return response.json()

    # -- repository lifecycle ------------------------------------------------

    async def ensure_project_repo(
        self, *, project_id: str, repo_name: str, description: str = ""
    ) -> RepoInfo:
        name = self._sanitize_repo_name(repo_name)
        owner = self._sanitize_username(self.owner)
        path = f"/api/v1/repos/{self._seg(owner)}/{self._seg(name)}"

        existing = await self._get_json(path)
        if existing is None:
            # The admin endpoint creates the repository inside the owner's
            # namespace, which a plain user token could not do.
            await self._send(
                "POST",
                f"/api/v1/admin/users/{self._seg(owner)}/repos",
                json={
                    "name": name,
                    "description": description,
                    "private": True,
                    "auto_init": True,
                    "default_branch": "main",
                },
            )
            existing = await self._get_json(path)
            if existing is None:
                raise ProviderUnavailableError("ساخت مخزن در Gitea تأیید نشد.")

        if not isinstance(existing, dict):
            raise ProviderUnavailableError("پاسخ Gitea برای مخزن قابل تفسیر نبود.")
        return self._to_repo_info(existing, fallback_name=name, fallback_owner=owner)

    def _to_repo_info(self, payload: dict, *, fallback_name: str, fallback_owner: str) -> RepoInfo:
        owner = ((payload.get("owner") or {}).get("login")) or fallback_owner
        name = payload.get("name") or fallback_name
        return RepoInfo(
            owner=owner,
            repo_name=name,
            clone_url=payload.get("clone_url") or f"{self.base_url}/{owner}/{name}.git",
            web_url=payload.get("html_url") or f"{self.base_url}/{owner}/{name}",
            default_branch=payload.get("default_branch") or "main",
            # Never let a remote payload make a tenant repository public.
            private=True,
        )

    async def ensure_credentials(
        self, *, project_id: str, owner: str = "", repo_name: str = ""
    ) -> GitCredentials:
        """Create (or reset) a dedicated Gitea user for this project.

        The user is added as a *collaborator on one repository only*, so the
        returned password cannot reach any other tenant's code.
        """
        repo_owner = self._sanitize_username(owner or self.owner)
        name = self._sanitize_repo_name(repo_name)
        username = self._sanitize_username(f"proj-{project_id[:24]}")

        # A long random password, generated fresh every call. Gitea hashes it.
        password = secrets.token_urlsafe(24)
        email = f"{username}@users.noreply.agy.invalid"

        created = await self._send(
            "POST",
            "/api/v1/admin/users",
            expect=(200, 201, 422),
            json={
                "username": username,
                "email": email,
                "password": password,
                "must_change_password": False,
                "visibility": "private",
            },
        )
        if created.status_code == 422:
            # Already exists — rotate its password instead of failing, so a
            # retried `ensure_credentials` stays idempotent.
            await self._send(
                "PATCH",
                f"/api/v1/admin/users/{self._seg(username)}",
                json={"password": password, "must_change_password": False},
            )

        # Grant write access to exactly this repository.
        await self._send(
            "PUT",
            f"/api/v1/repos/{self._seg(repo_owner)}/{self._seg(name)}"
            f"/collaborators/{self._seg(username)}",
            json={"permission": "write"},
        )

        return GitCredentials(username=username, password=password)

    # -- branches and commits ------------------------------------------------

    async def create_branch(
        self,
        *,
        project_id: str,
        branch_name: str,
        from_branch: str,
        owner: str = "",
        repo_name: str = "",
    ) -> BranchInfo:
        base_path = self._repo_path(owner, repo_name)
        response = await self._send(
            "POST",
            f"{base_path}/branches",
            expect=(200, 201),
            json={"new_branch_name": branch_name, "old_branch_name": from_branch},
        )
        payload = response.json()
        commit = payload.get("commit") or {}
        return BranchInfo(name=payload.get("name", branch_name), sha=commit.get("id", ""))

    async def commit_files(
        self,
        *,
        project_id: str,
        branch_name: str,
        files: dict[str, str],
        message: str,
        owner: str = "",
        repo_name: str = "",
    ) -> CommitInfo:
        """Commit every file, one Contents-API call each.

        Gitea's contents endpoint handles a single file per request, so a
        multi-file change is a sequence of commits. Each update needs the blob's
        current ``sha``, which we read first.
        """
        base_path = self._repo_path(owner, repo_name)
        last_sha = ""
        changed = 0

        for raw_path, content in files.items():
            path = self._safe_content_path(raw_path)
            encoded = quote(path, safe="/")
            existing = await self._get_json(f"{base_path}/contents/{encoded}?ref={branch_name}")
            body = {
                "content": base64.b64encode(content.encode("utf-8")).decode("ascii"),
                "message": message,
                "branch": branch_name,
            }
            if isinstance(existing, dict) and existing.get("sha"):
                body["sha"] = existing["sha"]
                response = await self._send(
                    "PUT", f"{base_path}/contents/{encoded}", write=True, json=body
                )
            else:
                response = await self._send(
                    "POST", f"{base_path}/contents/{encoded}", write=True, json=body
                )
            payload = response.json()
            last_sha = (payload.get("commit") or {}).get("sha") or last_sha
            changed += 1

        return CommitInfo(
            sha=last_sha, branch=branch_name, message=message, files_changed=changed
        )

    @staticmethod
    def _safe_content_path(raw_path: str) -> str:
        """Reject absolute paths and traversal before they reach the URL."""
        clean = (raw_path or "").strip().lstrip("/")
        if not clean:
            raise ValidationError("مسیر فایل نامعتبر است.")
        segments = clean.split("/")
        if any(seg in {"", ".", ".."} for seg in segments):
            raise ValidationError("مسیر فایل نامعتبر است.")
        return "/".join(segments)

    # -- pull requests -------------------------------------------------------

    async def open_pull_request(
        self,
        *,
        project_id: str,
        head: str,
        base: str,
        title: str,
        body: str = "",
        owner: str = "",
        repo_name: str = "",
    ) -> PullRequestInfo:
        response = await self._send(
            "POST",
            f"{self._repo_path(owner, repo_name)}/pulls",
            expect=(200, 201),
            json={"head": head, "base": base, "title": title, "body": body},
        )
        payload = response.json()
        return PullRequestInfo(
            number=int(payload.get("number", 0)),
            url=payload.get("html_url", ""),
            state=payload.get("state", "open"),
        )

    async def merge_pull_request(
        self, *, project_id: str, pr_number: int, owner: str = "", repo_name: str = ""
    ) -> MergeResult:
        try:
            await self._send(
                "POST",
                f"{self._repo_path(owner, repo_name)}/pulls/{int(pr_number)}/merge",
                json={"Do": "merge"},
            )
        except ProviderUnavailableError as exc:
            # A merge conflict is a normal outcome, not an outage.
            return MergeResult(merged=False, message=str(exc))
        return MergeResult(merged=True, message="PR ادغام شد.")

    # -- status --------------------------------------------------------------

    async def get_repo_status(
        self, *, project_id: str, owner: str = "", repo_name: str = ""
    ) -> RepoStatus:
        base_path = self._repo_path(owner, repo_name)
        repo = await self._get_json(base_path)
        if repo is None:
            return RepoStatus(exists=False)
        if not isinstance(repo, dict):
            raise ProviderUnavailableError("پاسخ Gitea برای مخزن قابل تفسیر نبود.")

        branches: list[str] = []
        branch_payload = await self._get_json(f"{base_path}/branches")
        if isinstance(branch_payload, list):
            branches = sorted(b.get("name", "") for b in branch_payload if b.get("name"))

        open_prs = 0
        pr_payload = await self._get_json(f"{base_path}/pulls?state=open&limit=50")
        if isinstance(pr_payload, list):
            open_prs = len(pr_payload)

        return RepoStatus(
            exists=True,
            private=True,
            default_branch=repo.get("default_branch") or "main",
            branches=branches,
            open_pull_requests=open_prs,
        )

    # -- helpers -------------------------------------------------------------

    def _repo_path(self, owner: str, repo_name: str) -> str:
        """Build ``/api/v1/repos/{owner}/{repo}`` from validated segments."""
        if not repo_name:
            raise ValidationError("نام مخزن مشخص نیست.")
        clean_owner = self._sanitize_username(owner or self.owner)
        clean_repo = self._sanitize_repo_name(repo_name)
        return f"/api/v1/repos/{self._seg(clean_owner)}/{self._seg(clean_repo)}"