burrow/Scripts/forgejo-prune-runners.py

#!/usr/bin/env python3
from __future__ import annotations

import json
import os
import pathlib
import subprocess
import time
import urllib.error
import urllib.request


def _read_token() -> str:
    token = os.environ.get("FORGEJO_API_TOKEN", "").strip()
    token_file = os.environ.get("FORGEJO_API_TOKEN_FILE", "").strip()
    if not token and token_file:
        token = pathlib.Path(token_file).read_text().strip()
    if not token:
        raise SystemExit("Forgejo API token is missing")
    if token.startswith("PENDING-"):
        raise SystemExit("Forgejo API token is pending")
    return token


def _request(method: str, url: str, token: str) -> tuple[int, str]:
    headers = {"Authorization": f"token {token}", "Accept": "application/json"}
    req = urllib.request.Request(url, headers=headers, method=method)
    try:
        with urllib.request.urlopen(req, timeout=20) as resp:
            body = resp.read().decode("utf-8")
            return resp.getcode(), body
    except urllib.error.HTTPError as exc:
        body = exc.read().decode("utf-8")
        return exc.code, body


def _list_runners(api_url: str, token: str, org: str | None) -> tuple[str, list[dict]]:
    if org:
        list_url = f"{api_url}/orgs/{org}/actions/runners"
    else:
        list_url = f"{api_url}/actions/runners"
    status, body = _request("GET", list_url, token)
    if status == 404:
        return list_url, []
    if status >= 400:
        raise RuntimeError(f"list runners failed ({status}) {body}")
    try:
        runners = json.loads(body)
    except json.JSONDecodeError as exc:
        raise RuntimeError(f"invalid runner list response: {exc}") from exc
    if not isinstance(runners, list):
        raise RuntimeError("runner list response is not a list")
    return list_url, runners


def _delete_runner(api_url: str, token: str, org: str | None, runner_id: int) -> bool:
    if org:
        delete_url = f"{api_url}/orgs/{org}/actions/runners/{runner_id}"
    else:
        delete_url = f"{api_url}/actions/runners/{runner_id}"
    status, body = _request("DELETE", delete_url, token)
    if status in (200, 204):
        return True
    print(f"[forgejo-prune-runners] delete {runner_id} failed: {status} {body}")
    return False


def _prune_db(ttl_seconds: int) -> int:
    cutoff = int(time.time()) - ttl_seconds
    now = int(time.time())
    sql = (
        "WITH updated AS ("
        "UPDATE action_runner "
        f"SET deleted = {now} "
        "WHERE (deleted IS NULL OR deleted = 0) "
        f"AND ((last_online IS NOT NULL AND last_online > 0 AND last_online < {cutoff}) "
        f"OR (COALESCE(last_online, 0) = 0 AND created < {cutoff})) "
        "RETURNING 1"
        ") SELECT count(*) FROM updated;"
    )
    result = subprocess.run(
        ["psql", "-h", "/run/postgresql", "-U", "forgejo", "forgejo", "-tAc", sql],
        check=True,
        capture_output=True,
        text=True,
    )
    output = (result.stdout or "").strip()
    try:
        return int(output)
    except ValueError:
        return 0


def main() -> None:
    api_url = os.environ.get("FORGEJO_API_URL", "https://git.burrow.net/api/v1").rstrip("/")
    org = os.environ.get("FORGEJO_ORG", "hackclub").strip() or None
    dry_run = os.environ.get("FORGEJO_DRY_RUN", "0") == "1"
    db_only = os.environ.get("FORGEJO_PRUNE_DB", "0") == "1"
    ttl_seconds = int(os.environ.get("FORGEJO_RUNNER_TTL_SEC", "3600"))

    if db_only:
        removed = _prune_db(ttl_seconds)
        print(f"[forgejo-prune-runners] pruned {removed} runners via DB")
        return

    token = _read_token()

    try:
        _, runners = _list_runners(api_url, token, org)
    except RuntimeError as exc:
        if org is not None:
            print(f"[forgejo-prune-runners] org runner list failed ({exc}); retrying instance scope")
            _, runners = _list_runners(api_url, token, None)
            org = None
        else:
            raise SystemExit(str(exc))

    if not runners:
        removed = _prune_db(ttl_seconds)
        print(f"[forgejo-prune-runners] pruned {removed} runners via DB fallback")
        return

    removed = 0
    for runner in runners:
        runner_id = runner.get("id")
        name = runner.get("name", "unknown")
        status = (runner.get("status") or "").lower()
        busy = bool(runner.get("busy"))
        if status == "online" or busy:
            continue
        if runner_id is None:
            continue
        if dry_run:
            print(f"[forgejo-prune-runners] would delete runner {runner_id} ({name}) status={status}")
            continue
        if _delete_runner(api_url, token, org, int(runner_id)):
            removed += 1
            print(f"[forgejo-prune-runners] deleted runner {runner_id} ({name})")

    print(f"[forgejo-prune-runners] done; removed {removed} runners")


if __name__ == "__main__":
    main()