feat(forgejo): add package prune script with retention depth 3

List and optionally delete package versions beyond the newest three,
protecting production Helm image tags. Adds Make targets and unit tests
for ACTIVITY-WP-0020.

scripts/forgejo_package_prune.py

@@ -0,0 +1,439 @@
+#!/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 sys
+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 _api_request(
+    method: str,
+    url: str,
+    token: str,
+    *,
+    timeout: float = 30.0,
+) -> Any:
+    req = urllib.request.Request(
+        url,
+        method=method,
+        headers={
+            "Authorization": f"token {token}",
+            "Accept": "application/json",
+        },
+    )
+    with urllib.request.urlopen(req, timeout=timeout) as resp:
+        body = resp.read().decode("utf-8")
+        return json.loads(body) if body else None
+
+
+def _paginate(
+    base_url: str,
+    token: str,
+    path: str,
+    *,
+    page_size: int = 50,
+) -> list[dict[str, Any]]:
+    items: list[dict[str, Any]] = []
+    page = 1
+    while True:
+        query = urllib.parse.urlencode({"limit": page_size, "page": page})
+        url = f"{base_url.rstrip('/')}{path}?{query}"
+        payload = _api_request("GET", url, token)
+        if not payload:
+            break
+        if isinstance(payload, list):
+            batch = payload
+        else:
+            batch = payload.get("data") or payload.get("items") or []
+        if not batch:
+            break
+        items.extend(batch)
+        if len(batch) < page_size:
+            break
+        page += 1
+    return items
+
+
+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 list_versions(
+    base_url: str,
+    token: str,
+    owner: str,
+    package_type: str,
+    name: str,
+) -> list[dict[str, Any]]:
+    owner_q = urllib.parse.quote(owner)
+    type_q = urllib.parse.quote(package_type)
+    name_q = urllib.parse.quote(name, safe="")
+    return _paginate(
+        base_url,
+        token,
+        f"/api/v1/packages/{owner_q}/{type_q}/{name_q}/versions",
+    )
+
+
+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
+
+        for package in packages:
+            name = str(package.get("name") or package.get("package_name") or "")
+            if not name:
+                continue
+            try:
+                versions = list_versions(base_url, token, owner, package_type, name)
+            except urllib.error.HTTPError as exc:
+                errors.append(f"versions {package_type}/{name}: HTTP {exc.code}")
+                continue
+            except Exception as exc:  # noqa: BLE001
+                errors.append(f"versions {package_type}/{name}: {exc}")
+                continue
+
+            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:
+    token = os.environ.get("FORGEJO_TOKEN", "").strip()
+    if token:
+        return token
+    token_file = os.environ.get("FORGEJO_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: set FORGEJO_TOKEN or FORGEJO_TOKEN_FILE "
+        "(Forgejo PAT with read:package and write:package)"
+    )
+
+
+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")
+    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())
+
+    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,
+    )
+
+    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())