rfc-app/scripts/seed_test_rfc.py

#!/usr/bin/env python3
"""Seed a per-RFC repo for manual testing of the §8 active-RFC view.

Slice 2 ships the active-RFC surface against a per-RFC repo at
`<org>/rfc-NNNN-<slug>`. Slice 5 owns the §13 graduation flow that
actually creates such a repo from a super-draft. Until Slice 5 lands,
this script is the dev shortcut: it creates the per-RFC repo, seeds
`RFC.md` on `main`, and graduates the meta-repo entry to `state: active`
with the integer ID and repo URL filled in — so a browser session can
exercise the §8 surface end-to-end.

Per §1 the per-RFC repo's content commits go through the bot wrapper.
The meta-entry frontmatter update is a one-shot bootstrap write — same
pattern `seed_meta_repo.py` uses for the initial meta-repo seed.

Usage:
    cd backend && .venv/bin/python ../scripts/seed_test_rfc.py \\
        --slug open-human-model \\
        --title "Open Human Model" \\
        [--id 0001] \\
        [--body-file path/to/RFC.md] \\
        [--owner ben] \\
        [--arbiter ben]

Re-running is safe; every step is upsert-shaped. If the meta entry
doesn't exist, it's created as a super-draft and graduated in one
pass. If it's already active, the script reports and exits without
clobbering anything.
"""
from __future__ import annotations

import argparse
import asyncio
import base64
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "backend"))

from app import db, entry as entry_mod  # noqa: E402
from app.bot import Actor, Bot  # noqa: E402
from app.config import load_config  # noqa: E402
from app.gitea import Gitea, GiteaError  # noqa: E402


DEFAULT_RFC_MD = """# {title}

*Placeholder body for the active-RFC view.*

Edit this through the app — the §8 surface is the only intended
contribution path. This file was seeded by `scripts/seed_test_rfc.py`
as a dev shortcut for exercising Slice 2 before Slice 5's graduation
flow lands.

## Why this RFC exists

A short paragraph the contributor would normally write before
graduation, preserved here so the editor has something legible to
render on first open.
"""


def parse_args() -> argparse.Namespace:
    p = argparse.ArgumentParser(description="Seed a per-RFC repo for §8 testing")
    p.add_argument("--slug", required=True, help="Kebab-case slug, e.g. open-human-model")
    p.add_argument("--title", required=True, help='Display title, e.g. "Open Human Model"')
    p.add_argument("--id", default=None, help="Integer ID without prefix, e.g. 0001")
    p.add_argument("--body-file", default=None, help="Path to a markdown file for RFC.md")
    p.add_argument("--owner", action="append", default=None, help="Gitea login; repeatable")
    p.add_argument("--arbiter", action="append", default=None, help="Gitea login; repeatable")
    p.add_argument("--seed-actor", default=None,
                   help="Gitea login to record as On-behalf-of (default: OWNER_GITEA_LOGIN)")
    return p.parse_args()


async def main() -> None:
    args = parse_args()
    config = load_config()
    db.run_migrations(config)
    db.init(config)

    if not entry_mod.is_valid_slug(args.slug):
        sys.exit(f"invalid slug: {args.slug!r}")

    integer_id = compute_integer_id(args.id)
    rfc_id = f"RFC-{integer_id}"
    repo_name = f"rfc-{integer_id}-{args.slug}"
    repo_full = f"{config.gitea_org}/{repo_name}"

    body = read_body(args)
    owners = args.owner or [config.owner_gitea_login or "ben"]
    arbiters = args.arbiter or owners[:1]

    gitea = Gitea(config)
    bot = Bot(gitea)
    actor = _seed_actor(config, args)

    try:
        # 1) Per-RFC repo + RFC.md (via the bot wrapper).
        print(f"ensuring per-RFC repo {repo_full}")
        await bot.ensure_rfc_repo_seed(
            actor,
            owner=config.gitea_org,
            repo=repo_name,
            slug=args.slug,
            title=args.title,
            body=body,
        )

        # 2) Webhook on the per-RFC repo so live events flow per §4.1.
        if config.webhook_secret:
            url = f"{config.app_url}/api/webhooks/gitea"
            print(f"ensuring webhook on {repo_full} -> {url}")
            await gitea.ensure_webhook(
                config.gitea_org, repo_name,
                url=url, secret=config.webhook_secret,
                events=["push", "pull_request", "create", "delete", "repository"],
            )
        else:
            print("skipping webhook setup (GITEA_WEBHOOK_SECRET not set)")

        # 3) Meta-repo entry: create if missing, graduate to active.
        await graduate_meta_entry(
            config, gitea,
            slug=args.slug, title=args.title, rfc_id=rfc_id, repo_full=repo_full,
            owners=owners, arbiters=arbiters, body_for_super_draft=body,
        )
    finally:
        await gitea.close()

    print()
    print(f"✓ seeded {rfc_id} — {args.title}")
    print(f"  repo:    {repo_full}")
    print(f"  meta:    rfcs/{args.slug}.md")
    print(f"  app URL: {config.app_url}/rfc/{args.slug}")
    print()
    print("On next reconciler sweep (or webhook fire) the cache will update.")
    print("You can force a sweep by restarting the backend.")


def compute_integer_id(arg: str | None) -> str:
    """Pick an integer ID. With --id provided, use it verbatim; otherwise
    take the next free 4-digit integer over the cached_rfcs.rfc_id column.
    """
    if arg:
        n = int(arg)
        return f"{n:04d}"
    rows = db.conn().execute(
        "SELECT rfc_id FROM cached_rfcs WHERE rfc_id LIKE 'RFC-%'"
    ).fetchall()
    used = set()
    for r in rows:
        try:
            used.add(int(r["rfc_id"].split("-", 1)[1]))
        except (IndexError, ValueError):
            continue
    nxt = (max(used) + 1) if used else 1
    return f"{nxt:04d}"


def read_body(args: argparse.Namespace) -> str:
    if args.body_file:
        return Path(args.body_file).read_text()
    return DEFAULT_RFC_MD.format(title=args.title)


def _seed_actor(config, args: argparse.Namespace) -> Actor:
    """Synthesize an `Actor` for the seed write. The seed script is
    bootstrap-shaped; we record it under whichever login the operator
    designates (default: OWNER_GITEA_LOGIN) so the §6.5 audit trail
    has a real human attribution.
    """
    login = args.seed_actor or config.owner_gitea_login or "seed"
    row = db.conn().execute(
        "SELECT id, gitea_login, display_name, email FROM users WHERE gitea_login = ?",
        (login,),
    ).fetchone()
    if row:
        return Actor(
            user_id=row["id"],
            gitea_login=row["gitea_login"],
            display_name=row["display_name"],
            email=row["email"] or "",
        )
    # The seed script's writes go through the §1 bot wrapper, which
    # records an `actions` row whose actor_user_id FK references
    # `users.id`. If `<login>` hasn't signed in yet, there's no row to
    # point at — refuse instead of synthesizing one, since a fake row
    # would muddy the user table that §6 / §15.9 read from.
    sys.exit(
        f"no users row for gitea_login={login!r}. Sign in via OAuth at "
        f"{config.app_url} first (or pass --seed-actor <login> for an "
        f"already-provisioned account)."
    )


async def graduate_meta_entry(
    config, gitea: Gitea, *,
    slug: str, title: str, rfc_id: str, repo_full: str,
    owners: list[str], arbiters: list[str], body_for_super_draft: str,
) -> None:
    """Idempotent: create the meta entry as super-draft if missing,
    then flip its frontmatter to `state: active` with the integer ID
    and repo URL filled in. Active entries are left alone.
    """
    path = f"rfcs/{slug}.md"
    existing = await gitea.read_file(config.gitea_org, config.meta_repo, path, ref="main")

    if existing is None:
        print(f"creating meta entry {path}")
        new_entry = entry_mod.Entry(
            slug=slug,
            title=title,
            state="active",
            id=rfc_id,
            repo=repo_full,
            proposed_by=owners[0],
            proposed_at=entry_mod.today(),
            graduated_at=entry_mod.today(),
            graduated_by=owners[0],
            owners=owners,
            arbiters=arbiters,
            tags=[],
            body="",
        )
        text = entry_mod.serialize(new_entry)
        await _bootstrap_create(gitea, config, path, text, f"Seed: graduate {slug}")
        return

    text, sha = existing
    entry = entry_mod.parse(text)
    if entry.state == "active":
        print(f"meta entry {path} is already active — leaving alone")
        return

    print(f"graduating meta entry {path} -> active")
    entry.state = "active"
    entry.id = rfc_id
    entry.repo = repo_full
    entry.graduated_at = entry_mod.today()
    entry.graduated_by = (entry.owners or owners)[0]
    if not entry.owners:
        entry.owners = owners
    if not entry.arbiters:
        entry.arbiters = arbiters
    new_text = entry_mod.serialize(entry)
    await _bootstrap_update(gitea, config, path, new_text, sha, f"Seed: graduate {slug}")


async def _bootstrap_create(gitea: Gitea, config, path: str, content: str, message: str) -> None:
    # Same bootstrap-only direct write the meta-repo seed uses.
    await gitea.create_file(
        config.gitea_org, config.meta_repo, path,
        content=content, message=message, branch="main",
    )


async def _bootstrap_update(gitea: Gitea, config, path: str, content: str, sha: str, message: str) -> None:
    await gitea.update_file(
        config.gitea_org, config.meta_repo, path,
        content=content, sha=sha, message=message, branch="main",
    )


if __name__ == "__main__":
    asyncio.run(main())