scheduler/src/executors/gitea_release_cleanup_executor.py

"""
Gitea Release Cleanup Executor

Cleans up old releases across all Gitea repositories, keeping only the most recent
releases per repository. Designed to run before Watchtower to prevent accumulation
of old container image tags.

Config schema:
    {
        "keep_count": 5,           # Number of releases to keep per repo (default: 5)
        "exclude_repos": [],       # Repository names to skip (default: [])
        "dry_run": false           # If true, only log what would be deleted (default: false)
    }

Example config:
    {
        "keep_count": 5,
        "exclude_repos": ["important-repo", "legacy-app"],
        "dry_run": false
    }

Required settings:
    - GITEA_URL: Base URL of Gitea instance
    - GITEA_TOKEN: API token with write:repository scope
"""
import logging
from typing import Any

import httpx

from src.config import Settings

logger = logging.getLogger(__name__)

# Default configuration values
DEFAULT_KEEP_COUNT = 5
DEFAULT_TIMEOUT = 60
API_BASE = "/api/v1"


async def execute(config: dict, settings: Settings) -> str:
    """
    Clean up old releases across all accessible Gitea repositories.

    Args:
        config: Task configuration (see module docstring)
        settings: Global scheduler settings

    Returns:
        Summary of cleanup actions taken

    Raises:
        ValueError: On configuration error
        Exception: On API call failure
    """
    # Validate settings
    if not settings.gitea_url:
        raise ValueError("GITEA_URL not configured")
    if not settings.gitea_token:
        raise ValueError("GITEA_TOKEN not configured - required for release deletion")

    # Parse configuration
    keep_count = config.get("keep_count", DEFAULT_KEEP_COUNT)
    exclude_repos = config.get("exclude_repos", [])
    dry_run = config.get("dry_run", False)

    if keep_count < 1:
        raise ValueError(f"keep_count must be at least 1, got {keep_count}")

    base_url = settings.gitea_url.rstrip("/")
    headers = {"Authorization": f"token {settings.gitea_token}"}

    stats = {
        "repos_scanned": 0,
        "repos_with_releases": 0,
        "releases_deleted": 0,
        "releases_skipped": 0,
        "errors": [],
    }

    mode = "DRY RUN" if dry_run else "LIVE"
    logger.info(f"Starting Gitea release cleanup ({mode}): keeping {keep_count} releases per repo")

    async with httpx.AsyncClient(timeout=DEFAULT_TIMEOUT) as client:
        # Fetch all repositories
        repos = await _fetch_user_repos(client, base_url, headers)
        logger.info(f"Found {len(repos)} repositories")

        for repo in repos:
            owner = repo["owner"]["login"]
            name = repo["name"]
            full_name = f"{owner}/{name}"

            # Check exclusion list
            if name in exclude_repos or full_name in exclude_repos:
                logger.debug(f"Skipping excluded repo: {full_name}")
                continue

            stats["repos_scanned"] += 1

            try:
                # Fetch releases for this repo
                releases = await _fetch_releases(client, base_url, headers, owner, name)

                if not releases:
                    continue

                stats["repos_with_releases"] += 1

                # Determine which releases to delete (beyond keep_count)
                to_delete = releases[keep_count:]

                if not to_delete:
                    logger.debug(f"{full_name}: {len(releases)} releases, nothing to delete")
                    continue

                logger.info(f"{full_name}: {len(releases)} releases, deleting {len(to_delete)}")

                # Delete old releases
                for release in to_delete:
                    release_id = release["id"]
                    tag_name = release["tag_name"]

                    if dry_run:
                        logger.info(f"  [DRY RUN] Would delete: {tag_name} (id={release_id})")
                        stats["releases_skipped"] += 1
                    else:
                        try:
                            await _delete_release(client, base_url, headers, owner, name, release_id)
                            logger.info(f"  Deleted: {tag_name}")
                            stats["releases_deleted"] += 1
                        except Exception as e:
                            error_msg = f"{full_name}/{tag_name}: {e}"
                            logger.warning(f"  Failed to delete {tag_name}: {e}")
                            stats["errors"].append(error_msg)

            except Exception as e:
                error_msg = f"{full_name}: {e}"
                logger.error(f"Error processing {full_name}: {e}")
                stats["errors"].append(error_msg)

    # Build summary
    summary = _build_summary(stats, dry_run)
    logger.info(f"Cleanup complete: {summary}")

    return summary


async def _fetch_user_repos(
    client: httpx.AsyncClient,
    base_url: str,
    headers: dict,
) -> list[dict[str, Any]]:
    """Fetch all repositories accessible to the authenticated user."""
    url = f"{base_url}{API_BASE}/user/repos"
    params = {"limit": 100}

    response = await client.get(url, headers=headers, params=params)
    response.raise_for_status()

    return response.json()


async def _fetch_releases(
    client: httpx.AsyncClient,
    base_url: str,
    headers: dict,
    owner: str,
    repo: str,
) -> list[dict[str, Any]]:
    """Fetch releases for a repository, sorted newest first (Gitea default)."""
    url = f"{base_url}{API_BASE}/repos/{owner}/{repo}/releases"
    params = {"limit": 100}

    response = await client.get(url, headers=headers, params=params)
    response.raise_for_status()

    return response.json()


async def _delete_release(
    client: httpx.AsyncClient,
    base_url: str,
    headers: dict,
    owner: str,
    repo: str,
    release_id: int,
) -> None:
    """Delete a specific release."""
    url = f"{base_url}{API_BASE}/repos/{owner}/{repo}/releases/{release_id}"

    response = await client.delete(url, headers=headers)
    response.raise_for_status()


def _build_summary(stats: dict, dry_run: bool) -> str:
    """Build a human-readable summary of the cleanup operation."""
    parts = [
        f"Scanned {stats['repos_scanned']} repos",
        f"{stats['repos_with_releases']} with releases",
    ]

    if dry_run:
        parts.append(f"{stats['releases_skipped']} releases would be deleted")
    else:
        parts.append(f"{stats['releases_deleted']} releases deleted")

    if stats["errors"]:
        parts.append(f"{len(stats['errors'])} errors")

    return ", ".join(parts)