flowdeck/app/services/github_adapter.py

"""FlowDeck — GitHub API adapter with caching (v5.2.0: ForgeAdapter)."""
from __future__ import annotations

import base64
import logging
from datetime import datetime, timedelta
from typing import Any

import httpx

from app.services.forge_adapter import ForgeAdapter

logger = logging.getLogger(__name__)

DEFAULT_TTL = 30  # seconds


class GitHubAdapter(ForgeAdapter):
    """Async GitHub API client (v3 REST) with simple TTL cache.

    Authenticated via OAuth2 Bearer token. Implements the ``ForgeAdapter``
    interface so Gitea and GitHub repos can be synced identically.
    """

    kind = "github"

    def __init__(self, access_token: str, ttl: int = DEFAULT_TTL,
                 transport: httpx.BaseTransport | None = None) -> None:
        self._base = "https://api.github.com"
        self._headers = {
            "Authorization": f"Bearer {access_token}",
            "Accept": "application/vnd.github+json",
        }
        self._transport = transport
        self._cache: dict[str, tuple[datetime, Any]] = {}
        self._ttl = timedelta(seconds=ttl)

    def _client(self) -> httpx.AsyncClient:
        return httpx.AsyncClient(timeout=15, transport=self._transport)

    # ── cache helpers ──

    def _cached(self, key: str) -> Any | None:
        entry = self._cache.get(key)
        if entry and entry[0] > datetime.now():
            return entry[1]
        return None

    def _set_cache(self, key: str, value: Any) -> None:
        self._cache[key] = (datetime.now() + self._ttl, value)

    # ── repos ──

    async def list_repos(self, page: int = 1, per_page: int = 50) -> list[dict]:
        """List repositories for the authenticated user (paginated)."""
        cache_key = f"repos:{page}:{per_page}"
        cached = self._cached(cache_key)
        if cached:
            return cached

        async with self._client() as client:
            resp = await client.get(
                f"{self._base}/user/repos",
                headers=self._headers,
                params={"page": page, "per_page": per_page, "sort": "updated"},
            )
            resp.raise_for_status()
            data = resp.json()
            self._set_cache(cache_key, data)
            return data

    async def list_all_repos(self) -> list[dict]:
        """Fetch all user repos across pages (up to 5 pages / 250 repos)."""
        all_repos: list[dict] = []
        for page in range(1, 6):
            repos = await self.list_repos(page=page)
            if not repos:
                break
            all_repos.extend(repos)
        return all_repos

    # ── repo tree ──

    async def get_repo_tree(
        self, owner: str, repo: str, sha: str | None = None, recursive: bool = True
    ) -> list[dict]:
        """Get the git tree for a repo. If ``sha`` is omitted, resolves the
        default branch first."""
        cache_key = f"tree:{owner}:{repo}:{sha or 'default'}:{recursive}"
        cached = self._cached(cache_key)
        if cached:
            return cached

        async with self._client() as client:
            # Resolve default branch commit SHA if not provided
            if sha is None:
                repo_info = await client.get(
                    f"{self._base}/repos/{owner}/{repo}",
                    headers=self._headers,
                )
                repo_info.raise_for_status()
                sha = repo_info.json()["default_branch"]

                # Now get the commit to find tree SHA
                branch_resp = await client.get(
                    f"{self._base}/repos/{owner}/{repo}/branches/{sha}",
                    headers=self._headers,
                )
                branch_resp.raise_for_status()
                sha = branch_resp.json()["commit"]["commit"]["tree"]["sha"]

            params: dict[str, Any] = {}
            if recursive:
                params["recursive"] = "1"

            resp = await client.get(
                f"{self._base}/repos/{owner}/{repo}/git/trees/{sha}",
                headers=self._headers,
                params=params,
            )
            resp.raise_for_status()
            data = resp.json()
            items = data.get("tree", [])
            # Truncated responses — fetch remaining pages if needed
            while data.get("truncated", False):
                logger.warning("GitHub tree truncated for %s/%s — results incomplete", owner, repo)
                break

            self._set_cache(cache_key, items)
            return items

    # ── file content ──

    async def get_file_content(self, owner: str, repo: str, path: str) -> str:
        """Get decoded file content from a repo."""
        cache_key = f"file:{owner}:{repo}:{path}"
        cached = self._cached(cache_key)
        if cached:
            return cached

        async with self._client() as client:
            resp = await client.get(
                f"{self._base}/repos/{owner}/{repo}/contents/{path}",
                headers=self._headers,
            )
            resp.raise_for_status()
            item = resp.json()

            content = ""
            if item.get("encoding") == "base64" and item.get("content"):
                try:
                    content = base64.b64decode(item["content"]).decode("utf-8")
                except Exception:
                    content = "[binary file]"

            self._set_cache(cache_key, content)
            return content

    # ── repo info (ForgeAdapter) ──

    async def get_repo_info(self, owner: str, repo: str) -> dict:
        """Repository metadata: default_branch, clone_url, languages, …"""
        cache_key = f"repo_info:{owner}:{repo}"
        cached = self._cached(cache_key)
        if cached:
            return cached

        async with self._client() as client:
            resp = await client.get(
                f"{self._base}/repos/{owner}/{repo}",
                headers=self._headers,
            )
            resp.raise_for_status()
            info = resp.json()
            repo_info = {
                "id": info.get("id"),
                "name": info.get("name"),
                "owner": (info.get("owner") or {}).get("login", owner),
                "full_name": info.get("full_name"),
                "clone_url": info.get("clone_url", ""),
                "default_branch": info.get("default_branch", "main"),
                "description": info.get("description") or "",
                "language": info.get("language") or "",
                "html_url": info.get("html_url", ""),
            }
            # Languages are a separate endpoint.
            try:
                lang_resp = await client.get(
                    f"{self._base}/repos/{owner}/{repo}/languages",
                    headers=self._headers,
                )
                if lang_resp.status_code == 200:
                    langs = lang_resp.json()
                    if langs:
                        repo_info["language"] = max(langs, key=langs.get)
            except Exception:
                pass

            self._set_cache(cache_key, repo_info)
            return repo_info

    async def get_languages(self, owner: str, repo: str) -> dict:
        """Bytes per language for a repo."""
        async with self._client() as client:
            resp = await client.get(
                f"{self._base}/repos/{owner}/{repo}/languages",
                headers=self._headers,
            )
            resp.raise_for_status()
            return resp.json()

    # ── token validation ──

    async def validate_token(self) -> bool:
        """Check whether the access token is still valid."""
        async with self._client() as client:
            resp = await client.get(
                f"{self._base}/user",
                headers=self._headers,
            )
            return resp.status_code == 200