rfc-app/backend/app/api_collections.py

"""§22 S2 — collection directory + create-collection.

GET  /api/projects/:id/collections        — list the project's visible collections.
GET  /api/projects/:id/collections/:cid    — one collection's settings.
POST /api/projects/:id/collections         — create a collection. Authorized by a
                                             deployment owner/admin (S2; scoped
                                             {owner, contributor} roles at the
                                             collection axis land in S3). The bot
                                             commits a `.collection.yaml` to the
                                             content repo, then the registry mirror
                                             upserts the collections row — §22.2
                                             keeps the registry the source of truth.
"""
from __future__ import annotations

import re
from typing import Any

import yaml
from fastapi import APIRouter, HTTPException, Request
from pydantic import BaseModel

from . import (
    auth,
    collections as collections_mod,
    projects as projects_mod,
    registry as registry_mod,
)
from .bot import Bot
from .config import Config
from .gitea import Gitea, GiteaError

_SLUG_RE = re.compile(r"^[a-z0-9](?:[a-z0-9-]*[a-z0-9])?$")


class CreateCollectionBody(BaseModel):
    collection_id: str
    type: str
    name: str | None = None
    visibility: str | None = None
    initial_state: str | None = None


def _project_viewer_caps(viewer: Any, project_id: str) -> dict[str, Any]:
    """§22 S4: the viewer's project-grain capabilities for role-aware UI — may
    they create a collection, may they manage membership (invite), and their
    project role. `role` maps the §22.6 legacy strings back to the unified
    `{owner, contributor}` vocabulary the frontend speaks."""
    legacy = auth.project_member_role(viewer, project_id)
    role = None
    if viewer is not None and viewer.role in ("owner", "admin"):
        role = "owner"
    elif legacy == "project_admin":
        role = "owner"
    elif legacy == "project_contributor":
        role = "contributor"
    return {
        "can_create_collection": auth.can_create_collection(viewer, project_id),
        "can_invite": auth.can_invite_at_project(viewer, project_id),
        # §22.8: a signed-in, granted account with no role at the project may ask
        # to join it (the request-to-join affordance). Owners/members and
        # not-yet-granted accounts don't see it.
        "can_request_join": (
            viewer is not None
            and viewer.permission_state == "granted"
            and auth.effective_role_at_scope(viewer, "project", project_id) is None
        ),
        "role": role,
    }


def make_router(config: Config, gitea: Gitea, bot: Bot) -> APIRouter:
    router = APIRouter()

    @router.get("/api/projects/{project_id}/collections")
    async def list_cols(project_id: str, request: Request) -> dict[str, Any]:
        viewer = auth.current_user(request)
        # §22.5 read gate: a gated project 404s a non-member.
        auth.require_project_readable(viewer, project_id)
        # §22.5 (S3): the directory is viewer-aware — a hidden/gated collection
        # is listed only for a scope-role holder who can read it; `unlisted` is
        # omitted from enumeration for everyone (link-only).
        items = [
            c
            for c in collections_mod.list_collections(project_id, include_unlisted=True)
            if c["visibility"] != "unlisted" and auth.can_read_collection(viewer, c["id"])
        ]
        # §22 S4: surface the viewer's project-level capabilities so the
        # directory can render role-aware affordances (the create-first-
        # collection CTA, the invite control) without a second round-trip.
        return {"items": items, "viewer": _project_viewer_caps(viewer, project_id)}

    @router.get("/api/projects/{project_id}/collections/{collection_id}")
    async def get_col(project_id: str, collection_id: str, request: Request) -> dict[str, Any]:
        viewer = auth.current_user(request)
        auth.require_project_readable(viewer, project_id)
        col = collections_mod.get_collection(collection_id)
        if col is None or col["project_id"] != project_id:
            raise HTTPException(404, "Not found")
        # §22.5 (S3): a hidden/gated collection 404s a non-scope-role viewer.
        auth.require_collection_readable(viewer, collection_id)
        # §22 S4: the viewer's collection-level capabilities drive the
        # propose-first empty state and the collection invite control.
        col = dict(col)
        col["viewer"] = {
            "can_contribute": auth.can_contribute_in_collection(viewer, collection_id),
            "can_invite": auth.can_invite_at_collection(viewer, collection_id),
            # §22.8: a signed-in, granted account with no role reaching this
            # collection may ask to join it.
            "can_request_join": (
                viewer is not None
                and viewer.permission_state == "granted"
                and auth.effective_scope_role(viewer, collection_id) is None
            ),
            "role": auth.effective_scope_role(viewer, collection_id),
        }
        return col

    @router.post("/api/projects/{project_id}/collections")
    async def create_col(
        project_id: str, body: CreateCollectionBody, request: Request
    ) -> dict[str, Any]:
        # §B.1 (S3) authority: a deployment owner/admin or a project/global-scope
        # grant holder (Owner or RFC Contributor) may create a collection. The
        # read gate runs first so a gated project 404s a non-member.
        user = auth.require_contributor(request)
        auth.require_project_readable(user, project_id)
        if not auth.can_create_collection(user, project_id):
            raise HTTPException(403, "You may not create collections in this project")
        cid = body.collection_id.strip().lower()
        if not _SLUG_RE.match(cid) or cid == "default":
            raise HTTPException(422, "collection id must be a slug and not 'default'")
        if body.type not in registry_mod.VALID_TYPES:
            raise HTTPException(422, f"invalid type {body.type!r}")
        if body.visibility is not None:
            if body.visibility not in registry_mod.VALID_VISIBILITY:
                raise HTTPException(422, f"invalid visibility {body.visibility!r}")
            # §22.5 (S3) strictness: a collection may be set only as strict or
            # stricter than its project — never more public.
            pvis = auth.project_visibility(project_id)
            if auth.visibility_rank(body.visibility) < auth.visibility_rank(pvis):
                raise HTTPException(
                    422,
                    f"collection visibility {body.visibility!r} is looser than "
                    f"the project's {pvis!r}; a collection may only narrow it",
                )
        if body.initial_state is not None and body.initial_state not in registry_mod.VALID_INITIAL_STATE:
            raise HTTPException(422, f"invalid initial_state {body.initial_state!r}")
        if collections_mod.get_collection(cid) is not None:
            raise HTTPException(409, f"collection `{cid}` already exists")
        content_repo = projects_mod.content_repo(project_id)
        if not content_repo:
            raise HTTPException(409, "project has no content repo")

        manifest: dict[str, Any] = {"type": body.type}
        if body.name:
            manifest["name"] = body.name
        if body.visibility:
            manifest["visibility"] = body.visibility
        if body.initial_state:
            manifest["initial_state"] = body.initial_state
        manifest_yaml = yaml.safe_dump(manifest, sort_keys=False)

        try:
            await bot.create_collection(
                user.as_actor(),
                org=config.gitea_org,
                content_repo=content_repo,
                collection_id=cid,
                manifest_yaml=manifest_yaml,
            )
        except GiteaError as e:
            raise HTTPException(502, f"Gitea: {e.detail}")

        # §22.2: re-read the registry so the new manifest becomes a row.
        await registry_mod.refresh_registry(config, gitea)
        col = collections_mod.get_collection(cid)
        if col is None:
            raise HTTPException(500, "collection committed but not mirrored")
        return col

    return router