rfc-app/backend/app/auth.py

"""Gitea OAuth and user provisioning.

OAuth identity is the basis for the app's user account per §18; the §6
authorization layer is built on top by reading from the users table. On
first sign-in we insert a row with role='contributor' (or 'owner' if the
gitea_login matches the configured OWNER_GITEA_LOGIN — bootstrapping for
owner zero per §6.1). On subsequent sign-ins we refresh the display name
and avatar from Gitea so a rename in Gitea propagates here.
"""
from __future__ import annotations

import secrets
from dataclasses import dataclass
from typing import Any

import httpx
from fastapi import HTTPException, Request

from . import collections as collections_mod
from . import db
from .bot import Actor
from .config import Config
from .projects import DEFAULT_PROJECT_ID


@dataclass
class SessionUser:
    user_id: int
    gitea_id: int
    gitea_login: str
    display_name: str
    email: str
    avatar_url: str
    role: str
    # v0.8.0 / §6.1 — admission gate. Three states: 'pending' (waiting
    # for an admin grant), 'granted' (active contributor), 'revoked'
    # (was granted, later removed). Existing rows at migration time
    # default to 'granted' so grandfathered users are unaffected; OTC
    # provisions fresh users with 'pending' (see `app/otc.py`).
    permission_state: str = "granted"

    def as_actor(self) -> Actor:
        return Actor(
            user_id=self.user_id,
            gitea_login=self.gitea_login,
            display_name=self.display_name,
            email=self.email,
        )


def authorization_url(config: Config, state: str) -> str:
    return (
        f"{config.gitea_url}/login/oauth/authorize"
        f"?client_id={config.oauth_client_id}"
        f"&redirect_uri={config.redirect_uri}"
        f"&response_type=code"
        f"&state={state}"
    )


async def exchange_code(config: Config, code: str) -> dict[str, Any]:
    async with httpx.AsyncClient(timeout=30.0) as client:
        resp = await client.post(
            f"{config.gitea_url}/login/oauth/access_token",
            json={
                "client_id": config.oauth_client_id,
                "client_secret": config.oauth_client_secret,
                "code": code,
                "grant_type": "authorization_code",
                "redirect_uri": config.redirect_uri,
            },
            headers={"Accept": "application/json"},
        )
        resp.raise_for_status()
        return resp.json()


async def fetch_user_profile(config: Config, access_token: str) -> dict[str, Any]:
    async with httpx.AsyncClient(timeout=30.0) as client:
        resp = await client.get(
            f"{config.gitea_url}/api/v1/user",
            headers={"Authorization": f"token {access_token}"},
        )
        resp.raise_for_status()
        return resp.json()


def allowlist_is_active() -> bool:
    """The private-beta gate is on iff the `allowed_emails` table has any
    rows. Empty list means "open" — any successful OAuth provisions a
    user; first row added flips the deployment into private-beta mode.
    See `migrations/011_allowlist.sql` for the reasoning.
    """
    row = db.conn().execute("SELECT 1 FROM allowed_emails LIMIT 1").fetchone()
    return row is not None


def is_allowed_sign_in(profile: dict[str, Any]) -> bool:
    """Decide whether a freshly-completed OAuth profile may sign in.

    v0.8.0 (item #6) replaces the allowlist gate with an admin-grant
    flow at the OTC `/request` surface, but the Gitea OAuth callback
    in `main.py` still consults this helper so the fallback path
    keeps the v0.3.0 admission shape during the OAuth migration
    window. The eventual removal of the OAuth callback (§19.2)
    retires this function alongside it.

    Three accept paths:
      1. The allowlist is empty (gate off).
      2. The Gitea profile's email is in `allowed_emails` (case-insensitive).
      3. A `users` row already exists for this `gitea_id` — grandfather
         per `migrations/011_allowlist.sql`.
    """
    gitea_id = profile.get("id")
    if gitea_id is not None:
        existing = db.conn().execute(
            "SELECT 1 FROM users WHERE gitea_id = ? LIMIT 1", (gitea_id,)
        ).fetchone()
        if existing is not None:
            return True
    if not allowlist_is_active():
        return True
    email = (profile.get("email") or "").strip()
    if not email:
        return False
    row = db.conn().execute(
        "SELECT 1 FROM allowed_emails WHERE email = ? LIMIT 1", (email,)
    ).fetchone()
    return row is not None


def provision_user(config: Config, profile: dict[str, Any]) -> SessionUser:
    """Insert or update the users row for this Gitea profile.

    Owner zero (§6.1) is whoever's gitea_login matches OWNER_GITEA_LOGIN.
    The owner role is granted on first sign-in and never revoked from a
    later config change — once owner, always owner until an explicit
    role transition (which lives in §6.1 and isn't part of slice 1).
    """
    gitea_id = profile["id"]
    login = profile["login"]
    display = profile.get("full_name") or login
    email = profile.get("email") or ""
    avatar = profile.get("avatar_url") or ""

    c = db.conn()
    existing = c.execute("SELECT * FROM users WHERE gitea_id = ?", (gitea_id,)).fetchone()
    if existing is None:
        role = "owner" if config.owner_gitea_login and login == config.owner_gitea_login else "contributor"
        # v0.8.0: a fresh OAuth-provisioned user is also subject to
        # the admin-grant flow. The OAuth fallback only fires for
        # users who pass `is_allowed_sign_in` (so they're already on
        # the legacy allowlist or are grandfathered by gitea_id);
        # 'granted' is the right default here since the allowlist
        # check is itself the admin gesture. A future release that
        # retires the OAuth callback (§19.2) collapses both paths
        # under the same gate.
        cur = c.execute(
            """
            INSERT INTO users (gitea_id, gitea_login, email, display_name, avatar_url, role, permission_state)
            VALUES (?, ?, ?, ?, ?, ?, 'granted')
            """,
            (gitea_id, login, email, display, avatar, role),
        )
        user_id = cur.lastrowid
        permission_state = "granted"
    else:
        user_id = existing["id"]
        role = existing["role"]
        permission_state = existing["permission_state"] or "granted"
        c.execute(
            """
            UPDATE users
               SET gitea_login = ?, email = ?, display_name = ?, avatar_url = ?, last_seen_at = datetime('now')
             WHERE id = ?
            """,
            (login, email, display, avatar, user_id),
        )

    return SessionUser(
        user_id=user_id,
        gitea_id=gitea_id,
        gitea_login=login,
        display_name=display,
        email=email,
        avatar_url=avatar,
        role=role,
        permission_state=permission_state,
    )


# ----- Session helpers -----

SESSION_USER_KEY = "user"
SESSION_STATE_KEY = "oauth_state"


def store_session(request: Request, user: SessionUser) -> None:
    request.session[SESSION_USER_KEY] = {
        "user_id": user.user_id,
        "gitea_id": user.gitea_id,
        "gitea_login": user.gitea_login,
        "display_name": user.display_name,
        "email": user.email,
        "avatar_url": user.avatar_url,
        "role": user.role,
        # v0.8.0: persist the admission state on the cookie payload so
        # the post-cookie audit doesn't second-guess the row. The DB
        # is re-read on every `current_user` call regardless (so an
        # admin grant takes effect on the next request); this field
        # is purely structural redundancy for the cookie shape.
        "permission_state": user.permission_state,
    }


def current_user(request: Request) -> SessionUser | None:
    raw = request.session.get(SESSION_USER_KEY)
    if not raw:
        return None
    # Re-read the role from the database every request so role changes
    # take effect on the next API call without forcing a logout.
    row = db.conn().execute(
        "SELECT id, gitea_id, gitea_login, email, display_name, avatar_url, role, permission_state FROM users WHERE id = ?",
        (raw["user_id"],),
    ).fetchone()
    if row is None:
        return None
    # v0.7.0: OTC-provisioned users have NULL gitea_id / gitea_login.
    # Coerce nulls to the SessionUser's typed defaults so downstream
    # code (Actor, _on_behalf_trailer) reads a stable shape regardless
    # of which sign-in path the row came from. The DB remains the
    # source of truth for "is this an OAuth-linked user" (gitea_id IS
    # NOT NULL); the in-memory SessionUser is the per-request handle.
    # v0.8.0: permission_state comes off the row directly. A NULL
    # column value (shouldn't happen under the migration's
    # NOT NULL DEFAULT, but be defensive) reads as 'granted' so the
    # gate fails open for grandfathered surfaces rather than locking
    # everyone out on a malformed row.
    return SessionUser(
        user_id=row["id"],
        gitea_id=row["gitea_id"] or 0,
        gitea_login=row["gitea_login"] or "",
        display_name=row["display_name"],
        email=row["email"] or "",
        avatar_url=row["avatar_url"] or "",
        role=row["role"],
        permission_state=row["permission_state"] or "granted",
    )


def require_user(request: Request) -> SessionUser:
    user = current_user(request)
    if user is None:
        raise HTTPException(status_code=401, detail="Not authenticated")
    return user


def require_contributor(request: Request) -> SessionUser:
    """§6.1: authenticated, not write-muted, and granted by an admin.

    v0.8.0 (item #6) widens this gate. A fresh OTC sign-in lands in
    `permission_state='pending'`; the user can read everything an
    anonymous viewer can read, but every write-shaped endpoint that
    funnels through this dependency now refuses with 403 until an
    admin grants them. The `pending` blast radius is the same as
    anonymous (item #4 / v0.6.0 already audited the anon-write
    refusal at every write site), so this widening is structurally
    a relabel — the same surfaces that already refused 401 to
    anonymous now also refuse 403 to pending.
    """
    user = require_user(request)
    row = db.conn().execute("SELECT muted FROM users WHERE id = ?", (user.user_id,)).fetchone()
    if row and row["muted"]:
        raise HTTPException(status_code=403, detail="Your account is muted")
    if user.permission_state != "granted":
        # 'pending' is the post-OTC waiting state; 'revoked' is the
        # admin-undid-the-grant state. Both refuse with the same 403
        # shape; the client distinguishes via `/api/auth/me` which
        # carries `permission_state` in the response.
        raise HTTPException(
            status_code=403,
            detail="Your beta access request is in review",
        )
    return user


def require_admin(request: Request) -> SessionUser:
    """§6.1: owner or admin."""
    user = require_user(request)
    if user.role not in ("owner", "admin"):
        raise HTTPException(status_code=403, detail="Admin or owner role required")
    return user


# ===========================================================================
# §22.6 / §22.7 — project-scoped authorization (the multi-project middle tier).
#
# A deployment hosts N projects (§22). Authorization for an action on an RFC is
# the *most-permissive union* of three tiers — the actor's deployment role
# (§6.1), their project role (§22.6), and their per-RFC authority (§6.3/§12) —
# with the §22.5 visibility gate and the §6.2 write-mute *subtractive* on top
# (§22.7). The per-RFC capability helpers below (`can_discuss_rfc`,
# `can_contribute_to_rfc`, `can_invite_to_rfc`) compose all three tiers, so the
# ~20 endpoint call sites inherit multi-project behavior unchanged.
#
# Through Slice M2 the only project is the migration-seeded `default` one (the
# N=1 case, §22.13); a project is resolved from an RFC slug via
# `cached_rfcs.project_id` (unique per slug while N=1). M3's registry mirror
# lets a deployment declare a second project; these gates already hold then.
#
# OPERATOR DECISIONS (M2):
#  * implicit-on-public — on a `public` project a granted deployment
#    `contributor` keeps the pre-multi-project write *baseline* (propose
#    freely; an owned RFC's discuss/contribute is still gated by the v0.16.0
#    per-RFC invite). No project_members row is needed and no backfill runs,
#    so the N=1 case stays whole. Explicit project_members rows and
#    gated/unlisted visibility are where the new tier actually bites.
#  * preserve curation — the implicit-public baseline does NOT override per-RFC
#    owner curation; only an *explicit* project_contributor/project_admin grant
#    (or a deployment owner/admin) bypasses it. So §22.7's "project_contributor
#    ⊇ rfc_collaborators(contributor)" holds for explicit grants, while a plain
#    granted contributor on public behaves exactly as it did before M2.
# ===========================================================================

_DEPLOYMENT_SUPERUSER_ROLES = ("owner", "admin")


def project_visibility(project_id: str) -> str:
    """The project's §22.5 visibility ('gated' | 'public' | 'unlisted'). A
    missing row reads as 'gated' — the safe default: an unknown project is
    invisible rather than open."""
    row = db.conn().execute(
        "SELECT visibility FROM projects WHERE id = ?", (project_id,)
    ).fetchone()
    if row is None:
        return "gated"
    return row["visibility"] or "gated"


def _is_default_project(project_id: str) -> bool:
    """True iff `project_id` owns the migration-seeded `default` collection — the
    deployment's primary project (§22.13), whatever its configured id. Only there
    do M2's role rows (which migrated to collection scope `default`) stand in for
    project-level authority."""
    return collections_mod.project_of_collection(collections_mod.DEFAULT_COLLECTION_ID) == project_id


def project_member_role(user: SessionUser | None, project_id: str) -> str | None:
    """The user's *project-grain* §22.6 role at this project, or None — the
    most-permissive of a **global** grant (inherits down to every project) and a
    **project**-scope grant. Mapped back to the legacy
    `project_admin`/`project_contributor` strings the project-grain authz speaks.

    Back-compat: on the deployment's *default* project only, M2's rows live at
    collection scope `default` (§B.3 migration), so a `default` collection-scope
    grant there is read as project-level too. A collection grant on any other
    project is NOT project authority — that is the four-layer collection resolver
    (`effective_scope_role`). Does not fold in the deployment tier
    (`is_project_superuser` adds it) or the implicit-on-public baseline."""
    if user is None:
        return None
    clauses = ["scope_type = 'global'", "(scope_type = 'project' AND scope_id = ?)"]
    params: list = [user.user_id, project_id]
    if _is_default_project(project_id):
        clauses.append("(scope_type = 'collection' AND scope_id = ?)")
        params.append(collections_mod.DEFAULT_COLLECTION_ID)
    row = db.conn().execute(
        "SELECT role FROM memberships WHERE user_id = ? AND (" + " OR ".join(clauses) + ") "
        "ORDER BY CASE role WHEN 'owner' THEN 0 ELSE 1 END LIMIT 1",
        params,
    ).fetchone()
    if row is None:
        return None
    return "project_admin" if row["role"] == "owner" else "project_contributor"


def project_of_rfc(rfc_slug: str) -> str:
    """The project an RFC belongs to, via its collection
    (`cached_rfcs.collection_id` -> `collections.project_id`, §22 three-tier).
    Falls back to the default project when the slug isn't cached — the same N=1
    default migration 026 backfills."""
    row = db.conn().execute(
        "SELECT c.project_id AS project_id "
        "FROM cached_rfcs r JOIN collections c ON c.id = r.collection_id "
        "WHERE r.slug = ?",
        (rfc_slug,),
    ).fetchone()
    if row is None:
        return DEFAULT_PROJECT_ID
    return row["project_id"] or DEFAULT_PROJECT_ID


def collection_of_rfc(rfc_slug: str) -> str:
    """The collection an RFC belongs to (`cached_rfcs.collection_id`). Falls back
    to the default collection when the slug isn't cached. Mirrors
    `project_of_rfc`'s first-match semantics; a slug shared across collections is
    a known routing ambiguity (the RFC-grain helpers take a bare slug) resolved
    by the collection-qualified routes in later slices."""
    row = db.conn().execute(
        "SELECT collection_id FROM cached_rfcs WHERE slug = ?", (rfc_slug,)
    ).fetchone()
    if row is None or not row["collection_id"]:
        return collections_mod.DEFAULT_COLLECTION_ID
    return row["collection_id"]


def is_project_superuser(user: SessionUser | None, project_id: str) -> bool:
    """Maximal authority within a project: a deployment owner/admin (superuser
    in every project, §22.7) or an explicit `project_admin` (§22.6). Both
    subsume the per-RFC owners/arbiters tier."""
    if user is None:
        return False
    if user.role in _DEPLOYMENT_SUPERUSER_ROLES:
        return True
    return project_member_role(user, project_id) == "project_admin"


def can_read_project(user: SessionUser | None, project_id: str) -> bool:
    """The §22.5 visibility gate at the project grain. `public`/`unlisted` are
    readable by anyone (anonymous included — `unlisted` is link-only but the link
    still reads); `gated` is readable only by a deployment owner/admin or a
    holder of any scope grant reaching the project — a global grant, a project
    grant, or membership at *any* collection within it (seeing a collection
    implies seeing its project). Used as the subtractive read gate (a gated
    project's entries 404 to non-members)."""
    vis = project_visibility(project_id)
    if vis in ("public", "unlisted"):
        return True
    # gated — scope-role holders + superusers only, subject to the §6 floor.
    if user is None or user.permission_state != "granted":
        return False
    if user.role in _DEPLOYMENT_SUPERUSER_ROLES:
        return True
    row = db.conn().execute(
        "SELECT 1 FROM memberships m WHERE m.user_id = ? AND ("
        "      m.scope_type = 'global'"
        "   OR (m.scope_type = 'project' AND m.scope_id = ?)"
        "   OR (m.scope_type = 'collection' AND m.scope_id IN "
        "        (SELECT id FROM collections WHERE project_id = ?))) LIMIT 1",
        (user.user_id, project_id, project_id),
    ).fetchone()
    return row is not None


def require_project_readable(user: SessionUser | None, project_id: str) -> None:
    """Raise 404 when the project is not readable by this viewer (§22.5: a
    gated project is invisible to non-members — indistinguishable from absent,
    so the shape matches an unknown slug)."""
    if not can_read_project(user, project_id):
        raise HTTPException(status_code=404, detail="RFC not found")


def _has_write_baseline(user: SessionUser | None, project_id: str) -> bool:
    """The implicit-on-public write baseline: a granted deployment
    `contributor` on a `public` project carries the pre-multi-project
    write standing (still subject to per-RFC curation). Deployment
    owner/admin are handled by `is_project_superuser`; on gated/unlisted a
    contributor has no baseline and needs an explicit project role."""
    if user is None or user.permission_state != "granted":
        return False
    return user.role == "contributor" and project_visibility(project_id) == "public"


def can_contribute_in_project(user: SessionUser | None, project_id: str) -> bool:
    """May the user contribute *new* content to the project (propose an entry)
    — the project-level (not RFC-specific) contribute standing. The union of
    the override grants (superuser / explicit project_contributor) and the
    implicit-public baseline, subject to the visibility gate."""
    if user is None or user.permission_state != "granted":
        return False
    if not can_read_project(user, project_id):
        return False
    if is_project_superuser(user, project_id):
        return True
    if project_member_role(user, project_id) == "project_contributor":
        return True
    return _has_write_baseline(user, project_id)


def can_discuss_in_project(user: SessionUser | None, project_id: str) -> bool:
    """May the user participate in discussion in the project at all — the
    project-level discuss standing (project_viewer ⊇ discussant, §22.7).
    A superset of `can_contribute_in_project` (a contributor can discuss)."""
    if user is None or user.permission_state != "granted":
        return False
    if not can_read_project(user, project_id):
        return False
    if is_project_superuser(user, project_id):
        return True
    if project_member_role(user, project_id) in ("project_viewer", "project_contributor"):
        return True
    return _has_write_baseline(user, project_id)


def visible_project_ids(user: SessionUser | None) -> list[str]:
    """Project ids whose entries may surface in a listing for this viewer — the
    §22.5 read gate applied to the catalog/idea lists. (The directory's
    `unlisted`-omission and the per-project routing are M3 concerns; for the M2
    listing filter we include every project the viewer can read.)"""
    rows = db.conn().execute("SELECT id FROM projects").fetchall()
    return [r["id"] for r in rows if can_read_project(user, r["id"])]


# ===========================================================================
# §22 three-tier — S3. The four-layer scope-role resolver (§B.2) and the
# collection-grain visibility gate.
#
# A grant attaches the unified role {owner, contributor} at a scope: global,
# project, or collection (§B.1). Grants inherit downward, are additive, and
# admit no negative override (§B.2). Effective authority over a *collection* is
# the most-permissive union of the layers reaching it:
#
#     global   (users.role owner/admin  ∪  memberships scope_type='global')
#   ∪ project  (memberships scope_type='project'    at the collection's project)
#   ∪ collection (memberships scope_type='collection' at the collection)
#
# minus the §22.5 visibility gate and §6.2 write-mute (subtractive, as today).
# Per-entry authority (owners / arbiters / rfc_collaborators) is a distinct,
# finer layer the RFC-grain helpers union in beneath collection.
#
# OPERATOR DECISIONS (S3, session 0076):
#  * scope-role-primary — a plain granted account (users.role 'contributor', no
#    membership) is a granted *account*, not a write-everywhere global role.
#    Write standing comes from an explicit scope grant; the lone exception is the
#    grandfathered implicit-public baseline below.
#  * grandfathered baseline — the migration-seeded `default` collection keeps the
#    pre-three-tier implicit-on-public write baseline (a granted deployment
#    contributor may propose while it is public), so the N=1 deployment (§22.13)
#    loses no capability. Every *explicitly-created* collection requires an
#    explicit scope grant to write.
#  * hidden-from-public — a `gated` collection is invisible to the public (404,
#    omitted from the directory) yet visible to any scope-role holder reaching it
#    (collection/project/global). A collection's visibility may be set only as
#    strict or stricter than its project's (the rank ordering below).
# ===========================================================================

# The global scope is a single tier per deployment; its grant rows use this
# sentinel scope_id (migration 030).
GLOBAL_SCOPE_ID = "*"

# §22.5 visibility strictness on the public-exposure axis: `public` is least
# strict, `gated` most strict. A collection may narrow its project's visibility
# but never widen it (`rank(collection) >= rank(project)`).
_VISIBILITY_RANK = {"public": 0, "unlisted": 1, "gated": 2}


def visibility_rank(visibility: str | None) -> int:
    """The strictness rank of a §22.5 visibility (higher = stricter). An unknown
    value reads as the strictest (`gated`) — the safe default."""
    return _VISIBILITY_RANK.get(visibility or "", _VISIBILITY_RANK["gated"])


def effective_scope_role(user: SessionUser | None, collection_id: str) -> str | None:
    """The most-permissive unified role ({'owner','contributor'}) the user holds
    over `collection_id`, folding §B.2's global → project → collection layers.
    Returns None when no scope grant reaches the collection. 'owner' outranks
    'contributor'; there is no negative override (a parent grant is never
    subtracted by a child). Subject to the §6 admission floor."""
    if user is None or user.permission_state != "granted":
        return None
    # Global tier — a deployment owner/admin is a global Owner (§B.1).
    if user.role in _DEPLOYMENT_SUPERUSER_ROLES:
        return "owner"
    pid = collections_mod.project_of_collection(collection_id)
    row = db.conn().execute(
        "SELECT role FROM memberships "
        "WHERE user_id = ? AND ("
        "      scope_type = 'global'"
        "   OR (scope_type = 'project' AND scope_id = ?)"
        "   OR (scope_type = 'collection' AND scope_id = ?)) "
        "ORDER BY CASE role WHEN 'owner' THEN 0 ELSE 1 END LIMIT 1",
        (user.user_id, pid, collection_id),
    ).fetchone()
    return row["role"] if row else None


def _effective_project_role(user: SessionUser | None, project_id: str) -> str | None:
    """The most-permissive role the user holds *over a project* — folding the
    global tier (deployment owner/admin, or a `scope_type='global'` grant) and a
    `scope_type='project'` grant on this project. Unlike `effective_scope_role`
    (which keys on a collection), this answers the project grain directly, for the
    §22.8 request-to-join membership check. Subject to the §6 admission floor."""
    if user is None or user.permission_state != "granted":
        return None
    if user.role in _DEPLOYMENT_SUPERUSER_ROLES:
        return "owner"
    row = db.conn().execute(
        "SELECT role FROM memberships "
        "WHERE user_id = ? AND ("
        "      scope_type = 'global'"
        "   OR (scope_type = 'project' AND scope_id = ?)) "
        "ORDER BY CASE role WHEN 'owner' THEN 0 ELSE 1 END LIMIT 1",
        (user.user_id, project_id),
    ).fetchone()
    return row["role"] if row else None


def effective_role_at_scope(
    user: SessionUser | None, scope_type: str, scope_id: str
) -> str | None:
    """The most-permissive scope role the user holds over a `(scope_type,
    scope_id)` target — the scope-grain twin of `effective_scope_role`. A
    `collection` target folds global → project → collection (the existing
    resolver); a `project` target folds global → project. Returns None when no
    grant reaches the scope. Drives the §22.8 "already a member?" gate."""
    if scope_type == "collection":
        return effective_scope_role(user, scope_id)
    if scope_type == "project":
        return _effective_project_role(user, scope_id)
    return None


def collection_visibility(collection_id: str) -> str:
    """The collection's own §22.5 visibility. A missing row reads as 'gated' —
    an unknown collection is invisible rather than open."""
    row = db.conn().execute(
        "SELECT visibility FROM collections WHERE id = ?", (collection_id,)
    ).fetchone()
    if row is None or not row["visibility"]:
        return "gated"
    return row["visibility"]


def effective_collection_visibility(collection_id: str) -> str:
    """The stricter of the collection's own visibility and its project's (§22.5
    'both gates'). A collection is constrained to be ≥ its project in strictness,
    but we max() defensively so a misconfigured looser collection can never widen
    its project's gate."""
    cvis = collection_visibility(collection_id)
    pid = collections_mod.project_of_collection(collection_id)
    pvis = project_visibility(pid) if pid else "gated"
    return cvis if visibility_rank(cvis) >= visibility_rank(pvis) else pvis


def can_read_collection(user: SessionUser | None, collection_id: str) -> bool:
    """§22.5 read/existence gate at the collection grain. `public`/`unlisted`
    read by anyone (anonymous included — `unlisted` is link-only but the link
    reads); `gated` ("hidden from public existence") reads only for a scope-role
    holder over the collection (collection/project/global) or a deployment
    owner/admin. The subtractive read gate — a gated collection 404s a
    non-holder, indistinguishable from absent."""
    vis = effective_collection_visibility(collection_id)
    if vis in ("public", "unlisted"):
        return True
    if user is None or user.permission_state != "granted":
        return False
    return effective_scope_role(user, collection_id) is not None


def require_collection_readable(user: SessionUser | None, collection_id: str) -> None:
    """Raise 404 when the collection is not readable by this viewer (§22.5: a
    hidden/gated collection is invisible to non-holders — the shape matches an
    unknown collection)."""
    if not can_read_collection(user, collection_id):
        raise HTTPException(status_code=404, detail="Not found")


def is_collection_superuser(user: SessionUser | None, collection_id: str) -> bool:
    """Maximal authority over a collection: an effective scope role of 'owner'
    reaching it (a collection Owner, a project Owner of its project, a global
    Owner, or a deployment owner/admin). Subsumes the per-entry owners/arbiters
    tier within the collection."""
    return effective_scope_role(user, collection_id) == "owner"


def _has_collection_write_baseline(user: SessionUser | None, collection_id: str) -> bool:
    """The grandfathered implicit-on-public write baseline, narrowed to the
    migration-seeded `default` collection (§22.13 N=1 case). A granted deployment
    `contributor` keeps its pre-three-tier write standing on the default
    collection while its effective visibility is public; every explicitly-created
    collection requires an explicit scope grant (S3 operator decision)."""
    if user is None or user.permission_state != "granted":
        return False
    if collection_id != collections_mod.DEFAULT_COLLECTION_ID:
        return False
    return user.role == "contributor" and effective_collection_visibility(collection_id) == "public"


def can_contribute_in_collection(user: SessionUser | None, collection_id: str) -> bool:
    """May the user contribute *new* content to the collection (propose an entry)
    — the collection-level contribute standing. The union of the scope-role grant
    (owner/contributor reaching the collection) and the grandfathered default
    baseline, subject to the visibility read gate."""
    if user is None or user.permission_state != "granted":
        return False
    if not can_read_collection(user, collection_id):
        return False
    if effective_scope_role(user, collection_id) is not None:
        return True
    return _has_collection_write_baseline(user, collection_id)


def can_discuss_in_collection(user: SessionUser | None, collection_id: str) -> bool:
    """May the user participate in discussion in the collection — the
    collection-level discuss standing. A superset of contribute for this pass
    (the read-only viewer tier is deferred, §B.3), so it mirrors
    `can_contribute_in_collection`."""
    return can_contribute_in_collection(user, collection_id)


def can_create_collection(user: SessionUser | None, project_id: str) -> bool:
    """May the user create a new collection in this project (§B.1)? Creating a
    collection is a *project-level* action: a deployment owner/admin, or any
    holder of a project-scope or global-scope grant (Owner OR RFC Contributor —
    'anyone at the project level with permission to create a collection'). A
    *collection*-scope grant cannot create sibling collections."""
    if user is None or user.permission_state != "granted":
        return False
    if user.role in _DEPLOYMENT_SUPERUSER_ROLES:
        return True
    row = db.conn().execute(
        "SELECT 1 FROM memberships "
        "WHERE user_id = ? AND ("
        "      scope_type = 'global'"
        "   OR (scope_type = 'project' AND scope_id = ?)) LIMIT 1",
        (user.user_id, project_id),
    ).fetchone()
    return row is not None


def can_create_project(user: SessionUser | None) -> bool:
    """§22 S5 (§A.2 / §B.1): may the user create a new project? "+ New project"
    is a **global-Owner** action — a deployment owner/admin (a global Owner per
    §B.1) or a holder of an explicit `scope_type='global'` Owner grant. Creating
    a project is deployment-level, so it is not reachable by a project- or
    collection-scope grant nor by a global RFC Contributor (that role creates
    collections, not projects). Subject to the §6 admission floor."""
    if user is None or user.permission_state != "granted":
        return False
    if user.role in _DEPLOYMENT_SUPERUSER_ROLES:
        return True
    row = db.conn().execute(
        "SELECT 1 FROM memberships "
        "WHERE user_id = ? AND scope_type = 'global' AND role = 'owner' LIMIT 1",
        (user.user_id,),
    ).fetchone()
    return row is not None


def can_invite_at_project(user: SessionUser | None, project_id: str) -> bool:
    """§22 S4 (C.2): may the user grant scope roles at this project (or at any
    collection within it)? Managing membership is an *Owner* capability whose
    reach covers the project — a deployment owner/admin, a global Owner, or this
    project's Owner. An RFC Contributor does not manage membership (C.2.4); a
    collection Owner's reach is its own collection only (C.2.3), so it is not
    offered project-scope invites. Identical to `is_project_superuser` — the
    invite gate IS "is an Owner over this project"."""
    return is_project_superuser(user, project_id)


def can_invite_at_collection(user: SessionUser | None, collection_id: str) -> bool:
    """§22 S4 (C.2): may the user grant scope roles at this collection? An Owner
    whose reach covers it — the collection's Owner, its project's Owner, a global
    Owner, or a deployment owner/admin (`is_collection_superuser`). This is the
    narrowest invite reach; a collection Owner who is nothing more may invite
    here but not at the project or globally (C.2.3)."""
    return is_collection_superuser(user, collection_id)


# v0.16.0 (roadmap item #12): per-RFC membership helpers.
#
# These don't replace `require_contributor` — they layer on top of it for
# endpoints that an RFC's owner can selectively open up. The "discussion"
# and "PR" write surfaces consult `is_rfc_writer(...)` / `is_rfc_discussant(...)`
# to admit users who are either platform-privileged (admin, RFC owner)
# OR who hold an explicit invitation-accepted per-RFC role.
#
# The platform gate still fires first: a user whose
# `permission_state != 'granted'` cannot write anywhere, invitation or
# not. v0.16.0 doesn't loosen that — a per-RFC invitation is additive
# *within* the granted-platform-user population. (Accepting an
# invitation as a pending user surfaces in the admin-page hook per
# the roadmap text; the platform grant remains the admin's decision.)


def _rfc_owners_set(rfc_slug: str) -> set[str]:
    """The gitea_logins named in the RFC's frontmatter owners array.

    Read from `cached_rfcs.owners_json`. Returns an empty set if the RFC
    isn't cached (the caller's earlier `_require_rfc_readable` will
    already have rejected that case in practice).
    """
    import json as _json
    row = db.conn().execute(
        "SELECT owners_json FROM cached_rfcs WHERE slug = ?", (rfc_slug,),
    ).fetchone()
    if row is None:
        return set()
    try:
        return set(_json.loads(row["owners_json"] or "[]"))
    except Exception:
        return set()


def is_rfc_owner(user: SessionUser | None, rfc_slug: str) -> bool:
    """True iff the user is named in the RFC's frontmatter `owners`
    list. The platform-level admin/owner check is separate; per §6.1 an
    app admin/owner has all per-RFC capabilities by construction, but
    this predicate is intentionally narrow — it answers "is this
    person on the RFC's owners line?" and nothing more.
    """
    if user is None:
        return False
    return user.gitea_login in _rfc_owners_set(rfc_slug)


def is_rfc_collaborator(user: SessionUser | None, rfc_slug: str, *, role_in_rfc: str | None = None) -> bool:
    """True iff the user has an accepted per-RFC collaborator row.

    `role_in_rfc`:
      * None — any role qualifies (the discussion-write check uses this
        shape: contributor strictly includes discussant).
      * 'contributor' — only the contributor role qualifies (the PR-write
        check uses this shape).
      * 'discussant' — only the discussant role qualifies (not used by
        v0.16.0 endpoints; included for symmetry).
    """
    if user is None:
        return False
    if role_in_rfc is None:
        row = db.conn().execute(
            "SELECT 1 FROM rfc_collaborators WHERE rfc_slug = ? AND user_id = ? LIMIT 1",
            (rfc_slug, user.user_id),
        ).fetchone()
        return row is not None
    row = db.conn().execute(
        "SELECT 1 FROM rfc_collaborators WHERE rfc_slug = ? AND user_id = ? AND role_in_rfc = ? LIMIT 1",
        (rfc_slug, user.user_id, role_in_rfc),
    ).fetchone()
    return row is not None


def can_discuss_rfc(user: SessionUser | None, rfc_slug: str) -> bool:
    """v0.16.0 — admit to PR-less discussion writes on this RFC.

    True if ANY of:
      * platform admin/owner (the §6.1 maximal-capability path),
      * the RFC has no frontmatter owners yet (the gate is open
        until an owner exists to set it — relevant for super-drafts
        pre-§13.1 claim),
      * RFC owner (frontmatter `owners` membership),
      * accepted per-RFC collaborator at any role (contributor strictly
        includes discussant).

    Returns False for anonymous viewers and for users whose
    `permission_state != 'granted'` — the platform-level gate must hold
    before any per-RFC layer can apply. The platform gate is also
    enforced earlier in the request via `require_contributor`; the
    helper here is defensive so callers that compose it with
    `current_user` directly still respect the gate.
    """
    if user is None:
        return False
    if user.permission_state != "granted":
        return False
    cid = collection_of_rfc(rfc_slug)
    # §22.5 visibility gate is subtractive (§22.7) — no capability in a
    # collection the viewer cannot even read.
    if not can_read_collection(user, cid):
        return False
    # §B.2 union, scope-role grants first — these bypass per-RFC curation (a
    # collection/project/global Owner or RFC Contributor ⊇ discussant).
    if effective_scope_role(user, cid) is not None:
        return True
    # per-RFC authority (union term).
    owners = _rfc_owners_set(rfc_slug)
    if user.gitea_login in owners:
        return True
    if is_rfc_collaborator(user, rfc_slug, role_in_rfc=None):
        return True
    # grandfathered implicit-public baseline (curation preserved): on the default
    # collection a granted deployment contributor may discuss only while the RFC
    # is unclaimed. The first §13.1 claim engages the per-RFC gate, mirroring the
    # pre-multi-project v0.16.0 contract.
    if not owners and _has_collection_write_baseline(user, cid):
        return True
    return False


def can_contribute_to_rfc(user: SessionUser | None, rfc_slug: str) -> bool:
    """v0.16.0 — admit to PR-shaped writes on this RFC.

    True if ANY of:
      * platform admin/owner,
      * the RFC has no frontmatter owners yet (gate open until an
        owner exists),
      * RFC owner,
      * accepted per-RFC collaborator at role 'contributor' (a
        'discussant' row is NOT sufficient — PRs are the
        higher-privilege surface).

    Same `permission_state` and anonymous-viewer refusals as
    `can_discuss_rfc`.
    """
    if user is None:
        return False
    if user.permission_state != "granted":
        return False
    cid = collection_of_rfc(rfc_slug)
    if not can_read_collection(user, cid):
        return False
    # §B.2 union, scope-role grants first (a collection/project/global RFC
    # Contributor ⊇ rfc_collaborators(contributor); an Owner ⊇ all).
    if effective_scope_role(user, cid) is not None:
        return True
    # per-RFC authority (union term). A 'discussant' row is NOT sufficient —
    # PRs are the higher-privilege surface.
    owners = _rfc_owners_set(rfc_slug)
    if user.gitea_login in owners:
        return True
    if is_rfc_collaborator(user, rfc_slug, role_in_rfc="contributor"):
        return True
    # grandfathered implicit-public baseline (curation preserved): until an owner
    # exists, a granted deployment contributor on the public default collection
    # may contribute.
    if not owners and _has_collection_write_baseline(user, cid):
        return True
    return False


def can_invite_to_rfc(user: SessionUser | None, rfc_slug: str) -> bool:
    """v0.16.0 — only RFC owners (frontmatter) and platform admin/owner
    can issue invitations. Per-RFC collaborators do not get the
    invite-others power; that stays with the RFC's owner."""
    if user is None:
        return False
    if user.permission_state != "granted":
        return False
    cid = collection_of_rfc(rfc_slug)
    if not can_read_collection(user, cid):
        return False
    # An Owner reaching the collection (collection/project/global Owner, or a
    # deployment owner/admin) may invite; otherwise only the RFC's frontmatter
    # owner. Per-RFC collaborators and the baseline do not get the invite power.
    if is_collection_superuser(user, cid):
        return True
    return is_rfc_owner(user, rfc_slug)


def new_state() -> str:
    return secrets.token_urlsafe(16)