tgit/app/pipeline/gitea.py

"""Gitea Actions as the uploads CI (ADR-016). Unlike SourceCraft, a Gitea run carries `head_sha` — no meta cube."""

import asyncio
import logging
import re
from collections.abc import Awaitable, Callable
from typing import Any

import httpx

from app.config import CISettings
from app.db.models import PipelineStatus
from app.pipeline.provider import CIRun

log = logging.getLogger(__name__)

_TIMESTAMP = re.compile(r"^\d{4}-\d\d-\d\dT[\d:.]+Z ?", re.MULTILINE)
_FAILED = {"failure", "cancelled"}

Sleep = Callable[[float], Awaitable[None]]


class GiteaError(Exception):
    pass


def _status(status: str, conclusion: str | None) -> PipelineStatus:
    if status != "completed":
        return PipelineStatus.RUNNING if status == "in_progress" else PipelineStatus.PENDING  # queued · waiting
    if conclusion == "success":
        return PipelineStatus.SUCCESS
    if conclusion == "skipped":
        return PipelineStatus.SKIPPED
    return PipelineStatus.FAILED  # failure · cancelled


class GiteaCI:
    def __init__(
        self,
        settings: CISettings,
        transport: httpx.AsyncBaseTransport | None = None,
        sleep: Sleep = asyncio.sleep,
    ) -> None:
        self._s = settings
        self._sleep = sleep
        self._base = f"/api/v1/repos/{settings.repo}/actions"
        self._http = httpx.AsyncClient(
            base_url=settings.api_url,
            headers={"Authorization": f"token {settings.token.get_secret_value()}"},
            timeout=settings.timeout,
            transport=transport,
        )

    async def aclose(self) -> None:
        await self._http.aclose()

    async def find(self, sha: str) -> CIRun | None:
        data = await self._get(f"{self._base}/runs", params={"head_sha": sha, "limit": 20})
        # path is "<file>@<ref>"; other workflows of the repo (if any) don't gate the deploy
        runs = [r for r in data.get("workflow_runs") or [] if str(r.get("path", "")).split("@")[0] == self._s.workflow]
        if not runs:
            return None
        run = max(runs, key=lambda r: int(r["id"]))  # a re-run is a new id
        status = _status(str(run.get("status")), run.get("conclusion"))
        failed_step = await self._failed_step(run["id"]) if status is PipelineStatus.FAILED else None
        return CIRun(str(run["id"]), status, str(run.get("html_url", "")), failed_step)

    async def failure_log(self, run: CIRun) -> str:
        job = await self._failed_job(run.id)
        if job is None:
            return ""
        text = await self._get_text(f"{self._base}/jobs/{job['id']}/logs")
        return _TIMESTAMP.sub("", text)

    async def _failed_job(self, run_id: object) -> dict[str, Any] | None:
        data = await self._get(f"{self._base}/runs/{run_id}/jobs")
        jobs = data.get("jobs") or []
        return next((j for j in jobs if j.get("conclusion") in _FAILED), None)

    async def _failed_step(self, run_id: object) -> str | None:
        job = await self._failed_job(run_id)
        if job is None:
            return None
        step = next((s for s in job.get("steps") or [] if s.get("conclusion") in _FAILED), None)
        if job.get("conclusion") == "cancelled" and step is None:
            return f"{job.get('name')} / cancelled"
        return f"{job.get('name')} / {step['name']}" if step else str(job.get("name"))

    async def _request(self, path: str, params: dict[str, Any] | None = None) -> httpx.Response:
        attempts = self._s.max_attempts
        for attempt in range(1, attempts + 1):
            try:
                response = await self._http.get(path, params=params)
            except httpx.TransportError as e:
                if attempt == attempts:
                    raise GiteaError(f"{path}: {type(e).__name__}") from e
                log.warning("gitea %s: %s, retry %d/%d", path, type(e).__name__, attempt, attempts - 1)
            else:
                if response.status_code < 500 or attempt == attempts:
                    if response.status_code >= 400:
                        raise GiteaError(f"{path}: HTTP {response.status_code}")
                    return response
                log.warning("gitea %s: HTTP %d, retry %d/%d", path, response.status_code, attempt, attempts - 1)
            await self._sleep(min(0.5 * 2 ** (attempt - 1), 8.0))
        raise AssertionError("unreachable")

    async def _get(self, path: str, params: dict[str, Any] | None = None) -> dict[str, Any]:
        response = await self._request(path, params)
        try:
            body = response.json()
        except ValueError as e:
            raise GiteaError(f"{path}: not JSON") from e
        if not isinstance(body, dict):
            raise GiteaError(f"{path}: expected a JSON object")
        return body

    async def _get_text(self, path: str) -> str:
        return (await self._request(path)).text