blumeops/mise-tasks/container-build-and-release

#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.12"
# dependencies = ["typer==0.25.0", "httpx==0.28.1"]
# ///
#MISE description="Trigger container build workflows via Forgejo API"
#USAGE arg "<container>" help="Container name (directory under containers/)"
#USAGE flag "--ref <ref>" help="Commit SHA or branch to build (defaults to current HEAD)"
#USAGE flag "--dry-run" help="Show what would be done without triggering"
"""Trigger container build workflow via Forgejo API dispatch.

Dispatches the unified build-container workflow, which handles both
Dockerfile and Nix builds in a single workflow.
"""

import subprocess
import sys
import time
from pathlib import Path

import httpx
import typer

REGISTRY = "registry.ops.eblu.me"
FORGE_URL = "https://forge.eblu.me"
FORGE_API = f"{FORGE_URL}/api/v1"
REPO = "eblume/blumeops"
FORGE_ACTIONS = f"{FORGE_URL}/{REPO}/actions"

WORKFLOW = "build-container.yaml"

app = typer.Typer(add_completion=False)


def git(*args: str) -> str:
    result = subprocess.run(
        ["git", *args], capture_output=True, text=True, check=True
    )
    return result.stdout.strip()


def get_forge_token() -> str:
    result = subprocess.run(
        ["op", "read", "op://blumeops/w3663ffnvkewbftncqxtcpeavy/api-token"],
        capture_output=True,
        text=True,
        check=True,
    )
    return result.stdout.strip()


def max_run_number(headers: dict[str, str]) -> int:
    """Return the highest current run_number for WORKFLOW, or 0 if none."""
    resp = httpx.get(
        f"{FORGE_API}/repos/{REPO}/actions/tasks",
        params={"limit": 50},
        headers=headers,
        timeout=15,
    )
    if resp.status_code != 200:
        return 0
    runs = [
        t["run_number"]
        for t in resp.json().get("workflow_runs", [])
        if t.get("workflow_id") == WORKFLOW
    ]
    return max(runs, default=0)


def find_dispatched_run(
    ref: str, floor: int, headers: dict[str, str], timeout_s: int = 20
) -> int | None:
    """Poll the tasks endpoint for the run triggered by our dispatch.

    Matches by head_sha + workflow + run_number > floor so we don't pick up
    an older build of the same commit or a concurrent unrelated dispatch.
    """
    deadline = time.monotonic() + timeout_s
    while time.monotonic() < deadline:
        resp = httpx.get(
            f"{FORGE_API}/repos/{REPO}/actions/tasks",
            params={"limit": 20},
            headers=headers,
            timeout=15,
        )
        if resp.status_code == 200:
            for task in resp.json().get("workflow_runs", []):
                if (
                    task.get("head_sha") == ref
                    and task.get("workflow_id") == WORKFLOW
                    and task.get("run_number", 0) > floor
                ):
                    return task["run_number"]
        time.sleep(1)
    return None


def list_containers() -> None:
    typer.echo("Available containers:")
    for d in sorted(Path("containers").iterdir()):
        if not d.is_dir():
            continue
        types = []
        if (d / "container.py").exists():
            types.append("dagger")
        if (d / "Dockerfile").exists():
            types.append("dockerfile")
        if (d / "default.nix").exists():
            types.append("nix")
        if types:
            typer.echo(f"  - {d.name} ({', '.join(types)})")


@app.command()
def main(
    container: str = typer.Argument(help="Container name (directory under containers/)"),
    ref: str = typer.Option("", "--ref", help="Commit SHA to build (defaults to current HEAD)"),
    dry_run: bool = typer.Option(False, "--dry-run", help="Show what would be done without triggering"),
) -> None:
    """Trigger container build workflows via Forgejo API dispatch."""
    container_dir = Path("containers") / container
    has_container_py = (container_dir / "container.py").exists()
    has_dockerfile = (container_dir / "Dockerfile").exists()
    has_nix = (container_dir / "default.nix").exists()

    if not has_container_py and not has_dockerfile and not has_nix:
        typer.echo(f"Error: No container.py, Dockerfile, or default.nix found in '{container_dir}'")
        typer.echo()
        list_containers()
        raise typer.Exit(1)

    if not ref:
        ref = git("rev-parse", "HEAD")
    else:
        # Resolve short SHAs or branch names to full SHA
        ref = git("rev-parse", ref)

    short_sha = ref[:7]

    image = f"blumeops/{container}"

    # Show expected builds
    builds = []
    if has_container_py or has_dockerfile:
        label = "dagger" if has_container_py else "dockerfile"
        builds.append(f"  {label:12s} -> {REGISTRY}/{image}:v<version>-{short_sha}")
    if has_nix:
        builds.append(f"  {'nix':12s} -> {REGISTRY}/{image}:v<version>-{short_sha}-nix")

    if dry_run:
        typer.echo("[dry-run mode]")
    typer.echo(f"Container:  {container}")
    typer.echo(f"Commit:     {ref} ({short_sha})")
    typer.echo(f"Expected builds:")
    for b in builds:
        typer.echo(b)
    typer.echo()

    if dry_run:
        typer.echo(f"[dry-run] Would dispatch {WORKFLOW}")
        typer.echo()
        typer.echo("Monitor builds with: mise run runner-logs")
        typer.echo(f"           or visit: {FORGE_ACTIONS}")
        return

    token = get_forge_token()
    headers = {
        "Authorization": f"token {token}",
        "Content-Type": "application/json",
    }

    # Verify the commit has been pushed to the remote
    resp = httpx.get(
        f"{FORGE_API}/repos/{REPO}/git/commits/{ref}",
        headers=headers,
        timeout=15,
    )
    if resp.status_code != 200:
        typer.echo(f"Error: commit {short_sha} not found on remote")
        typer.echo("Push your changes before triggering a build: git push origin main")
        raise typer.Exit(1)

    # Snapshot the highest existing run_number so we can identify the one
    # our dispatch creates.
    floor = max_run_number(headers)

    url = f"{FORGE_API}/repos/{REPO}/actions/workflows/{WORKFLOW}/dispatches"
    payload = {
        "ref": "main",
        "inputs": {
            "container": container,
            "ref": ref,
        },
    }
    resp = httpx.post(url, json=payload, headers=headers, timeout=30)
    if resp.status_code == 204:
        typer.echo(f"Dispatched {WORKFLOW}")
    else:
        typer.echo(f"Error dispatching {WORKFLOW}: {resp.status_code} {resp.text}")
        raise typer.Exit(1)

    typer.echo()
    run_number = find_dispatched_run(ref, floor, headers)
    if run_number is not None:
        typer.echo(f"Monitor builds with: mise run runner-logs {run_number}")
    else:
        typer.echo("Monitor builds with: mise run runner-logs")
    typer.echo(f"           or visit: {FORGE_ACTIONS}")


if __name__ == "__main__":
    app()