railiance-platform/scripts/forgejo_package_prune.py

#!/usr/bin/env python3
"""Forgejo package retention prune — keep newest N versions per package."""

from __future__ import annotations

import argparse
import json
import os
import re
import shutil
import subprocess
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from typing import Any

DEFAULT_BASE = "https://forgejo.coulomb.social"
DEFAULT_OWNER = "coulomb"
DEFAULT_TYPES = ("container", "pypi", "npm", "generic")
DEFAULT_MAX_VERSIONS = 3
DEFAULT_APPS_ROOT = Path.home() / "railiance-apps"
FORGEJO_IMAGE_RE = re.compile(
    r"^forgejo\.coulomb\.social/(?:coulomb/)?(?P<name>[^:/]+)(?::(?P<tag>[^/\s]+))?$",
    re.IGNORECASE,
)


@dataclass(frozen=True)
class VersionRef:
    package_type: str
    name: str
    version: str

    def key(self) -> tuple[str, str, str]:
        return (self.package_type, self.name, self.version)


@dataclass(frozen=True)
class DeletePlan:
    package_type: str
    name: str
    version: str
    created_at: str
    protected: bool
    reason: str


def _parse_created_at(value: str | None) -> datetime:
    if not value:
        return datetime.min
    try:
        return datetime.fromisoformat(value.replace("Z", "+00:00"))
    except ValueError:
        return datetime.min


def collect_protected_versions(apps_root: Path) -> set[tuple[str, str, str]]:
    protected: set[tuple[str, str, str]] = set()
    if not apps_root.is_dir():
        return protected

    patterns = [
        apps_root / "helm" / "*-values.yaml",
        apps_root / "charts" / "*" / "values.yaml",
    ]
    paths: list[Path] = []
    for pattern in patterns:
        paths.extend(sorted(pattern.parent.glob(pattern.name)))

    try:
        import yaml  # type: ignore
    except ImportError:
        yaml = None

    for path in paths:
        text = path.read_text(encoding="utf-8")
        if yaml is not None:
            try:
                data = yaml.safe_load(text) or {}
            except Exception:
                data = {}
            image = data.get("image") if isinstance(data, dict) else None
            if isinstance(image, dict):
                repo = str(image.get("repository") or "").strip()
                tag = str(image.get("tag") or "").strip()
                if repo and tag:
                    match = FORGEJO_IMAGE_RE.match(repo) or FORGEJO_IMAGE_RE.match(
                        f"{repo}:{tag}"
                    )
                    if match:
                        name = match.group("name")
                        protected.add(("container", name, tag))
            continue

        repo_match = re.search(
            r"repository:\s*forgejo\.coulomb\.social/coulomb/([^\s]+)",
            text,
            re.IGNORECASE,
        )
        tag_match = re.search(r'^\s*tag:\s*"?([^"\s#]+)"?\s*$', text, re.MULTILINE)
        if repo_match and tag_match:
            protected.add(("container", repo_match.group(1), tag_match.group(1)))

    return protected


def collect_live_cluster_versions(
    *, kubectl: str = "kubectl", timeout: float = 60.0
) -> tuple[set[tuple[str, str, str]], list[str]]:
    """Protect image tags currently running in the cluster (best-effort).

    Enumerates all pod container images across namespaces via kubectl and
    protects any `forgejo.coulomb.social/coulomb/<name>:<tag>`. This closes the
    gap where a live deployment pins a tag not declared in Helm values (e.g.
    CI-deployed apps). Failures (no kubectl, no cluster access) return an empty
    set with a note — pruning a reachable registry must not hard-depend on
    cluster access, but the note surfaces reduced protection coverage.
    """
    protected: set[tuple[str, str, str]] = set()
    if shutil.which(kubectl) is None:
        return protected, ["live-tag protection skipped: kubectl not found"]
    jsonpath = (
        "{range .items[*]}"
        "{range .spec.containers[*]}{.image}{'\\n'}{end}"
        "{range .spec.initContainers[*]}{.image}{'\\n'}{end}"
        "{end}"
    )
    try:
        result = subprocess.run(
            [kubectl, "get", "pods", "--all-namespaces", "-o", f"jsonpath={jsonpath}"],
            capture_output=True,
            text=True,
            timeout=timeout,
            check=True,
        )
    except Exception as exc:  # noqa: BLE001
        return protected, [f"live-tag protection skipped: kubectl query failed ({exc})"]
    for line in result.stdout.splitlines():
        image = line.strip()
        if not image:
            continue
        match = FORGEJO_IMAGE_RE.match(image)
        if match and match.group("tag"):
            protected.add(("container", match.group("name"), match.group("tag")))
    return protected, []


def _api_request(
    method: str,
    url: str,
    token: str,
    *,
    timeout: float = 60.0,
    retries: int = 2,
) -> Any:
    req = urllib.request.Request(
        url,
        method=method,
        headers={
            "Authorization": f"token {token}",
            "Accept": "application/json",
        },
    )
    last_exc: Exception | None = None
    for attempt in range(retries + 1):
        try:
            with urllib.request.urlopen(req, timeout=timeout) as resp:
                body = resp.read().decode("utf-8")
                return json.loads(body) if body else None
        except urllib.error.HTTPError:
            raise  # 4xx/5xx are real responses — surface them, do not retry
        except (urllib.error.URLError, TimeoutError, OSError) as exc:
            # Transient: connection reset, read timeout, TLS handshake timeout.
            last_exc = exc
            if attempt < retries:
                time.sleep(2 * (attempt + 1))
                continue
            raise
    if last_exc:  # pragma: no cover - defensive
        raise last_exc


def list_packages(
    base_url: str,
    token: str,
    owner: str,
    package_type: str,
) -> list[dict[str, Any]]:
    owner_q = urllib.parse.quote(owner)
    items: list[dict[str, Any]] = []
    page = 1
    page_size = 50
    while True:
        query = urllib.parse.urlencode(
            {"limit": page_size, "page": page, "type": package_type}
        )
        url = f"{base_url.rstrip('/')}/api/v1/packages/{owner_q}?{query}"
        payload = _api_request("GET", url, token)
        batch = payload if isinstance(payload, list) else []
        if not batch:
            break
        items.extend(batch)
        if len(batch) < page_size:
            break
        page += 1
    return items


def delete_version(
    base_url: str,
    token: str,
    owner: str,
    package_type: str,
    name: str,
    version: str,
    *,
    dry_run: bool,
) -> None:
    owner_q = urllib.parse.quote(owner)
    type_q = urllib.parse.quote(package_type)
    name_q = urllib.parse.quote(name, safe="")
    version_q = urllib.parse.quote(version, safe="")
    path = f"/api/v1/packages/{owner_q}/{type_q}/{name_q}/{version_q}"
    if dry_run:
        return
    url = f"{base_url.rstrip('/')}{path}"
    _api_request("DELETE", url, token)


def build_delete_plans(
    *,
    base_url: str,
    token: str,
    owner: str,
    package_types: list[str],
    max_versions: int,
    protected: set[tuple[str, str, str]],
) -> tuple[list[DeletePlan], list[str]]:
    plans: list[DeletePlan] = []
    errors: list[str] = []

    for package_type in package_types:
        try:
            packages = list_packages(base_url, token, owner, package_type)
        except urllib.error.HTTPError as exc:
            errors.append(f"list {package_type}: HTTP {exc.code}")
            continue
        except Exception as exc:  # noqa: BLE001
            errors.append(f"list {package_type}: {exc}")
            continue

        # Forgejo's package list endpoint returns one entry per (name, version).
        # Group by package name; each group is that package's version set — there
        # is no separate per-package "/versions" endpoint.
        by_name: dict[str, list[dict[str, Any]]] = {}
        for package in packages:
            name = str(package.get("name") or package.get("package_name") or "")
            if not name:
                continue
            by_name.setdefault(name, []).append(package)

        for name, versions in by_name.items():
            sorted_versions = sorted(
                versions,
                key=lambda item: _parse_created_at(str(item.get("created_at") or "")),
                reverse=True,
            )
            keep = {
                str(item.get("version") or "")
                for item in sorted_versions[:max_versions]
                if str(item.get("version") or "")
            }
            for item in sorted_versions[max_versions:]:
                version = str(item.get("version") or "")
                if not version or version in keep:
                    continue
                key = (package_type, name, version)
                is_protected = key in protected
                plans.append(
                    DeletePlan(
                        package_type=package_type,
                        name=name,
                        version=version,
                        created_at=str(item.get("created_at") or ""),
                        protected=is_protected,
                        reason="protected_production_tag" if is_protected else "beyond_retention_depth",
                    )
                )
    return plans, errors


def load_token() -> str:
    for env_name in ("FORGEJO_TOKEN", "FORGEJO_ADMIN_TOKEN"):
        token = os.environ.get(env_name, "").strip()
        if token:
            return token
    token_file = os.environ.get(
        "FORGEJO_TOKEN_FILE",
        os.environ.get("FORGEJO_ADMIN_TOKEN_FILE", "/tmp/forgejo-tegwick-api-token"),
    )
    path = Path(token_file).expanduser()
    if path.is_file():
        return path.read_text(encoding="utf-8").strip()
    raise SystemExit(
        "ERROR: Forgejo API token required (read:package + write:package).\n"
        "  export FORGEJO_TOKEN=<pat>     # or FORGEJO_ADMIN_TOKEN\n"
        "  # or save PAT to /tmp/forgejo-tegwick-api-token (chmod 600)\n"
        "  Generate: https://forgejo.coulomb.social/user/settings/applications\n"
        "  See: railiance-platform/docs/forgejo-package-prune.md"
    )


def emit_summary(
    *,
    owner: str,
    max_versions: int,
    package_types: list[str],
    protected: set[tuple[str, str, str]],
    plans: list[DeletePlan],
    errors: list[str],
    apply: bool,
    deleted: list[DeletePlan],
) -> dict[str, Any]:
    would_delete = [p for p in plans if not p.protected]
    skipped_protected = [p for p in plans if p.protected]
    return {
        "kind": "forgejo_package_prune",
        "owner": owner,
        "max_versions": max_versions,
        "package_types": package_types,
        "protected_count": len(protected),
        "candidate_count": len(would_delete),
        "skipped_protected_count": len(skipped_protected),
        "deleted_count": len(deleted),
        "apply": apply,
        "would_delete": [
            {
                "type": p.package_type,
                "name": p.name,
                "version": p.version,
                "created_at": p.created_at,
            }
            for p in would_delete
        ],
        "skipped_protected": [
            {
                "type": p.package_type,
                "name": p.name,
                "version": p.version,
                "reason": p.reason,
            }
            for p in skipped_protected
        ],
        "deleted": [
            {
                "type": p.package_type,
                "name": p.name,
                "version": p.version,
            }
            for p in deleted
        ],
        "errors": errors,
    }


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description="Prune old Forgejo package versions")
    parser.add_argument("--owner", default=DEFAULT_OWNER)
    parser.add_argument("--base-url", default=os.environ.get("FORGEJO_BASE_URL", DEFAULT_BASE))
    parser.add_argument("--max-versions", type=int, default=DEFAULT_MAX_VERSIONS)
    parser.add_argument(
        "--types",
        default=",".join(DEFAULT_TYPES),
        help="Comma-separated package types",
    )
    parser.add_argument("--apps-root", type=Path, default=DEFAULT_APPS_ROOT)
    parser.add_argument("--apply", action="store_true")
    parser.add_argument("--json", action="store_true", help="Emit JSON summary on stdout")
    parser.add_argument(
        "--no-protect-live",
        dest="protect_live",
        action="store_false",
        help="Skip protecting image tags currently running in the cluster (kubectl)",
    )
    parser.set_defaults(protect_live=True)
    args = parser.parse_args(argv)

    apply = bool(args.apply)
    dry_run = not apply
    package_types = [part.strip() for part in args.types.split(",") if part.strip()]
    token = load_token()
    protected = collect_protected_versions(args.apps_root.expanduser())
    protect_notes: list[str] = []
    if args.protect_live:
        live, protect_notes = collect_live_cluster_versions()
        protected |= live
        print(f"Protected live cluster tags: {len(live)}", file=sys.stderr)
        for note in protect_notes:
            print(f"  WARN: {note}", file=sys.stderr)

    print(f"Forgejo package prune — owner={args.owner} keep={args.max_versions}", file=sys.stderr)
    print(f"Protected production tags: {len(protected)}", file=sys.stderr)

    plans, errors = build_delete_plans(
        base_url=args.base_url,
        token=token,
        owner=args.owner,
        package_types=package_types,
        max_versions=max(1, args.max_versions),
        protected=protected,
    )

    deleted: list[DeletePlan] = []
    for plan in plans:
        if plan.protected:
            print(
                f"  skip protected  {plan.package_type}/{plan.name}:{plan.version}",
                file=sys.stderr,
            )
            continue
        if dry_run:
            print(
                f"  would delete    {plan.package_type}/{plan.name}:{plan.version}",
                file=sys.stderr,
            )
            continue
        try:
            delete_version(
                args.base_url,
                token,
                args.owner,
                plan.package_type,
                plan.name,
                plan.version,
                dry_run=False,
            )
            deleted.append(plan)
            print(
                f"  deleted         {plan.package_type}/{plan.name}:{plan.version}",
                file=sys.stderr,
            )
        except urllib.error.HTTPError as exc:
            errors.append(f"delete {plan.package_type}/{plan.name}:{plan.version}: HTTP {exc.code}")
        except Exception as exc:  # noqa: BLE001
            errors.append(f"delete {plan.package_type}/{plan.name}:{plan.version}: {exc}")

    summary = emit_summary(
        owner=args.owner,
        max_versions=args.max_versions,
        package_types=package_types,
        protected=protected,
        plans=plans,
        errors=errors,
        apply=apply,
        deleted=deleted,
    )
    summary["live_protection"] = args.protect_live
    summary["protection_notes"] = protect_notes

    if args.json or not sys.stdout.isatty():
        print(json.dumps(summary, indent=2))
    else:
        print(json.dumps(summary, indent=2))

    return 1 if errors else 0


if __name__ == "__main__":
    raise SystemExit(main())