state-hub/api/services/forge_repository.py

"""Narrow Forgejo repository boundary used by repository rename operations.

The lifecycle service depends on this protocol instead of calling Forgejo
directly.  Tests replace it with an in-memory gateway, which makes interruption
and resume tests deterministic and guarantees that API tests cannot rename a
live repository.
"""

from __future__ import annotations

import os
from dataclasses import dataclass
from pathlib import Path
from typing import Protocol
from urllib.parse import quote

import httpx

from api.config import settings
from api.services.forge_credential import forge_read_token


WRITE_TOKEN_ENV = "FORGE_RENAME_TOKEN"
WRITE_TOKEN_FILE_ENV = "FORGE_RENAME_TOKEN_FILE"


class ForgeRepositoryError(RuntimeError):
    pass


class ForgeRepositoryUnreadable(ForgeRepositoryError):
    pass


class ForgeRepositoryConflict(ForgeRepositoryError):
    pass


@dataclass(frozen=True)
class ForgeRepositorySnapshot:
    repository_id: int
    owner: str
    name: str
    full_name: str
    default_branch: str
    head_commit: str
    clone_url: str | None
    html_url: str | None
    projection_readable: bool
    projection_source_present: bool
    projection_entry_count: int | None

    def as_dict(self) -> dict:
        return {
            "repository_id": self.repository_id,
            "owner": self.owner,
            "name": self.name,
            "full_name": self.full_name,
            "default_branch": self.default_branch,
            "head_commit": self.head_commit,
            "clone_url": self.clone_url,
            "html_url": self.html_url,
            "projection_readable": self.projection_readable,
            "projection_source_present": self.projection_source_present,
            "projection_entry_count": self.projection_entry_count,
        }


class ForgeRepositoryGateway(Protocol):
    async def inspect(
        self, *, instance: str, owner: str, name: str
    ) -> ForgeRepositorySnapshot | None: ...

    async def rename(
        self, *, instance: str, owner: str, old_name: str, new_name: str
    ) -> ForgeRepositorySnapshot: ...


def _write_token() -> str | None:
    path = os.environ.get(WRITE_TOKEN_FILE_ENV)
    if path:
        try:
            return Path(path).read_text(encoding="utf-8").strip() or None
        except OSError:
            return None
    return (os.environ.get(WRITE_TOKEN_ENV) or "").strip() or None


class ForgejoRepositoryGateway:
    """Forgejo v1 API adapter with separate read and rename credentials."""

    @staticmethod
    def _headers(token: str | None) -> dict[str, str]:
        return {"Authorization": f"token {token}"} if token else {}

    @staticmethod
    def _api(instance: str, owner: str, name: str) -> str:
        base = instance.rstrip("/")
        return f"{base}/api/v1/repos/{quote(owner, safe='')}/{quote(name, safe='')}"

    async def _inspect_with_token(
        self, *, instance: str, owner: str, name: str, token: str | None
    ) -> ForgeRepositorySnapshot | None:
        url = self._api(instance, owner, name)
        timeout = settings.repository_rename_forge_timeout_seconds
        try:
            async with httpx.AsyncClient(timeout=timeout) as client:
                response = await client.get(url, headers=self._headers(token))
                if response.status_code == 404:
                    if token is None:
                        raise ForgeRepositoryUnreadable(
                            "Forge returned not-found without an authenticated absence proof"
                        )
                    return None
                response.raise_for_status()
                repo = response.json()
                branch_name = str(repo.get("default_branch") or "")
                if not branch_name:
                    raise ForgeRepositoryUnreadable("Forge repository has no default branch")
                branch = await client.get(
                    f"{url}/branches/{quote(branch_name, safe='')}",
                    headers=self._headers(token),
                )
                branch.raise_for_status()
                branch_data = branch.json()
                commit = branch_data.get("commit") or {}
                head = str(commit.get("id") or commit.get("sha") or "")
                if not head:
                    raise ForgeRepositoryUnreadable("Forge default branch has no readable head")
                projection = await client.get(
                    f"{url}/contents/workplans",
                    params={"ref": head},
                    headers=self._headers(token),
                )
                source_present = projection.status_code == 200
                if projection.status_code not in {200, 404}:
                    projection.raise_for_status()
                entries = projection.json() if source_present else None
        except ForgeRepositoryError:
            raise
        except (httpx.HTTPError, ValueError, KeyError) as exc:
            raise ForgeRepositoryUnreadable(
                f"Forge repository inspection failed ({type(exc).__name__})"
            ) from exc
        return ForgeRepositorySnapshot(
            repository_id=int(repo["id"]),
            owner=str((repo.get("owner") or {}).get("login") or owner),
            name=str(repo["name"]),
            full_name=str(repo.get("full_name") or f"{owner}/{name}"),
            default_branch=branch_name,
            head_commit=head,
            clone_url=repo.get("clone_url"),
            html_url=repo.get("html_url"),
            projection_readable=True,
            projection_source_present=source_present,
            projection_entry_count=(len(entries) if isinstance(entries, list) else None),
        )

    async def inspect(
        self, *, instance: str, owner: str, name: str
    ) -> ForgeRepositorySnapshot | None:
        return await self._inspect_with_token(
            instance=instance, owner=owner, name=name, token=forge_read_token()
        )

    async def rename(
        self, *, instance: str, owner: str, old_name: str, new_name: str
    ) -> ForgeRepositorySnapshot:
        token = _write_token()
        if not token:
            raise ForgeRepositoryUnreadable("Forge rename credential is unavailable")
        url = self._api(instance, owner, old_name)
        try:
            async with httpx.AsyncClient(
                timeout=settings.repository_rename_forge_timeout_seconds
            ) as client:
                response = await client.patch(
                    url,
                    json={"name": new_name},
                    headers=self._headers(token),
                )
                if response.status_code in {409, 422}:
                    raise ForgeRepositoryConflict("Forge rejected the target repository name")
                response.raise_for_status()
        except ForgeRepositoryError:
            raise
        except httpx.HTTPError as exc:
            raise ForgeRepositoryUnreadable(
                f"Forge repository rename failed ({type(exc).__name__})"
            ) from exc
        renamed = await self._inspect_with_token(
            instance=instance, owner=owner, name=new_name, token=token
        )
        if renamed is None:
            raise ForgeRepositoryUnreadable("Renamed Forge repository is not readable")
        return renamed


_gateway = ForgejoRepositoryGateway()


def get_forge_repository_gateway() -> ForgeRepositoryGateway:
    return _gateway