kanbante/app/gitea_client.py

"""
Async Gitea API client for Kanbante.

Changes from the 0.1.0 client:
- Fully asynchronous (httpx.AsyncClient): routes no longer block the
  event loop on Gitea calls.
- Real pagination: Gitea caps every list response at MAX_RESPONSE_ITEMS
  (50 by default) and silently truncates larger `limit` values, so all
  list methods follow pages until the result is complete.
- The Authorization header is only sent when a token is configured.
- The current user's id is resolved from /user (no hardcoded uid or
  username); without a token it falls back to a public search.
- Write operations used by the two-way sync: create_issue() and
  set_issue_state().
- Short in-memory cache for list responses (KANBANTE_CACHE_TTL seconds),
  invalidated on every mutation.
- .env loading targets the project root only (no hardcoded home path).
"""

import asyncio
import logging
import time
from collections.abc import Awaitable, Callable
from pathlib import Path
from typing import Any

import httpx
from dotenv import load_dotenv

from app.config import get_settings

load_dotenv(Path(__file__).resolve().parent.parent / ".env")

logger = logging.getLogger("kanbante.gitea")

PAGE_SIZE = 50  # Gitea's default MAX_RESPONSE_ITEMS
MAX_PAGES = 40  # hard stop: 40 * 50 = 2000 items per list


class GiteaClient:
    """Lightweight async Gitea API client for Kanbante.

    Since 0.24.0 this class is the reference implementation of
    the forge protocol: GitHubClient subclasses it and overrides
    the endpoints whose shape differs. ``api_suffix`` and
    ``auth_scheme`` are the two seams; ``key_prefix`` namespaces
    the persisted snapshots per connection ("c<id>:").
    """

    kind = "gitea"
    supports_attachments = True
    api_suffix = "/api/v1"
    auth_scheme = "token"

    def __init__(
        self,
        base_url: str | None = None,
        token: str | None = None,
        cache_ttl: float | None = None,
        transport: httpx.AsyncBaseTransport | None = None,
    ) -> None:
        settings = get_settings()
        self.base_url = (base_url or settings.gitea_url).rstrip("/")
        self.token = settings.gitea_token if token is None else token
        self.cache_ttl = settings.cache_ttl if cache_ttl is None else cache_ttl
        self.key_prefix = ""
        self.connection_id = 1
        self.connection_slug = "gitea"
        headers = {"Accept": "application/json"}
        headers.update(self._extra_headers())
        if self.token:
            headers["Authorization"] = f"{self.auth_scheme} {self.token}"
        self.client = httpx.AsyncClient(
            base_url=f"{self.base_url}{self.api_suffix}",
            headers=headers,
            timeout=15.0,
            transport=transport,
        )
        self._user: dict | None = None
        self._cache: dict[str, tuple[float, Any]] = {}
        self._refreshing: set[str] = set()
        self._cache_gen = 0
        # Snapshot bookkeeping (chantier B): cache key -> UTC time
        # the payload was fetched from Gitea, for the freshness
        # label and the persisted `snapshots` table.
        self._snapshot_times: dict[str, str] = {}

    def _extra_headers(self) -> dict:
        return {}

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

    # ── Low-level helpers ──────────────────────────────────

    async def _get(self, path: str, **params: Any) -> Any:
        r = await self.client.get(path, params=params)
        r.raise_for_status()
        return r.json()

    async def _get_all_pages(self, path: str, **params: Any) -> list[dict]:
        """Fetch every page of a list endpoint.

        Stops on a short page, on X-Total-Count when Gitea provides it,
        or at MAX_PAGES as a safety net.
        """
        items: list[dict] = []
        page = 1
        while page <= MAX_PAGES:
            r = await self.client.get(
                path, params={**params, "page": page, "limit": PAGE_SIZE}
            )
            r.raise_for_status()
            data = r.json()
            if isinstance(data, dict) and "data" in data:
                batch = data["data"]
            elif isinstance(data, list):
                batch = data
            else:
                batch = []
            items.extend(batch)

            total = r.headers.get("X-Total-Count")
            if total is not None:
                try:
                    if len(items) >= int(total):
                        break
                except ValueError:
                    pass
            if len(batch) < PAGE_SIZE:
                break
            page += 1
        return items

    def _cache_put(self, key: str, value: Any) -> None:
        if self.cache_ttl > 0:
            self._cache[key] = (time.monotonic(), value)
            self._persist_snapshot(key, value)

    # ── Snapshots (chantier B, 0.22.0) ──────────────────────

    def _prefixed(self, key: str) -> str:
        return f"{self.key_prefix}{key}"

    def _persist_snapshot(self, key: str, value: Any) -> None:
        """Write-through to the SQLite snapshots table.

        Persistence must never break a read: any failure (DB
        locked, unserializable payload) is logged and swallowed.
        """
        try:
            from app import models

            models.save_snapshot(self._prefixed(key), value)
            import datetime

            self._snapshot_times[key] = datetime.datetime.now(
                datetime.timezone.utc
            ).strftime("%Y-%m-%d %H:%M:%S")
        except Exception as exc:
            logger.debug("snapshot persist failed for %s: %s", key, exc)

    def hydrate_from_snapshots(self) -> int:
        """Preload the cache from persisted snapshots (startup).

        Entries are inserted with an already-stale timestamp: the
        SWR layer serves them instantly AND schedules a background
        refresh, so the first page after a restart renders from
        local data (tens of ms) while Gitea data catches up.
        Returns the number of entries hydrated.
        """
        if self.cache_ttl <= 0:
            return 0
        try:
            from app import models

            snapshots = models.load_snapshots()
        except Exception as exc:
            logger.warning("snapshot hydration failed: %s", exc)
            return 0
        stale_at = time.monotonic() - self.cache_ttl - 1.0
        count = 0
        for stored_key, (payload, updated_at) in snapshots.items():
            if self.key_prefix:
                if not stored_key.startswith(self.key_prefix):
                    continue
                key = stored_key[len(self.key_prefix):]
            else:
                key = stored_key
            self._cache[key] = (stale_at, payload)
            self._snapshot_times[key] = updated_at
            count += 1
        if count:
            logger.info("hydrated %d snapshot(s) from SQLite", count)
        return count

    def latest_snapshot_at(self) -> str | None:
        """Freshest Gitea fetch time across tracked keys (UTC str)."""
        if not self._snapshot_times:
            return None
        return max(self._snapshot_times.values())

    async def _force(self, key: str, fetcher: Callable[[], Awaitable[Any]]) -> Any:
        """Fetch now, bypassing freshness (background refresher)."""
        value = await fetcher()
        self._cache_put(key, value)
        return value

    async def refresh_starred(self) -> list[dict]:
        return await self._force(
            "user:starred", lambda: self._get_all_pages("/user/starred")
        )

    async def refresh_repo_issues(
        self, owner: str, repo: str, state: str = "all"
    ) -> list[dict]:
        return await self._force(
            f"issues:{owner}/{repo}:{state}",
            lambda: self._get_all_pages(
                f"/repos/{owner}/{repo}/issues", state=state
            ),
        )

    async def refresh_milestones(
        self, owner: str, repo: str
    ) -> list[dict]:
        return await self._force(
            f"milestones:{owner}/{repo}:open",
            lambda: self._get_all_pages(
                f"/repos/{owner}/{repo}/milestones", state="open"
            ),
        )

    def _cache_clear(self) -> None:
        self._cache.clear()
        # Invalidate in-flight background refreshes: a refresh that
        # started before a mutation must not resurrect stale data.
        self._cache_gen += 1

    async def _cached_list(
        self, key: str, fetcher: Callable[[], Awaitable[Any]]
    ) -> Any:
        """Stale-while-revalidate cache for list endpoints (0.19.0).

        Fresh (< TTL): serve the cache. Stale: serve the cached
        value IMMEDIATELY and refresh it in the background (once
        per key) — the user never waits on Gitea for a page when
        any earlier answer exists. Absent: fetch and wait.
        Mutations clear the cache (see _cache_clear), so writes are
        never hidden by a stale entry.
        """
        if self.cache_ttl <= 0:
            return await fetcher()
        hit = self._cache.get(key)
        if hit is not None:
            if (time.monotonic() - hit[0]) < self.cache_ttl:
                return hit[1]
            if key not in self._refreshing:
                self._refreshing.add(key)
                asyncio.create_task(
                    self._refresh_entry(key, fetcher, self._cache_gen)
                )
            return hit[1]
        value = await fetcher()
        self._cache_put(key, value)
        return value

    async def _refresh_entry(
        self,
        key: str,
        fetcher: Callable[[], Awaitable[Any]],
        generation: int,
    ) -> None:
        try:
            value = await fetcher()
            if generation == self._cache_gen:
                self._cache_put(key, value)
        except Exception as exc:
            logger.debug("background refresh failed for %s: %s", key, exc)
        finally:
            self._refreshing.discard(key)

    def invalidate_cache(self) -> None:
        """Drop every cached list response (called on webhook events)."""
        self._cache_clear()

    # ── API ────────────────────────────────────────────────

    async def ping(self) -> bool:
        """True if the Gitea instance answers /version."""
        try:
            r = await self.client.get("/version", timeout=5.0)
            return r.status_code == 200
        except Exception:
            return False

    async def get_user(self) -> dict:
        """Get authenticated user info (cached on the instance)."""
        if self._user is None:
            self._user = await self._get("/user")
        return self._user

    async def _fetch_search_repos(self) -> list[dict]:
        params: dict[str, Any] = {}
        try:
            user = await self.get_user()
            if user and user.get("id"):
                params["uid"] = user["id"]
        except Exception as exc:  # no/invalid token → public search
            logger.debug(
                "get_user failed, falling back to public search: %s", exc
            )
        return await self._get_all_pages("/repos/search", **params)

    async def search_repos(self) -> list[dict]:
        """All repos visible to the token owner (paginated, SWR-cached)."""
        return await self._cached_list("search_repos", self._fetch_search_repos)

    async def refresh_search_repos(self) -> list[dict]:
        """Force-fetch the repo index (background refresher)."""
        return await self._force("search_repos", self._fetch_search_repos)

    async def get_repo(self, owner: str, repo: str) -> dict:
        return await self._get(f"/repos/{owner}/{repo}")

    async def get_user_repos(self, username: str) -> list[dict]:
        """Get all repos for a user (paginated)."""
        return await self._get_all_pages(f"/users/{username}/repos")

    async def get_org_repos(self, org: str) -> list[dict]:
        """Get all repos for an organization (paginated)."""
        return await self._get_all_pages(f"/orgs/{org}/repos")

    async def get_repo_issues(
        self, owner: str, repo: str, state: str = "all"
    ) -> list[dict]:
        """Get issues for a repo (paginated, cached).

        Note: Gitea's issues endpoint also returns pull requests; a
        PR carries a non-null "pull_request" object, a plain issue
        carries "pull_request": null — callers filter on truthiness
        (see app.main._is_issue).
        """
        key = f"issues:{owner}/{repo}:{state}"
        return await self._cached_list(
            key,
            lambda: self._get_all_pages(
                f"/repos/{owner}/{repo}/issues", state=state
            ),
        )

    async def get_all_issues(
        self, repos: list[dict], state: str = "open"
    ) -> dict[str, list[dict]]:
        """Get issues for multiple repos, keyed by full_name."""
        issues_by_repo: dict[str, list[dict]] = {}
        for repo in repos:
            full_name = repo["full_name"]
            owner, name = full_name.split("/", 1)
            try:
                issues_by_repo[full_name] = await self.get_repo_issues(
                    owner, name, state=state
                )
            except Exception as exc:
                logger.warning("issues fetch failed for %s: %s", full_name, exc)
                issues_by_repo[full_name] = []
        return issues_by_repo

    async def get_issue(self, owner: str, repo: str, number: int) -> dict:
        """One issue with body, labels, assignees, milestone."""
        return await self._get(f"/repos/{owner}/{repo}/issues/{number}")

    async def get_issue_comments(
        self, owner: str, repo: str, number: int
    ) -> list[dict]:
        """All comments of an issue (paginated)."""
        return await self._get_all_pages(
            f"/repos/{owner}/{repo}/issues/{number}/comments"
        )

    async def create_comment(
        self, owner: str, repo: str, number: int, body: str
    ) -> dict:
        """Post a comment on an issue and return it."""
        r = await self.client.post(
            f"/repos/{owner}/{repo}/issues/{number}/comments",
            json={"body": body},
        )
        r.raise_for_status()
        return r.json()

    # ── Labels (kanban:* column mirroring) ─────────────────

    async def get_labels(self, owner: str, repo: str) -> list[dict]:
        return await self._get_all_pages(f"/repos/{owner}/{repo}/labels")

    async def create_label(
        self, owner: str, repo: str, name: str, color: str
    ) -> dict:
        r = await self.client.post(
            f"/repos/{owner}/{repo}/labels",
            json={"name": name, "color": color.lstrip("#")},
        )
        r.raise_for_status()
        return r.json()

    async def ensure_label(
        self, owner: str, repo: str, name: str, color: str
    ) -> dict:
        """Return the repo label with this name, creating it if needed."""
        key = f"{owner}/{repo}"
        cache = self.__dict__.setdefault("_label_cache", {})
        if key not in cache:
            cache[key] = {l["name"]: l for l in await self.get_labels(owner, repo)}
        if name not in cache[key]:
            cache[key][name] = await self.create_label(owner, repo, name, color)
        return cache[key][name]

    async def add_issue_label(
        self, owner: str, repo: str, number: int, label_id: int
    ) -> None:
        r = await self.client.post(
            f"/repos/{owner}/{repo}/issues/{number}/labels",
            json={"labels": [label_id]},
        )
        r.raise_for_status()

    async def remove_issue_label(
        self, owner: str, repo: str, number: int, label_id: int
    ) -> None:
        r = await self.client.delete(
            f"/repos/{owner}/{repo}/issues/{number}/labels/{label_id}"
        )
        r.raise_for_status()

    async def delete_label(self, owner: str, repo: str, label_id: int) -> None:
        """Delete a label definition from the repo (all issues lose it)."""
        r = await self.client.delete(f"/repos/{owner}/{repo}/labels/{label_id}")
        r.raise_for_status()
        self.forget_label_cache(owner, repo)

    def forget_label_cache(self, owner: str, repo: str) -> None:
        cache = self.__dict__.get("_label_cache")
        if cache:
            cache.pop(f"{owner}/{repo}", None)

    # ── Issue search (global + assigned) ───────────────────

    async def search_issues(
        self,
        q: str = "",
        state: str = "open",
        assigned_by: str | None = None,
    ) -> list[dict]:
        """Search issues across every repo visible to the token.

        Uses Gitea's /repos/issues/search (type=issues excludes PRs).
        Results carry their repository in issue["repository"].
        """
        params: dict[str, Any] = {"state": state, "type": "issues"}
        if q:
            params["q"] = q
        if assigned_by:
            params["assigned_by"] = assigned_by
        return await self._get_all_pages("/repos/issues/search", **params)

    # ── Milestones ─────────────────────────────────────────

    async def get_milestones(
        self, owner: str, repo: str, state: str = "open"
    ) -> list[dict]:
        key = f"milestones:{owner}/{repo}:{state}"
        return await self._cached_list(
            key,
            lambda: self._get_all_pages(
                f"/repos/{owner}/{repo}/milestones", state=state
            ),
        )

    async def list_starred(self) -> list[dict]:
        """Repos starred by the token user (issue #20: pinning rides
        on Gitea stars — the forge stays the source of truth)."""
        return await self._cached_list(
            "user:starred", lambda: self._get_all_pages("/user/starred")
        )

    async def set_starred(self, owner: str, repo: str, starred: bool) -> None:
        """Star / unstar a repo for the token user."""
        if starred:
            r = await self.client.put(f"/user/starred/{owner}/{repo}")
        else:
            r = await self.client.delete(f"/user/starred/{owner}/{repo}")
        r.raise_for_status()
        self._cache_clear()

    async def create_issue(
        self, owner: str, repo: str, title: str, body: str = ""
    ) -> dict:
        """Create an issue in Gitea and return it."""
        r = await self.client.post(
            f"/repos/{owner}/{repo}/issues", json={"title": title, "body": body}
        )
        r.raise_for_status()
        self._cache_clear()
        return r.json()

    async def update_issue(
        self,
        owner: str,
        repo: str,
        number: int,
        title: str | None = None,
        body: str | None = None,
    ) -> dict:
        """Edit an issue's title and/or body (0.31.0)."""
        payload: dict = {}
        if title is not None:
            payload["title"] = title
        if body is not None:
            payload["body"] = body
        if not payload:
            return await self.get_issue(owner, repo, number)
        r = await self.client.patch(
            f"/repos/{owner}/{repo}/issues/{number}", json=payload
        )
        r.raise_for_status()
        self._cache_clear()
        return r.json()

    # ── Pièces jointes des issues (0.31.0) ────────────────

    async def list_issue_assets(
        self, owner: str, repo: str, number: int
    ) -> list[dict]:
        """Attachments of an issue (Gitea assets)."""
        return await self._get(
            f"/repos/{owner}/{repo}/issues/{number}/assets"
        )

    async def upload_issue_asset(
        self,
        owner: str,
        repo: str,
        number: int,
        filename: str,
        content: bytes,
        content_type: str = "application/octet-stream",
    ) -> dict:
        """Attach a file to an issue; returns the asset record."""
        r = await self.client.post(
            f"/repos/{owner}/{repo}/issues/{number}/assets",
            files={"attachment": (filename, content, content_type)},
        )
        r.raise_for_status()
        return r.json()

    async def delete_issue_asset(
        self, owner: str, repo: str, number: int, asset_id: int
    ) -> None:
        r = await self.client.delete(
            f"/repos/{owner}/{repo}/issues/{number}/assets/{asset_id}"
        )
        r.raise_for_status()

    async def download_issue_asset(
        self, owner: str, repo: str, number: int, asset_id: int
    ) -> tuple[bytes, str]:
        """(content, content_type) of one attachment.

        Gitea nuance (vérifiée en réel sur 1.27.3): GET on the
        asset endpoint returns its JSON record, NOT the bytes —
        the file itself lives at `browser_download_url`
        (/attachments/<uuid>), fetched here with the connection's
        token so private repos work too.
        """
        record = await self._get(
            f"/repos/{owner}/{repo}/issues/{number}/assets/{asset_id}"
        )
        url = record.get("browser_download_url") if isinstance(record, dict) else None
        if not url:
            raise ValueError("La forge ne fournit pas d'URL de téléchargement")
        r = await self.client.get(url)
        r.raise_for_status()
        return r.content, r.headers.get(
            "content-type", "application/octet-stream"
        )

    async def set_issue_state(
        self, owner: str, repo: str, issue_number: int, state: str
    ) -> dict:
        """Open or close an issue in Gitea (state: "open" | "closed")."""
        if state not in ("open", "closed"):
            raise ValueError(f"Invalid issue state: {state}")
        r = await self.client.patch(
            f"/repos/{owner}/{repo}/issues/{issue_number}", json={"state": state}
        )
        r.raise_for_status()
        self._cache_clear()
        return r.json()

    async def set_archived(
        self, owner: str, repo: str, archived: bool
    ) -> dict:
        """Archive / unarchive the repo ON THE FORGE (0.24.0, P4)."""
        r = await self.client.put(
            f"/repos/{owner}/{repo}", json={"archived": archived}
        )
        r.raise_for_status()
        self._cache_clear()
        return r.json()

    async def create_branch(
        self, owner: str, repo: str, new_branch: str, old_branch: str
    ) -> dict:
        """Create a branch from another one (issue #3)."""
        return await self._request(
            "POST",
            f"/repos/{owner}/{repo}/branches",
            json={
                "new_branch_name": new_branch,
                "old_branch_name": old_branch,
            },
        )

    async def list_branches(self, owner: str, repo: str) -> list[dict]:
        """All branches of a repo (code links, issue #1)."""
        return await self._get_all_pages(f"/repos/{owner}/{repo}/branches")

    async def list_pulls(
        self, owner: str, repo: str, state: str = "all"
    ) -> list[dict]:
        """All pull requests of a repo (code links, issue #1)."""
        return await self._get_all_pages(
            f"/repos/{owner}/{repo}/pulls", state=state
        )

    async def list_commits(
        self, owner: str, repo: str, limit: int = 75
    ) -> list[dict]:
        """Most recent commits of a repo (code links, issue #1)."""
        data = await self._get(f"/repos/{owner}/{repo}/commits", limit=limit)
        return data if isinstance(data, list) else []

    async def set_issue_assignees(
        self, owner: str, repo: str, issue_number: int, logins: list[str]
    ) -> dict:
        """Replace the assignee set of an issue (triage "assign me")."""
        r = await self.client.patch(
            f"/repos/{owner}/{repo}/issues/{issue_number}",
            json={"assignees": logins},
        )
        r.raise_for_status()
        self._cache_clear()
        return r.json()