data_nexus/app/services/gogs_client.py

import httpx
from app.config import settings
import logging
from typing import Optional

logger = logging.getLogger(__name__)

# Default timeout for API requests (seconds)
_DEFAULT_TIMEOUT = 30.0


class GogsClient:
    def __init__(self):
        self.base_url = settings.GOGS_URL.rstrip('/')
        self.token = settings.GOGS_TOKEN
        self.headers = {"Authorization": f"token {self.token}"}

    # ------------------------------------------------------------------
    # Repository discovery
    # ------------------------------------------------------------------

    async def list_user_repos(self) -> list[dict]:
        """Fetch *all* repositories visible to the authenticated user.

        Gogs paginates with `?page=N` (default 20 per page).
        We iterate until an empty page is returned.
        """
        repos: list[dict] = []

        async with httpx.AsyncClient(timeout=_DEFAULT_TIMEOUT) as client:
            url = f"{self.base_url}/api/v1/user/repos"
            resp = await client.get(url, headers=self.headers)
            resp.raise_for_status()
            batch = resp.json()
            repos.extend(batch)

        logger.info(f"Fetched {len(repos)} repositories in total")
        return repos

    # ------------------------------------------------------------------
    # Webhook management
    # ------------------------------------------------------------------

    async def list_repo_webhooks(self, owner: str, repo: str) -> list[dict]:
        """List all webhooks configured on a repository."""
        url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/hooks"
        async with httpx.AsyncClient(timeout=_DEFAULT_TIMEOUT) as client:
            resp = await client.get(url, headers=self.headers)
            resp.raise_for_status()
            return resp.json()

    async def create_repo_webhook(
        self,
        owner: str,
        repo: str,
        webhook_url: str,
        secret: str = "",
        events: Optional[list[str]] = None,
    ) -> dict:
        """Create a push webhook on a repository.

        Returns the created webhook payload from Gogs.
        """
        url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/hooks"
        payload = {
            "type": "gogs",
            "config": {
                "url": webhook_url,
                "content_type": "json",
                "secret": secret,
            },
            "events": events or ["push"],
            "active": True,
        }
        async with httpx.AsyncClient(timeout=_DEFAULT_TIMEOUT) as client:
            resp = await client.post(url, headers=self.headers, json=payload)
            resp.raise_for_status()
            return resp.json()

    # ------------------------------------------------------------------
    # Manifest / file operations (existing)
    # ------------------------------------------------------------------

    async def get_manifest(self, owner: str, repo: str, ref: str) -> str | None:
        """Fetch manifest.yaml raw content from a given ref (commit / branch)."""
        url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/raw/{ref}/manifest.yaml"
        async with httpx.AsyncClient(timeout=_DEFAULT_TIMEOUT) as client:
            resp = await client.get(url, headers=self.headers)
            if resp.status_code == 404:
                return None
            resp.raise_for_status()
            return resp.text

    async def get_tree(self, owner: str, repo: str, commit_id: str, path: str = "") -> list:
        """Get the file tree of a repository."""
        url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/contents/{path}?ref={commit_id}"
        async with httpx.AsyncClient(timeout=_DEFAULT_TIMEOUT) as client:
            resp = await client.get(url, headers=self.headers)
            resp.raise_for_status()
            return resp.json()

    async def get_file_info(self, owner: str, repo: str, commit_id: str, file_path: str) -> dict | None:
        """Get single file info including SHA.

        Returns dict with 'sha', 'size', 'path' or None if not found.
        """
        url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/contents/{file_path}?ref={commit_id}"
        try:
            async with httpx.AsyncClient(timeout=_DEFAULT_TIMEOUT) as client:
                resp = await client.get(url, headers=self.headers)
                if resp.status_code == 404:
                    return None
                resp.raise_for_status()
                data = resp.json()
                # contents API returns file info directly for single file
                if isinstance(data, dict) and data.get("type") == "file":
                    return {
                        "path": file_path,
                        "sha": data.get("sha"),
                        "size": data.get("size", 0),
                        "type": "blob"
                    }
                return None
        except httpx.HTTPStatusError as e:
            logger.error(f"Failed to get file info for {file_path}: {e}")
            return None

    async def get_directory_tree(self, owner: str, repo: str, commit_id: str, dir_path: str) -> list:
        """Get all files under a specific directory (recursive) using concurrency."""
        import asyncio
        all_files = []

        async with httpx.AsyncClient(timeout=_DEFAULT_TIMEOUT, headers=self.headers) as client:
            async def fetch_contents(path: str):
                """Recursively fetch directory contents using contents API in parallel."""
                url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/contents/{path}?ref={commit_id}"
                try:
                    resp = await client.get(url)
                    if resp.status_code == 404:
                        logger.warning(f"Directory not found: {path}")
                        return
                    resp.raise_for_status()
                    data = resp.json()

                    if isinstance(data, list):
                        tasks = []
                        for item in data:
                            if item.get("type") == "file":
                                all_files.append({
                                    "path": item.get("path"),
                                    "sha": item.get("sha"),
                                    "size": item.get("size", 0),
                                    "type": "blob"
                                })
                            elif item.get("type") == "dir":
                                tasks.append(fetch_contents(item.get("path")))
                        
                        if tasks:
                            await asyncio.gather(*tasks)
                except Exception as e:
                    logger.error(f"Failed to get contents for {path}: {e}")

            await fetch_contents(dir_path)
        return all_files

    async def get_file_content(self, owner: str, repo: str, commit_id: str, file_path: str) -> bytes:
        """Download raw file content."""
        # Gogs raw file URL format: /{owner}/{repo}/raw/{ref}/{path}
        url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/raw/{commit_id}/{file_path}"
        async with httpx.AsyncClient(timeout=_DEFAULT_TIMEOUT) as client:
            resp = await client.get(url, headers=self.headers)
            resp.raise_for_status()
            return resp.content