blumeops/mise-tasks/spork-create

#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.12"
# dependencies = ["httpx>=0.28.1", "rich>=14.0.0", "typer>=0.24.0"]
# ///
#MISE description="Create a spork (floating-branch soft-fork) of a mirrored upstream project"
#USAGE arg "<repo_name>" help="Repository name in the mirrors/ org on forge (e.g. kingfisher)"
#USAGE flag "--description <description>" help="Repository description override"
#USAGE flag "--main-branch <main_branch>" help="Name of the upstream main branch (default: auto-detect)"
#USAGE flag "--dry-run" help="Show what would be done without creating"
#USAGE flag "--no-clone" help="Skip cloning to ~/code/3rd/"
"""Create a spork of a mirrored upstream project.

A "spork" is a floating-branch soft-fork strategy. It creates a mutable
clone of a read-only mirror in the eblume/ org on Forge, sets up a
'blumeops' branch with a mirror-sync workflow, and optionally clones
locally with the three-remote setup (origin, mirror, upstream).

Prerequisites:
  - Mirror must already exist at mirrors/<repo_name> on forge
  - 1Password CLI authenticated (for Forge API token)

See docs/explanation/spork-strategy.md for the full strategy.
"""

import json
import subprocess
import sys
import textwrap
from pathlib import Path
from typing import Annotated, Optional

import httpx
import typer
from rich.console import Console

FORGE_URL = "https://forge.eblu.me"
FORGE_API = f"{FORGE_URL}/api/v1"
FORGE_SSH = "ssh://forgejo@forge.ops.eblu.me:2222"
MIRROR_ORG = "mirrors"
OWNER = "eblume"
LOCAL_BASE = Path.home() / "code" / "3rd"
OP_TOKEN_REF = "op://blumeops/w3663ffnvkewbftncqxtcpeavy/api-token"

console = Console()
app = typer.Typer(add_completion=False)


def op_read(ref: str) -> str:
    """Read a secret from 1Password."""
    result = subprocess.run(
        ["op", "read", ref], capture_output=True, text=True, check=True
    )
    return result.stdout.strip()


def forge_api(
    client: httpx.Client,
    method: str,
    path: str,
    token: str,
    json_data: dict | None = None,
) -> httpx.Response:
    """Make an authenticated Forge API request."""
    return client.request(
        method,
        f"{FORGE_API}{path}",
        headers={"Authorization": f"token {token}"},
        json=json_data,
    )


def get_mirror_info(client: httpx.Client, token: str, name: str) -> dict:
    """Get mirror repo info, or exit if it doesn't exist."""
    resp = forge_api(client, "GET", f"/repos/{MIRROR_ORG}/{name}", token)
    if resp.status_code == 404:
        console.print(
            f"[red]Error:[/red] Mirror [bold]{MIRROR_ORG}/{name}[/bold] not found on forge."
        )
        console.print("Create it first: [dim]mise run mirror-create <upstream-url>[/dim]")
        raise SystemExit(1)
    resp.raise_for_status()
    return resp.json()


def detect_main_branch(client: httpx.Client, token: str, name: str) -> str:
    """Detect the default branch of the mirror."""
    info = get_mirror_info(client, token, name)
    return info.get("default_branch", "main")


def repo_exists(client: httpx.Client, token: str, owner: str, name: str) -> bool:
    """Check if a repo already exists."""
    resp = forge_api(client, "GET", f"/repos/{owner}/{name}", token)
    return resp.status_code == 200


def fork_mirror(client: httpx.Client, token: str, name: str) -> dict:
    """Fork the mirror into the eblume org."""
    resp = forge_api(
        client,
        "POST",
        f"/repos/{MIRROR_ORG}/{name}/forks",
        token,
        json_data={"name": name},
    )
    if resp.status_code == 409:
        console.print(f"[yellow]Fork already exists:[/yellow] {OWNER}/{name}")
        return forge_api(client, "GET", f"/repos/{OWNER}/{name}", token).json()
    resp.raise_for_status()
    return resp.json()


def mirror_sync_workflow(main_branch: str, repo_name: str) -> str:
    """Generate the mirror-sync workflow YAML."""
    return textwrap.dedent(f"""\
        # Mirror Sync — Spork Strategy
        #
        # Keeps the '{main_branch}' branch tracking upstream (via mirror) and
        # rebases the 'blumeops' branch on top. See docs/explanation/spork-strategy.md
        # in the blumeops repo for the full strategy.
        #
        # On conflict: the workflow fails. Manual rebase resolution required.

        name: Mirror Sync

        on:
          schedule:
            - cron: '0 5 * * *'  # Daily at 05:00 UTC
          workflow_dispatch:

        jobs:
          sync:
            runs-on: k8s
            steps:
              - name: Checkout blumeops branch
                uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
                with:
                  ref: blumeops
                  fetch-depth: 0

              - name: Configure git
                run: |
                  git config user.name "Forgejo Actions"
                  git config user.email "actions@forge.eblu.me"

              - name: Add mirror remote
                run: |
                  git remote add mirror "${{{{ env.MIRROR_URL }}}}" || true
                  git fetch mirror
                env:
                  MIRROR_URL: {FORGE_URL}/{MIRROR_ORG}/{repo_name}.git

              - name: Fast-forward {main_branch} from mirror
                run: |
                  git checkout -B {main_branch} origin/{main_branch}
                  git merge --ff-only mirror/{main_branch}
                  git push origin {main_branch}

              - name: Rebase blumeops onto {main_branch}
                run: |
                  git checkout blumeops
                  git rebase {main_branch}
                  git push --force-with-lease origin blumeops

              - name: Rebase feature branches
                run: |
                  # Rebase feature/local/* onto blumeops
                  for branch in $(git branch -r --list 'origin/feature/local/*'); do
                    local_name="${{branch#origin/}}"
                    echo "Rebasing $local_name onto blumeops..."
                    git checkout -B "$local_name" "$branch"
                    git rebase blumeops || {{
                      echo "::error::Rebase conflict on $local_name"
                      git rebase --abort
                      continue
                    }}
                    git push --force-with-lease origin "$local_name"
                  done

                  # Rebase feature/upstream/* onto {main_branch}
                  for branch in $(git branch -r --list 'origin/feature/upstream/*'); do
                    local_name="${{branch#origin/}}"
                    echo "Rebasing $local_name onto {main_branch}..."
                    git checkout -B "$local_name" "$branch"
                    git rebase {main_branch} || {{
                      echo "::error::Rebase conflict on $local_name"
                      git rebase --abort
                      continue
                    }}
                    git push --force-with-lease origin "$local_name"
                  done

              - name: Build deploy branch
                run: |
                  git checkout -B deploy blumeops

                  # Merge all feature branches into deploy
                  for branch in $(git branch -r --list 'origin/feature/local/*' 'origin/feature/upstream/*'); do
                    local_name="${{branch#origin/}}"
                    echo "Merging $local_name into deploy..."
                    git merge --no-ff "$local_name" -m "deploy: merge $local_name" || {{
                      echo "::error::Merge conflict on $local_name into deploy"
                      git merge --abort
                      continue
                    }}
                  done

                  git push --force-with-lease origin deploy
    """)


def set_default_branch(
    client: httpx.Client, token: str, owner: str, name: str, branch: str
) -> None:
    """Set the default branch of a repo."""
    resp = forge_api(
        client,
        "PATCH",
        f"/repos/{owner}/{name}",
        token,
        json_data={"default_branch": branch},
    )
    resp.raise_for_status()


def get_upstream_url(client: httpx.Client, token: str, name: str) -> str | None:
    """Try to find the upstream URL from the mirror's config."""
    info = get_mirror_info(client, token, name)
    return info.get("original_url") or info.get("clone_url")


@app.command()
def main(
    repo_name: Annotated[str, typer.Argument(help="Repository name in mirrors/ org")],
    description: Annotated[Optional[str], typer.Option(help="Description override")] = None,
    main_branch: Annotated[Optional[str], typer.Option("--main-branch", help="Upstream main branch name")] = None,
    dry_run: Annotated[bool, typer.Option("--dry-run", help="Preview without creating")] = False,
    no_clone: Annotated[bool, typer.Option("--no-clone", help="Skip local clone")] = False,
) -> None:
    """Create a spork of a mirrored upstream project."""
    console.print(f"[bold]Sporking[/bold] {MIRROR_ORG}/{repo_name}")
    console.print()

    # --- Preflight checks (fail fast before any mutations) ---
    local_path = LOCAL_BASE / repo_name
    if not no_clone and local_path.exists():
        console.print(
            f"[red]Error:[/red] Local directory already exists: [bold]{local_path}[/bold]"
        )
        console.print(
            "Remove it first or use [dim]--no-clone[/dim] to skip local setup."
        )
        raise SystemExit(1)

    token = op_read(OP_TOKEN_REF)

    with httpx.Client(timeout=30) as client:
        # Verify mirror exists
        mirror_info = get_mirror_info(client, token, repo_name)
        detected_main = main_branch or mirror_info.get("default_branch", "main")
        upstream_url = mirror_info.get("original_url", "unknown")
        desc = description or mirror_info.get("description", "")

        # Verify fork doesn't already exist
        if repo_exists(client, token, OWNER, repo_name):
            console.print(
                f"[red]Error:[/red] Fork already exists: [bold]{OWNER}/{repo_name}[/bold]"
            )
            console.print(
                "If re-sporking, delete the fork on forge first."
            )
            raise SystemExit(1)

        console.print(f"  Mirror:       {MIRROR_ORG}/{repo_name}")
        console.print(f"  Upstream:     {upstream_url}")
        console.print(f"  Main branch:  {detected_main}")
        console.print(f"  Description:  {desc}")
        console.print(f"  Fork target:  {OWNER}/{repo_name}")
        if not no_clone:
            console.print(f"  Local clone:  {local_path}")
        console.print()

        if dry_run:
            console.print("[dim][dry-run] Would perform the following:[/dim]")
            console.print(f"  1. Fork {MIRROR_ORG}/{repo_name} → {OWNER}/{repo_name}")
            console.print(f"  2. Create 'blumeops' branch from '{detected_main}'")
            console.print("  3. Add .forgejo/workflows/mirror-sync.yaml")
            console.print("  4. Set 'blumeops' as default branch")
            if not no_clone:
                console.print(f"  5. Clone to {local_path} with 3 remotes")
            return

        # --- All checks passed, start mutating ---
        console.print("Forking mirror...")
        fork_mirror(client, token, repo_name)
        console.print(f"[green]Created fork:[/green] {OWNER}/{repo_name}")

        # Enable Actions (forks from mirrors have it disabled by default)
        console.print("Enabling Actions...")
        resp = forge_api(
            client,
            "PATCH",
            f"/repos/{OWNER}/{repo_name}",
            token,
            json_data={"has_actions": True},
        )
        resp.raise_for_status()
        console.print("[green]Actions enabled[/green]")

        # 3. Clone to temp dir, create blumeops branch with workflow
        console.print("Setting up blumeops branch...")
        import tempfile

        with tempfile.TemporaryDirectory() as tmpdir:
            clone_url = f"{FORGE_SSH}/{OWNER}/{repo_name}.git"
            subprocess.run(
                ["git", "clone", clone_url, tmpdir],
                check=True,
                capture_output=True,
            )

            # Create blumeops branch from detected main
            subprocess.run(
                ["git", "checkout", "-b", "blumeops", f"origin/{detected_main}"],
                cwd=tmpdir,
                check=True,
                capture_output=True,
            )

            # Remove any upstream .forgejo/ directory (rare, but possible)
            existing_forgejo = Path(tmpdir) / ".forgejo"
            if existing_forgejo.exists():
                import shutil
                shutil.rmtree(existing_forgejo)
                console.print("[yellow]Removed upstream .forgejo/ directory[/yellow]")

            # Add mirror-sync workflow
            workflow_dir = Path(tmpdir) / ".forgejo" / "workflows"
            workflow_dir.mkdir(parents=True, exist_ok=True)
            workflow_path = workflow_dir / "mirror-sync.yaml"
            workflow_path.write_text(mirror_sync_workflow(detected_main, repo_name))

            # Commit and push
            subprocess.run(
                ["git", "add", ".forgejo/"],
                cwd=tmpdir,
                check=True,
                capture_output=True,
            )
            subprocess.run(
                ["git", "config", "user.name", "Erich Blume"],
                cwd=tmpdir,
                check=True,
                capture_output=True,
            )
            subprocess.run(
                ["git", "config", "user.email", "blume.erich@gmail.com"],
                cwd=tmpdir,
                check=True,
                capture_output=True,
            )
            subprocess.run(
                ["git", "commit", "-m", "spork: add mirror-sync workflow\n\nBootstrap the blumeops branch with the spork mirror-sync\nworkflow. See blumeops docs/explanation/spork-strategy.md."],
                cwd=tmpdir,
                check=True,
                capture_output=True,
            )
            subprocess.run(
                ["git", "push", "-u", "origin", "blumeops"],
                cwd=tmpdir,
                check=True,
                capture_output=True,
            )
            console.print("[green]Created and pushed blumeops branch[/green]")

        # 4. Set default branch to blumeops
        console.print("Setting default branch to blumeops...")
        set_default_branch(client, token, OWNER, repo_name, "blumeops")
        console.print("[green]Default branch set to blumeops[/green]")

    # 5. Local clone with three remotes
    if no_clone:
        console.print("[dim]Skipping local clone (--no-clone)[/dim]")
    else:
        console.print(f"Cloning to {local_path}...")
        clone_url = f"{FORGE_SSH}/{OWNER}/{repo_name}.git"
        subprocess.run(
            ["git", "clone", clone_url, str(local_path)],
            check=True,
            capture_output=True,
        )
        # Add mirror and upstream remotes
        mirror_url = f"{FORGE_SSH}/{MIRROR_ORG}/{repo_name}.git"
        subprocess.run(
            ["git", "remote", "add", "mirror", mirror_url],
            cwd=local_path,
            check=True,
            capture_output=True,
        )
        if upstream_url and upstream_url != "unknown":
            subprocess.run(
                ["git", "remote", "add", "upstream", upstream_url],
                cwd=local_path,
                check=True,
                capture_output=True,
            )
        subprocess.run(
            ["git", "fetch", "--all"],
            cwd=local_path,
            check=True,
            capture_output=True,
        )
        console.print(f"[green]Local clone ready at {local_path}[/green]")

    # Summary
    console.print()
    console.print("[bold green]Spork complete![/bold green]")
    console.print()
    console.print("  Remotes:")
    console.print(f"    origin   → {FORGE_URL}/{OWNER}/{repo_name}")
    console.print(f"    mirror   → {FORGE_URL}/{MIRROR_ORG}/{repo_name}")
    console.print(f"    upstream → {upstream_url}")
    console.print()
    console.print("  Branches:")
    console.print(f"    {detected_main:<12} — clean upstream tracking (never commit here)")
    console.print("    blumeops     — local infra + workflows (default branch)")
    console.print()
    console.print("  Next steps:")
    console.print("    • Create feature branches:")
    console.print(f"      git checkout -b feature/upstream/my-change {detected_main}")
    console.print("      git checkout -b feature/local/my-change blumeops")
    console.print("    • Mirror-sync runs daily at 05:00 UTC")
    console.print("    • See: docs/explanation/spork-strategy.md")


if __name__ == "__main__":
    app()