#!/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"
DEFAULT_FORGEJO_ADMIN_BAO_PATH = "platform/workloads/forgejo/forgejo-admin"
DEFAULT_FORGEJO_ADMIN_BAO_FIELD = "API_TOKEN"
LEGACY_FORGEJO_TOKEN_FILE = Path("/tmp/forgejo-tegwick-api-token")
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_images_from_files(
    paths: list[Path],
) -> tuple[set[tuple[str, str, str]], list[str]]:
    """Protect image tags listed in exported live-image files.

    Each file holds one image ref per line (`kubectl get pods ... jsonpath`
    output from another cluster). This closes the multi-cluster gap
    (ACTIVITY-WP-0020-T07): the prune host's kubectl only sees its own
    cluster, so every other production cluster exports its live images to a
    file that is merged here. A missing file is a WARN, not a failure —
    but it means reduced protection coverage, so the note must surface.
    """
    protected: set[tuple[str, str, str]] = set()
    notes: list[str] = []
    for raw_path in paths:
        path = raw_path.expanduser()
        if not path.is_file():
            notes.append(f"live-images file missing: {path}")
            continue
        for line in path.read_text(encoding="utf-8").splitlines():
            image = line.strip()
            if not image or image.startswith("#"):
                continue
            match = FORGEJO_IMAGE_RE.match(image)
            if match and match.group("tag"):
                protected.add(("container", match.group("name"), match.group("tag")))
    return protected, notes


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 _read_token_file(path: Path) -> str:
    return path.read_text(encoding="utf-8").strip()


def _truthy_env(name: str) -> bool:
    return os.environ.get(name, "").strip().lower() in {"1", "true", "yes", "on"}


def _load_token_from_file_env() -> str | None:
    for env_name in ("FORGEJO_TOKEN_FILE", "FORGEJO_ADMIN_TOKEN_FILE"):
        raw_path = os.environ.get(env_name, "").strip()
        if not raw_path:
            continue
        path = Path(raw_path).expanduser()
        if not path.is_file():
            raise SystemExit(f"ERROR: {env_name} points to a missing file: {path}")
        token = _read_token_file(path)
        if token:
            return token
        raise SystemExit(f"ERROR: {env_name} points to an empty file: {path}")
    return None


def _load_token_from_openbao() -> tuple[str | None, str | None]:
    bao_bin = os.environ.get("FORGEJO_ADMIN_BAO_CLI", "bao").strip() or "bao"
    bao_path = (
        os.environ.get("FORGEJO_ADMIN_BAO_PATH", DEFAULT_FORGEJO_ADMIN_BAO_PATH).strip()
        or DEFAULT_FORGEJO_ADMIN_BAO_PATH
    )
    bao_field = (
        os.environ.get("FORGEJO_ADMIN_BAO_FIELD", DEFAULT_FORGEJO_ADMIN_BAO_FIELD).strip()
        or DEFAULT_FORGEJO_ADMIN_BAO_FIELD
    )
    if shutil.which(bao_bin) is None:
        return None, f"{bao_bin} CLI not found"
    try:
        result = subprocess.run(
            [bao_bin, "kv", "get", f"-field={bao_field}", bao_path],
            capture_output=True,
            text=True,
            check=True,
        )
    except subprocess.CalledProcessError as exc:
        detail = exc.stderr.strip() or exc.stdout.strip() or f"exit {exc.returncode}"
        return None, f"{bao_bin} kv get failed: {detail}"
    except OSError as exc:
        return None, f"{bao_bin} invocation failed: {exc}"
    token = result.stdout.strip()
    if not token:
        return None, f"{bao_bin} kv get returned an empty {bao_field} field"
    return token, None


def _token_help_message(bao_error: str | None) -> str:
    lines = [
        "ERROR: Forgejo API token required (read:package + write:package).",
        "  Primary: bao login -method=oidc -path=netkingdom role=forgejo-admin-workload-kv-read",
        f"  Default lane: {DEFAULT_FORGEJO_ADMIN_BAO_PATH} field {DEFAULT_FORGEJO_ADMIN_BAO_FIELD}",
        "  Override lane via FORGEJO_ADMIN_BAO_PATH / FORGEJO_ADMIN_BAO_FIELD if needed.",
        "  Break-glass: set FORGEJO_TOKEN / FORGEJO_ADMIN_TOKEN, or set FORGEJO_TOKEN_FILE / FORGEJO_ADMIN_TOKEN_FILE.",
        "  Legacy /tmp fallback stays opt-in only via FORGEJO_ALLOW_LEGACY_FILE_FALLBACK=1.",
        "  See: railiance-platform/docs/forgejo-package-prune.md",
    ]
    if bao_error:
        lines.insert(3, f"  OpenBao lookup failed: {bao_error}")
    return "\n".join(lines)


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 = _load_token_from_file_env()
    if token:
        return token
    token, bao_error = _load_token_from_openbao()
    if token:
        return token

    if _truthy_env("FORGEJO_ALLOW_LEGACY_FILE_FALLBACK"):
        if LEGACY_FORGEJO_TOKEN_FILE.is_file():
            token = _read_token_file(LEGACY_FORGEJO_TOKEN_FILE)
            if token:
                return token
            suffix = f"legacy file {LEGACY_FORGEJO_TOKEN_FILE} is empty"
        else:
            suffix = f"legacy file {LEGACY_FORGEJO_TOKEN_FILE} is missing"
        bao_error = f"{bao_error}; {suffix}" if bao_error else suffix

    raise SystemExit(_token_help_message(bao_error))


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)
    parser.add_argument(
        "--live-images-file",
        dest="live_images_files",
        type=Path,
        action="append",
        default=[],
        help=(
            "File with one image ref per line, exported from another "
            "production cluster; repeatable. Tags matching the Forgejo "
            "registry are protected in addition to the local kubectl scan."
        ),
    )
    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)
    if args.live_images_files:
        file_live, file_notes = collect_live_images_from_files(args.live_images_files)
        protected |= file_live
        protect_notes.extend(file_notes)
        print(f"Protected exported live tags: {len(file_live)}", file=sys.stderr)
        for note in file_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())