rfc-app/backend/app/api_admin.py

"""§17 admin endpoints + Slice 7 user search.

The admin's repertoire grew across the prior slices without earning a
centralized home: role grants (§6.1), the §6.2 app-wide write-mute, the
§6.5 / §5 audit logs, and the §13 graduation-readiness queue. Slice 7
consolidates them behind `/api/admin/*` so the chrome can hold them in
one tabbed surface (`Admin.jsx`).

The endpoints in this module:

  - `GET  /api/admin/users`                       — list users with role + mute
  - `POST /api/admin/users/<id>/role`             — set role per §6.1
  - `POST /api/admin/users/<id>/mute`             — set the §6.2 write-mute
  - `POST /api/admin/users`                       — v0.17.0: create user + invite
  - `GET  /api/admin/users/invites`               — v0.17.0: pending invites
  - `GET  /api/admin/audit`                       — paged `actions` log
  - `GET  /api/admin/permission-events`           — paged `permission_events` log
  - `GET  /api/admin/graduation-queue`            — super-drafts ready to graduate
  - `GET  /api/users/search?q=…`                  — typeahead for §15.8 mute add

Permission gates: every `/api/admin/*` endpoint requires
`require_admin` (owner or admin); the role-change endpoint additionally
refuses any non-owner caller granting `owner`, since owner-zero is the
only owner bootstrap path per §6.1 and a downgrade from owner needs the
sitting owner's hand. The user-search endpoint is open to any
authenticated viewer — it powers the §15.8 mute typeahead and contains
no privileged information beyond what every authenticated viewer
already sees in the catalog and chat-author surfaces.
"""
from __future__ import annotations

import json
from typing import Any

from fastapi import APIRouter, HTTPException, Query, Request
from pydantic import BaseModel, Field

from . import auth, db, email_invite, invites
from .config import Config
from .email import EmailConfig


# ---------------------------------------------------------------------------
# Pydantic bodies
# ---------------------------------------------------------------------------


class RoleBody(BaseModel):
    role: str = Field(pattern="^(owner|admin|contributor)$")


class MuteBody(BaseModel):
    muted: bool


class PermissionStateBody(BaseModel):
    # v0.9.0: the admin flip from the user-management page (roadmap
    # item #7). `pending` is not surfaceable from the admin UI —
    # only the OTC verify path lands a row in `pending` — but we
    # accept it in the pattern in case a future restore-to-queue
    # gesture wants to re-pend a granted user; today the UI only
    # exposes `granted` and `revoked`.
    state: str = Field(pattern="^(pending|granted|revoked)$")


class AllowlistAddBody(BaseModel):
    email: str = Field(min_length=3, max_length=320)
    note: str | None = Field(default=None, max_length=200)


class CreateUserInviteBody(BaseModel):
    """v0.17.0 / roadmap item #16 — admin-create user + invite email.

    The admin types these fields on the "Create user + invite" modal on
    `/admin/users`. The email + role are required; first/last name and
    the optional custom message round out the body.

    Bounds mirror the rest of the codebase:
      * `email`: 320 chars — RFC 5321 envelope limit, same as
        `OtcRequestBody` / `BetaRequestBody` / `AllowlistAddBody`.
      * `first_name` / `last_name`: 120 chars — same as the v0.8.0
        `BetaRequestBody` capture form.
      * `role`: pydantic regex pinned to the §6.1 set so an unknown
        role fails at the body bound (422) instead of landing as a
        CHECK constraint violation in the migration.
      * `custom_message`: 500 chars — the brief calls this out as
        the max. The frontend modal shows a "remaining chars"
        counter to match.
    """
    email: str = Field(min_length=3, max_length=320)
    first_name: str = Field(default="", max_length=120)
    last_name: str = Field(default="", max_length=120)
    role: str = Field(pattern="^(owner|admin|contributor)$")
    custom_message: str = Field(default="", max_length=500)


# ---------------------------------------------------------------------------
# Router
# ---------------------------------------------------------------------------


def make_router(config: Config) -> APIRouter:
    del config  # unused for now; reserved for future per-deployment knobs
    router = APIRouter()

    # ----- User listing -----

    @router.get("/api/admin/users")
    async def list_users(request: Request) -> dict[str, Any]:
        """v0.9.0: the user-management surface (roadmap item #7).

        The listing carries every column the admin queue needs to triage
        pending beta-access requests alongside the existing role/mute
        affordances. Sort order surfaces pending requests first (so the
        admin lands on the inbox shape), then granted, then revoked;
        within a state, ownership/role and recency are the tiebreakers
        so the legacy ordering (owner first, then admin, then by name)
        is preserved inside the granted bucket.

        `permission_decided_by_login` joins the deciding admin row so
        the UI can render "granted by @ben" without a second round-trip.

        v0.16.0 (roadmap item #12) additive: each user row now carries
        an `rfc_invitations` array — the per-RFC invitations the user
        has accepted. This is the "permission-grant requests from
        invited users" hook the roadmap text calls for: when a user
        accepts a per-RFC invite and they're not yet platform-granted,
        the admin sees "here because @ben invited them to <RFC> as
        <role>" alongside their pending row, informing (not deciding)
        the platform grant. The two write surfaces remain distinct —
        the RFC's owner controls per-RFC roles; the admin controls
        platform-grant state.
        """
        auth.require_admin(request)
        rows = db.conn().execute(
            """
            SELECT u.id, u.gitea_login, u.display_name, u.email, u.role, u.muted,
                   u.created_at, u.last_seen_at,
                   u.permission_state, u.first_name, u.last_name,
                   u.beta_request_reason,
                   u.permission_decided_by, u.permission_decided_at,
                   d.gitea_login AS decided_by_login,
                   d.display_name AS decided_by_display
              FROM users u
         LEFT JOIN users d ON d.id = u.permission_decided_by
             ORDER BY
                CASE u.permission_state
                  WHEN 'pending' THEN 0
                  WHEN 'granted' THEN 1
                  WHEN 'revoked' THEN 2
                  ELSE 3
                END,
                u.role = 'owner' DESC, u.role = 'admin' DESC,
                COALESCE(u.last_seen_at, u.created_at) DESC,
                u.display_name COLLATE NOCASE
            """
        ).fetchall()
        # v0.16.0 — per-user accepted per-RFC invitations. One query
        # over the full set, indexed bucket-by-user-id in Python so
        # the per-row attachment below is O(1). Empty array for users
        # who hold no accepted invitations.
        invitation_rows = db.conn().execute(
            """
            SELECT c.user_id, c.rfc_slug, c.role_in_rfc, c.created_at,
                   r.title AS rfc_title,
                   i.id AS invitation_id, i.invitee_email,
                   ui.gitea_login AS inviter_login,
                   ui.display_name AS inviter_display
              FROM rfc_collaborators c
         LEFT JOIN cached_rfcs r ON r.slug = c.rfc_slug
         LEFT JOIN rfc_invitations i ON i.id = c.invitation_id
         LEFT JOIN users ui ON ui.id = i.inviter_user_id
             ORDER BY c.created_at DESC
            """
        ).fetchall()
        per_user_invites: dict[int, list[dict]] = {}
        for ir in invitation_rows:
            per_user_invites.setdefault(ir["user_id"], []).append({
                "rfc_slug": ir["rfc_slug"],
                "rfc_title": ir["rfc_title"] or ir["rfc_slug"],
                "role_in_rfc": ir["role_in_rfc"],
                "invited_at": ir["created_at"],
                "invitation_id": ir["invitation_id"],
                "invitee_email": ir["invitee_email"],
                "inviter_login": ir["inviter_login"],
                "inviter_display": ir["inviter_display"],
            })
        # v0.17.0 / roadmap item #16: a user row whose `last_seen_at`
        # is NULL is one of two things — a brand-new row that was just
        # provisioned (rare, and the v0.7.0 OTC verify path stamps
        # last_seen_at on the same call that creates the row), or an
        # admin-created invite-pending row (v0.17.0 — created by
        # `POST /api/admin/users`). We surface a `pending_invite_id`
        # field by joining through `user_invite_tokens` so the
        # Users tab can render a "(pending invite)" badge alongside
        # the role/state controls. Filters to invites that are
        # neither expired nor claimed — once the invitee clicks
        # through, the badge clears (and `last_seen_at` populates).
        pending_invite_rows = db.conn().execute(
            """
            SELECT invited_user_id, id AS invite_id, expires_at
              FROM user_invite_tokens
             WHERE claimed_at IS NULL
               AND datetime(expires_at) > datetime('now')
            """
        ).fetchall()
        pending_invites = {
            r["invited_user_id"]: {
                "invite_id": r["invite_id"],
                "expires_at": r["expires_at"],
            }
            for r in pending_invite_rows
        }
        return {
            "items": [
                {
                    "id": r["id"],
                    "gitea_login": r["gitea_login"],
                    "display_name": r["display_name"],
                    "email": r["email"] or "",
                    "role": r["role"],
                    "muted": bool(r["muted"]),
                    "created_at": r["created_at"],
                    "last_seen_at": r["last_seen_at"],
                    "permission_state": r["permission_state"] or "granted",
                    "first_name": r["first_name"] or "",
                    "last_name": r["last_name"] or "",
                    "beta_request_reason": r["beta_request_reason"] or "",
                    "permission_decided_at": r["permission_decided_at"],
                    "permission_decided_by_login": r["decided_by_login"],
                    "permission_decided_by_display": r["decided_by_display"],
                    # v0.16.0 additive — never null, always an array.
                    "rfc_invitations": per_user_invites.get(r["id"], []),
                    # v0.17.0: present iff the row is invited-but-not-
                    # claimed-yet. The frontend renders a "(pending
                    # invite)" badge when this is non-null.
                    "pending_invite": pending_invites.get(r["id"]),
                }
                for r in rows
            ]
        }

    # ----- Create user + invite (v0.17.0 / roadmap item #16) -----

    @router.post("/api/admin/users")
    async def create_user_with_invite(
        body: CreateUserInviteBody, request: Request,
    ) -> dict[str, Any]:
        """Provision a fresh `users` row with a pre-assigned role + send
        an invite email carrying a claim link.

        Refusals:
          * `422` — the admin tries to invite their own email (no
            self-invite; symmetric to `set_permission`'s self-flip
            refusal and `set_role`'s self-downgrade refusal). Use
            the existing role-change channel for self-edits.
          * `422` — the admin tries to grant `owner` without being
            owner themselves. §6.1: owner-zero is the only owner
            bootstrap path; new owners come from a sitting owner's
            hand. A 422 here matches the message shape; a 403 would
            also be defensible, but staying with 422 keeps the
            "your input is bad" framing.
          * `409` — the email already maps to a `users` row. The
            admin should use the existing role / grant gestures on
            the existing user, not create a duplicate.
          * `422` — pydantic-level: malformed email, role outside
            the §6.1 set, custom_message over 500 chars.

        On success:
          1. The invitee `users` row lands with the chosen role and
             `permission_state='granted'` (admin's hand is the grant)
             and `last_seen_at IS NULL` (the "(pending invite)"
             discriminator the listing surface joins through).
          2. The `user_invite_tokens` row lands with the bcrypt-
             hashed opaque token; the raw token rides only in the
             email link.
          3. The invite email dispatches with subject "You're
             invited to <app> by <admin>" and the custom message
             embedded in a clearly-delimited block if present.
          4. A `permission_events` row records the admin-create
             gesture so the §6.5 / `permissions` admin tab carries
             the audit trail alongside the existing grant/revoke
             flips.
        """
        viewer = auth.require_admin(request)
        email_clean = body.email.strip().lower()
        if "@" not in email_clean or len(email_clean.split("@")[-1]) < 2:
            raise HTTPException(422, "Email looks malformed")

        # Self-invite refusal. Compare the admin's own email
        # case-insensitively against the invite target.
        viewer_row = db.conn().execute(
            "SELECT email FROM users WHERE id = ?", (viewer.user_id,)
        ).fetchone()
        viewer_email = (viewer_row["email"] or "").strip().lower() if viewer_row else ""
        if viewer_email and viewer_email == email_clean:
            raise HTTPException(
                422,
                "You cannot invite yourself — use the role-change channel "
                "if you need to edit your own row",
            )

        # Owner-grant refusal: §6.1 says only a sitting owner can mint
        # a new owner. An admin trying to invite-as-owner is refused
        # at 422; the admin should ask the owner to issue the invite,
        # or invite as `admin` and let the owner promote later.
        if body.role == "owner" and viewer.role != "owner":
            raise HTTPException(
                422,
                "Only an owner can invite a new owner — invite as admin and "
                "ask the owner to promote, or have the owner issue this invite",
            )

        # Duplicate-email refusal. A pre-existing row (regardless of
        # permission_state) means the admin should use the existing
        # role / grant gestures, not create a parallel user.
        existing = db.conn().execute(
            "SELECT id FROM users WHERE email = ? COLLATE NOCASE LIMIT 1",
            (email_clean,),
        ).fetchone()
        if existing is not None:
            raise HTTPException(409, "A user with this email already exists")

        # Create the invitee row + token row + send the email.
        outcome = invites.create_invite(
            email=email_clean,
            first_name=body.first_name,
            last_name=body.last_name,
            role=body.role,
            custom_message=body.custom_message,
            created_by_admin_id=viewer.user_id,
        )

        # Audit row in permission_events so the admin Permissions tab
        # carries the gesture. The before-state is "n/a" (the row
        # did not exist); the after-state is the granted role. We
        # use a new `event_kind='user_invited'` so the existing
        # grant/revoke kinds stay scoped to their flip surface.
        db.conn().execute(
            """
            INSERT INTO permission_events
                (actor_user_id, subject_user_id, event_kind, details)
            VALUES (?, ?, 'user_invited', ?)
            """,
            (
                viewer.user_id,
                outcome.invited_user_id,
                json.dumps({
                    "email": email_clean,
                    "role": body.role,
                    "invite_id": outcome.invite_id,
                    "custom_message_chars": len(body.custom_message or ""),
                }),
            ),
        )

        # Build the claim URL using the same APP_URL the email module
        # reads. The token rides as a query-string param to the
        # frontend route `/invites/claim?token=…`; the frontend POSTs
        # it back to `/api/invites/claim` which consumes the row.
        cfg = EmailConfig.from_env()
        from urllib.parse import urlencode
        claim_url = f"{cfg.app_url}/invites/claim?{urlencode({'token': outcome.raw_token})}"

        # Fetch the inviter display so the email body can render
        # "Ben Stull (ben@example.com) has invited you to …". We
        # read off the row fresh rather than trusting the session
        # cookie's cached display_name.
        inviter_row = db.conn().execute(
            "SELECT display_name, email FROM users WHERE id = ?",
            (viewer.user_id,),
        ).fetchone()
        inviter_display = (
            (inviter_row["display_name"] if inviter_row else "") or viewer.display_name or "An admin"
        )
        inviter_email_for_body = (inviter_row["email"] if inviter_row else "") or viewer.email or ""

        email_invite.send_invite_email(
            to_address=email_clean,
            claim_url=claim_url,
            inviter_display=inviter_display,
            inviter_email=inviter_email_for_body,
            custom_message=body.custom_message,
        )

        return {
            "ok": True,
            "invite_id": outcome.invite_id,
            "invited_user_id": outcome.invited_user_id,
            "email": email_clean,
            "role": body.role,
        }

    @router.get("/api/admin/users/invites")
    async def list_user_invites(request: Request) -> dict[str, Any]:
        """List active (not claimed, not expired) admin-issued invites.

        Powers the admin's "I sent these but they haven't been claimed
        yet" view. The frontend uses this alongside `list_users` —
        the user-listing's `pending_invite` field carries the per-row
        flag; this endpoint carries the full invite shape for a
        dedicated drill-in surface.
        """
        auth.require_admin(request)
        rows = invites.list_pending_invites()
        # Join through to the admin display names so the surface can
        # render "invited by @ben" without a second client call.
        admin_ids = {r.created_by_admin_id for r in rows}
        admin_lookup: dict[int, dict[str, str]] = {}
        if admin_ids:
            placeholders = ",".join("?" * len(admin_ids))
            admin_rows = db.conn().execute(
                f"SELECT id, gitea_login, display_name FROM users "
                f"WHERE id IN ({placeholders})",
                tuple(admin_ids),
            ).fetchall()
            admin_lookup = {
                ar["id"]: {
                    "gitea_login": ar["gitea_login"] or "",
                    "display_name": ar["display_name"] or "",
                }
                for ar in admin_rows
            }
        return {
            "items": [
                {
                    "id": r.id,
                    "email": r.email,
                    "role": r.role,
                    "first_name": r.first_name,
                    "last_name": r.last_name,
                    "custom_message": r.custom_message,
                    "created_at": r.created_at,
                    "expires_at": r.expires_at,
                    "invited_user_id": r.invited_user_id,
                    "created_by_admin_id": r.created_by_admin_id,
                    "created_by_login": admin_lookup.get(
                        r.created_by_admin_id, {}
                    ).get("gitea_login", ""),
                    "created_by_display": admin_lookup.get(
                        r.created_by_admin_id, {}
                    ).get("display_name", ""),
                }
                for r in rows
            ]
        }

    # ----- Role change (§6.1) -----

    @router.post("/api/admin/users/{user_id}/role")
    async def set_role(user_id: int, body: RoleBody, request: Request) -> dict[str, Any]:
        viewer = auth.require_admin(request)
        target = db.conn().execute(
            "SELECT id, role, gitea_login FROM users WHERE id = ?", (user_id,)
        ).fetchone()
        if target is None:
            raise HTTPException(404, "User not found")
        # §6.1: only an owner can grant `owner`, and only an owner can
        # revoke an owner. Admins can flip contributor ↔ admin freely.
        if body.role == "owner" and viewer.role != "owner":
            raise HTTPException(403, "Only an owner can grant the owner role")
        if target["role"] == "owner" and viewer.role != "owner":
            raise HTTPException(403, "Only an owner can change an owner's role")
        # An owner cannot demote themselves — that would orphan owner-zero
        # if they are the only owner, and the role-change UI should not
        # smuggle a "downgrade self" path through this endpoint.
        if target["id"] == viewer.user_id and body.role != viewer.role:
            raise HTTPException(403, "Use the explicit succession path to change your own role")

        before = target["role"]
        if before == body.role:
            return {"ok": True, "role": body.role, "changed": False}

        db.conn().execute(
            "UPDATE users SET role = ? WHERE id = ?", (body.role, user_id),
        )
        db.conn().execute(
            """
            INSERT INTO permission_events
                (actor_user_id, subject_user_id, event_kind, details)
            VALUES (?, ?, 'role_changed', ?)
            """,
            (
                viewer.user_id,
                user_id,
                json.dumps({"before": before, "after": body.role}),
            ),
        )
        return {"ok": True, "role": body.role, "changed": True}

    # ----- Permission state (§6.1, v0.9.0 roadmap item #7) -----

    @router.post("/api/admin/users/{user_id}/permission")
    async def set_permission(user_id: int, body: PermissionStateBody, request: Request) -> dict[str, Any]:
        """Flip a user's `permission_state` between pending/granted/revoked.

        v0.8.0 wired the column shape but shipped no admin UI for it —
        the grant gesture was a manual `UPDATE users` against the DB.
        v0.9.0 (roadmap item #7) lands the admin user-management page;
        this endpoint is its single write surface.

        Audit shape: every flip writes a `permission_events` row with
        event_kind in {'permission_granted', 'permission_revoked',
        'permission_repended'} so §6.5's log carries the change. The
        `permission_decided_by` / `permission_decided_at` columns on
        the user row are co-stamped so the user listing can render
        "granted by @ben at <date>" without a second join through
        the audit table.

        Refuses with 422 if the admin tries to flip their own row
        (no self-grant / self-revoke; symmetric to set_mute's
        self-mute refusal and set_role's self-downgrade refusal).
        """
        viewer = auth.require_admin(request)
        target = db.conn().execute(
            "SELECT id, role, permission_state FROM users WHERE id = ?",
            (user_id,),
        ).fetchone()
        if target is None:
            raise HTTPException(404, "User not found")
        if target["id"] == viewer.user_id:
            raise HTTPException(422, "You cannot change your own permission state")

        before = target["permission_state"] or "granted"
        after = body.state
        if before == after:
            return {"ok": True, "permission_state": after, "changed": False}

        db.conn().execute(
            """
            UPDATE users
               SET permission_state = ?,
                   permission_decided_by = ?,
                   permission_decided_at = datetime('now')
             WHERE id = ?
            """,
            (after, viewer.user_id, user_id),
        )
        event_kind = {
            "granted": "permission_granted",
            "revoked": "permission_revoked",
            "pending": "permission_repended",
        }[after]
        db.conn().execute(
            """
            INSERT INTO permission_events
                (actor_user_id, subject_user_id, event_kind, details)
            VALUES (?, ?, ?, ?)
            """,
            (
                viewer.user_id,
                user_id,
                event_kind,
                json.dumps({"before": before, "after": after}),
            ),
        )
        return {"ok": True, "permission_state": after, "changed": True}

    # ----- Write-mute (§6.2) -----

    @router.post("/api/admin/users/{user_id}/mute")
    async def set_mute(user_id: int, body: MuteBody, request: Request) -> dict[str, Any]:
        viewer = auth.require_admin(request)
        target = db.conn().execute(
            "SELECT id, role, muted FROM users WHERE id = ?", (user_id,)
        ).fetchone()
        if target is None:
            raise HTTPException(404, "User not found")
        if target["id"] == viewer.user_id:
            raise HTTPException(422, "You cannot write-mute yourself")
        # §6.2 + §6.1: admins/owners are not write-mutable. The write-mute
        # is a contributor-only refusal — an admin's authority is the
        # role-change channel, not a mute.
        if target["role"] in ("owner", "admin"):
            raise HTTPException(
                403,
                "Owners and admins cannot be write-muted — change the role instead",
            )

        before = bool(target["muted"])
        after = bool(body.muted)
        if before == after:
            return {"ok": True, "muted": after, "changed": False}

        db.conn().execute(
            "UPDATE users SET muted = ? WHERE id = ?", (1 if after else 0, user_id),
        )
        db.conn().execute(
            """
            INSERT INTO permission_events
                (actor_user_id, subject_user_id, event_kind, details)
            VALUES (?, ?, ?, ?)
            """,
            (
                viewer.user_id,
                user_id,
                "muted" if after else "restored",
                json.dumps({"before": before, "after": after}),
            ),
        )
        return {"ok": True, "muted": after, "changed": True}

    # ----- Audit log (`actions` + `permission_events`) -----

    @router.get("/api/admin/audit")
    async def list_audit(
        request: Request,
        action_kind: str | None = None,
        actor_user_id: int | None = None,
        rfc_slug: str | None = None,
        limit: int = Query(default=100, ge=1, le=500),
        before_id: int | None = None,
    ) -> dict[str, Any]:
        auth.require_admin(request)
        clauses: list[str] = []
        args: list[Any] = []
        if action_kind:
            clauses.append("a.action_kind = ?")
            args.append(action_kind)
        if actor_user_id is not None:
            clauses.append("a.actor_user_id = ?")
            args.append(actor_user_id)
        if rfc_slug:
            clauses.append("a.rfc_slug = ?")
            args.append(rfc_slug)
        if before_id is not None:
            clauses.append("a.id < ?")
            args.append(before_id)
        where = ("WHERE " + " AND ".join(clauses)) if clauses else ""
        rows = db.conn().execute(
            f"""
            SELECT a.id, a.actor_user_id, a.on_behalf_of, a.action_kind,
                   a.rfc_slug, a.branch_name, a.pr_number, a.bot_commit_sha,
                   a.details, a.created_at,
                   u.gitea_login AS actor_login, u.display_name AS actor_display
              FROM actions a
         LEFT JOIN users u ON u.id = a.actor_user_id
             {where}
             ORDER BY a.id DESC
             LIMIT ?
            """,
            (*args, limit),
        ).fetchall()
        # The distinct-action-kinds list powers the filter chip in
        # `Admin.jsx`; cheap to compute alongside the page since the
        # action_kind set is bounded.
        kinds = [
            r["action_kind"]
            for r in db.conn().execute(
                "SELECT DISTINCT action_kind FROM actions ORDER BY action_kind"
            )
        ]
        return {
            "items": [
                {
                    "id": r["id"],
                    "action_kind": r["action_kind"],
                    "actor_user_id": r["actor_user_id"],
                    "actor_login": r["actor_login"],
                    "actor_display": r["actor_display"],
                    "on_behalf_of": r["on_behalf_of"],
                    "rfc_slug": r["rfc_slug"],
                    "branch_name": r["branch_name"],
                    "pr_number": r["pr_number"],
                    "bot_commit_sha": r["bot_commit_sha"],
                    "details": _safe_json(r["details"]),
                    "created_at": r["created_at"],
                }
                for r in rows
            ],
            "action_kinds": kinds,
            "has_more": len(rows) == limit,
        }

    @router.get("/api/admin/outbound-emails")
    async def list_outbound_emails(
        request: Request,
        kind: str | None = None,
        status: str | None = None,
        to_address: str | None = None,
        limit: int = Query(default=100, ge=1, le=500),
        before_id: int | None = None,
    ) -> dict[str, Any]:
        """v0.18.0 Slice 4: read-only inspection of the
        `outbound_emails` audit table.

        Answers questions like "did this person ever get their
        invite?" without grepping VM logs. Filterable by kind
        ('otc' | 'invite' | 'notification' | 'bundle' | 'digest'),
        status ('sent' | 'failed' | 'deferred' | 'bounced'), and
        to_address; the latter is exact-match because the audit
        question is usually "the specific person who said they
        didn't receive it." Per the proposal, no admin UI ships
        with v0.18.0 — operator queries via curl + jq for now.
        """
        auth.require_admin(request)
        clauses: list[str] = []
        args: list[Any] = []
        if kind:
            clauses.append("kind = ?")
            args.append(kind)
        if status:
            clauses.append("status = ?")
            args.append(status)
        if to_address:
            clauses.append("LOWER(to_address) = LOWER(?)")
            args.append(to_address)
        if before_id is not None:
            clauses.append("id < ?")
            args.append(before_id)
        where = ("WHERE " + " AND ".join(clauses)) if clauses else ""
        rows = db.conn().execute(
            f"""
            SELECT id, to_address, from_address, subject, kind, sent_at,
                   status, error, notification_id, message_id
              FROM outbound_emails
             {where}
             ORDER BY id DESC
             LIMIT ?
            """,
            (*args, limit),
        ).fetchall()
        return {
            "items": [
                {
                    "id": r["id"],
                    "to_address": r["to_address"],
                    "from_address": r["from_address"],
                    "subject": r["subject"],
                    "kind": r["kind"],
                    "sent_at": r["sent_at"],
                    "status": r["status"],
                    "error": r["error"],
                    "notification_id": r["notification_id"],
                    "message_id": r["message_id"],
                }
                for r in rows
            ],
            "has_more": len(rows) == limit,
        }

    @router.get("/api/admin/permission-events")
    async def list_permission_events(
        request: Request,
        limit: int = Query(default=100, ge=1, le=500),
        before_id: int | None = None,
    ) -> dict[str, Any]:
        auth.require_admin(request)
        clauses: list[str] = []
        args: list[Any] = []
        if before_id is not None:
            clauses.append("p.id < ?")
            args.append(before_id)
        where = ("WHERE " + " AND ".join(clauses)) if clauses else ""
        rows = db.conn().execute(
            f"""
            SELECT p.id, p.event_kind, p.details, p.created_at,
                   p.actor_user_id, p.subject_user_id,
                   au.gitea_login AS actor_login, au.display_name AS actor_display,
                   su.gitea_login AS subject_login, su.display_name AS subject_display
              FROM permission_events p
         LEFT JOIN users au ON au.id = p.actor_user_id
         LEFT JOIN users su ON su.id = p.subject_user_id
             {where}
             ORDER BY p.id DESC
             LIMIT ?
            """,
            (*args, limit),
        ).fetchall()
        return {
            "items": [
                {
                    "id": r["id"],
                    "event_kind": r["event_kind"],
                    "actor_login": r["actor_login"],
                    "actor_display": r["actor_display"],
                    "subject_login": r["subject_login"],
                    "subject_display": r["subject_display"],
                    "details": _safe_json(r["details"]),
                    "created_at": r["created_at"],
                }
                for r in rows
            ],
            "has_more": len(rows) == limit,
        }

    # ----- Graduation-readiness queue (§13.2) -----

    @router.get("/api/admin/graduation-queue")
    async def graduation_queue(request: Request) -> dict[str, Any]:
        auth.require_admin(request)
        # §13 / §13.2: a super-draft is ready when (a) it has at least one
        # owner claimed via §13.1 and (b) it has zero open meta_body_edit
        # PRs. We compute both with one pass, returning the ready set and
        # the not-yet-ready set so the admin sees what is gating each.
        rows = db.conn().execute(
            """
            SELECT slug, title, owners_json, arbiters_json, tags_json,
                   proposed_at, last_entry_commit_at
              FROM cached_rfcs
             WHERE state = 'super-draft'
             ORDER BY COALESCE(last_entry_commit_at, proposed_at) DESC
            """
        ).fetchall()
        items_ready: list[dict[str, Any]] = []
        items_blocked: list[dict[str, Any]] = []
        for r in rows:
            owners = json.loads(r["owners_json"] or "[]")
            blocking = db.conn().execute(
                """
                SELECT COUNT(*) AS n FROM cached_prs
                 WHERE rfc_slug = ? AND state = 'open' AND pr_kind = 'meta_body_edit'
                """,
                (r["slug"],),
            ).fetchone()["n"]
            payload = {
                "slug": r["slug"],
                "title": r["title"],
                "owners": owners,
                "tags": json.loads(r["tags_json"] or "[]"),
                "proposed_at": r["proposed_at"],
                "last_entry_commit_at": r["last_entry_commit_at"],
                "blocking_prs": blocking,
                "owners_set": len(owners) > 0,
            }
            if payload["owners_set"] and blocking == 0:
                items_ready.append(payload)
            else:
                items_blocked.append(payload)
        return {"ready": items_ready, "blocked": items_blocked}

    # ----- §13.7: retired (soft-deleted) entries — site owners only -----

    @router.get("/api/admin/retired-rfcs")
    async def retired_rfcs(request: Request) -> dict[str, Any]:
        # Retired entries are hidden from every browsing surface (§13.7);
        # this owner-gated list is how a site owner discovers them to
        # un-retire. Admins do not get this surface — un-retire authority is
        # site-owner-only, so neither is the list that feeds it.
        viewer = auth.require_admin(request)
        if viewer.role != "owner":
            raise HTTPException(403, "Site owner role required")
        rows = db.conn().execute(
            """
            SELECT slug, title, rfc_id, owners_json, tags_json,
                   proposed_at, updated_at
              FROM cached_rfcs
             WHERE state = 'retired'
             ORDER BY updated_at DESC
            """
        ).fetchall()
        items = []
        for r in rows:
            # The state it would return to on un-retire, mirroring
            # api_graduation._prior_state_before_retire's audit lookup.
            prior = db.conn().execute(
                """
                SELECT details FROM actions
                 WHERE rfc_slug = ? AND action_kind = 'retire'
                 ORDER BY id DESC LIMIT 1
                """,
                (r["slug"],),
            ).fetchone()
            restored_state = None
            if prior and prior["details"]:
                try:
                    restored_state = json.loads(prior["details"]).get("prior_state")
                except (ValueError, TypeError):
                    restored_state = None
            if restored_state not in ("super-draft", "active"):
                restored_state = "active" if r["rfc_id"] else "super-draft"
            items.append({
                "slug": r["slug"],
                "title": r["title"],
                "id": r["rfc_id"],
                "owners": json.loads(r["owners_json"] or "[]"),
                "tags": json.loads(r["tags_json"] or "[]"),
                "retired_at": r["updated_at"],
                "restores_to": restored_state,
            })
        return {"items": items}

    # ----- User search (typeahead for §15.8 mute add) -----

    @router.get("/api/users/search")
    async def search_users(
        request: Request, q: str = Query(default="", min_length=0, max_length=80),
    ) -> dict[str, Any]:
        viewer = auth.require_user(request)
        needle = q.strip().lower()
        # An empty query returns recent users; a non-empty query matches
        # against gitea_login and display_name with a prefix-first
        # ranking so the typeahead surfaces obvious matches early.
        if not needle:
            rows = db.conn().execute(
                """
                SELECT id, gitea_login, display_name, role
                  FROM users WHERE id != ?
                 ORDER BY last_seen_at DESC LIMIT 10
                """,
                (viewer.user_id,),
            ).fetchall()
        else:
            like = f"%{needle}%"
            prefix = f"{needle}%"
            rows = db.conn().execute(
                """
                SELECT id, gitea_login, display_name, role
                  FROM users
                 WHERE id != ?
                   AND (LOWER(gitea_login) LIKE ? OR LOWER(display_name) LIKE ?)
                 ORDER BY
                    CASE WHEN LOWER(gitea_login) LIKE ? THEN 0 ELSE 1 END,
                    LOWER(gitea_login)
                 LIMIT 10
                """,
                (viewer.user_id, like, like, prefix),
            ).fetchall()
        return {
            "items": [
                {
                    "id": r["id"],
                    "gitea_login": r["gitea_login"],
                    "display_name": r["display_name"],
                    "role": r["role"],
                }
                for r in rows
            ]
        }

    # ----- Private-beta allowlist (`migrations/011_allowlist.sql`) -----

    @router.get("/api/admin/allowlist")
    async def list_allowlist(request: Request) -> dict[str, Any]:
        auth.require_admin(request)
        rows = db.conn().execute(
            """
            SELECT a.email, a.note, a.created_at,
                   u.gitea_login AS added_by_login,
                   u.display_name AS added_by_display
              FROM allowed_emails a
         LEFT JOIN users u ON u.id = a.added_by_user_id
             ORDER BY a.created_at DESC
            """
        ).fetchall()
        return {
            "active": len(rows) > 0,
            "items": [
                {
                    "email": r["email"],
                    "note": r["note"] or "",
                    "added_by_login": r["added_by_login"],
                    "added_by_display": r["added_by_display"],
                    "created_at": r["created_at"],
                }
                for r in rows
            ],
        }

    @router.post("/api/admin/allowlist")
    async def add_allowlist(body: AllowlistAddBody, request: Request) -> dict[str, Any]:
        viewer = auth.require_admin(request)
        email = body.email.strip()
        if "@" not in email or len(email.split("@")[-1]) < 2:
            raise HTTPException(422, "Email looks malformed")
        existing = db.conn().execute(
            "SELECT 1 FROM allowed_emails WHERE email = ? LIMIT 1", (email,)
        ).fetchone()
        if existing is not None:
            raise HTTPException(409, "Email already on the allowlist")
        db.conn().execute(
            """
            INSERT INTO allowed_emails (email, added_by_user_id, note)
            VALUES (?, ?, ?)
            """,
            (email, viewer.user_id, body.note),
        )
        # Audit trail: when the email already maps to a known user, emit a
        # permission_events row so §6.5's log stays the single place to
        # look for "who let this person in." For brand-new emails the
        # allowed_emails row itself carries (added_by_user_id, created_at)
        # which is sufficient until the user actually signs in.
        subject = db.conn().execute(
            "SELECT id FROM users WHERE email = ? COLLATE NOCASE LIMIT 1", (email,)
        ).fetchone()
        if subject is not None:
            db.conn().execute(
                """
                INSERT INTO permission_events
                    (actor_user_id, subject_user_id, event_kind, details)
                VALUES (?, ?, 'allowlist_added', ?)
                """,
                (
                    viewer.user_id,
                    subject["id"],
                    json.dumps({"email": email, "note": body.note or ""}),
                ),
            )
        return {"ok": True, "email": email}

    @router.delete("/api/admin/allowlist/{email}")
    async def remove_allowlist(email: str, request: Request) -> dict[str, Any]:
        viewer = auth.require_admin(request)
        existing = db.conn().execute(
            "SELECT 1 FROM allowed_emails WHERE email = ? LIMIT 1", (email,)
        ).fetchone()
        if existing is None:
            raise HTTPException(404, "Email not on the allowlist")
        db.conn().execute("DELETE FROM allowed_emails WHERE email = ?", (email,))
        subject = db.conn().execute(
            "SELECT id FROM users WHERE email = ? COLLATE NOCASE LIMIT 1", (email,)
        ).fetchone()
        if subject is not None:
            db.conn().execute(
                """
                INSERT INTO permission_events
                    (actor_user_id, subject_user_id, event_kind, details)
                VALUES (?, ?, 'allowlist_removed', ?)
                """,
                (
                    viewer.user_id,
                    subject["id"],
                    json.dumps({"email": email}),
                ),
            )
        return {"ok": True, "email": email}

    return router


def _safe_json(blob: str | None) -> Any:
    if not blob:
        return None
    try:
        return json.loads(blob)
    except Exception:
        return blob