rfc-app/backend/app/memberships.py

"""§22 S4 (C.2) — scope-role grant operations over the `memberships` table.

The membership *gates* (who may invite, who holds which role) live in
`auth.py`; this module holds the *mutations* the invitation surface drives —
granting, the "broader scope supersedes narrower" cleanup, listing, and
revocation — mirroring how `invites.py` owns the create/claim/list of per-user
invite tokens while the gate (`auth.can_invite_to_rfc`) lives in `auth.py`.

The model (Part B / S3): a `memberships` row is `(scope_type ∈ {global,
project, collection}, scope_id, user_id, role ∈ {owner, contributor})`, unique
per `(scope_type, scope_id, user_id)`. A grant is a direct write of that row
(the C.2 scenarios write the row immediately and §15-notify an existing
account — there is no accept round-trip; inviting a not-yet-account email is
out of S4 scope and handled by the admin-create-invite path).

The "broader scope supersedes narrower" rule (C.2.6): granting a role at a
broader scope removes this user's narrower rows that the new grant *subsumes*
— a narrower row whose role is no more permissive than the new one. A narrower
row that is *more* permissive is kept (no negative override: a child Owner
grant survives a parent Contributor grant, and the §B.2 resolver still unions
most-permissively).
"""
from __future__ import annotations

from typing import Any

from . import collections as collections_mod
from . import db

# Higher rank = more permissive. Used by the supersede rule: a narrower grant
# is pruned only when its rank ≤ the new broader grant's rank.
_ROLE_RANK = {"contributor": 1, "owner": 2}

VALID_SCOPE_TYPES = ("global", "project", "collection")
VALID_ROLES = ("owner", "contributor")
GLOBAL_SCOPE_ID = "*"


def user_by_email(email: str) -> dict[str, Any] | None:
    """The `users` row (id, display_name, email, permission_state) for an
    email, case-insensitively, or None. The grantee must already be an account
    — S4 grants a scope role to an existing user, it does not provision one."""
    row = db.conn().execute(
        "SELECT id, display_name, email, permission_state, role "
        "FROM users WHERE lower(email) = lower(?) "
        "ORDER BY id LIMIT 1",
        (email.strip(),),
    ).fetchone()
    return dict(row) if row else None


def grant(
    *,
    scope_type: str,
    scope_id: str,
    user_id: int,
    role: str,
    granted_by: int | None,
) -> None:
    """Write (or update) the membership row, then apply the C.2.6
    broader-scope-supersedes cleanup. Idempotent on `(scope_type, scope_id,
    user_id)` — re-granting at the same scope updates the role and the grantor.

    The grant is recorded regardless of the grantee's deployment
    `permission_state`: a `pending` account's row is written (C.2.7), but the
    §6 admission floor in `auth.effective_scope_role` keeps it conferring no
    write until the account is granted at the deployment."""
    db.conn().execute(
        "INSERT INTO memberships (scope_type, scope_id, user_id, role, granted_by) "
        "VALUES (?, ?, ?, ?, ?) "
        "ON CONFLICT (scope_type, scope_id, user_id) "
        "DO UPDATE SET role = excluded.role, granted_by = excluded.granted_by, "
        "granted_at = datetime('now')",
        (scope_type, scope_id, user_id, role, granted_by),
    )
    _prune_subsumed(scope_type=scope_type, scope_id=scope_id, user_id=user_id, role=role)


def _prune_subsumed(*, scope_type: str, scope_id: str, user_id: int, role: str) -> None:
    """Remove this user's narrower rows that the just-written broader grant
    subsumes (same-or-lower role rank within the broader scope's subtree). A
    collection grant subsumes nothing narrower (the per-entry tier is separate);
    a project grant subsumes its collections; a global grant subsumes every
    project and collection."""
    rank = _ROLE_RANK[role]
    keep_ranks = [r for r, v in _ROLE_RANK.items() if v <= rank]
    if not keep_ranks:
        return
    placeholders = ",".join("?" for _ in keep_ranks)
    if scope_type == "project":
        # Narrower = collection-scope rows for collections in this project.
        db.conn().execute(
            f"DELETE FROM memberships "
            f"WHERE user_id = ? AND scope_type = 'collection' "
            f"  AND role IN ({placeholders}) "
            f"  AND scope_id IN (SELECT id FROM collections WHERE project_id = ?)",
            (user_id, *keep_ranks, scope_id),
        )
    elif scope_type == "global":
        # Narrower = every project- and collection-scope row for this user.
        db.conn().execute(
            f"DELETE FROM memberships "
            f"WHERE user_id = ? AND scope_type IN ('project', 'collection') "
            f"  AND role IN ({placeholders})",
            (user_id, *keep_ranks),
        )


def revoke(*, scope_type: str, scope_id: str, user_id: int) -> bool:
    """Remove a membership row at exactly this scope. Returns True if a row was
    removed. Revocation is scope-exact: it does not cascade to broader or
    narrower grants (each is its own administrative act)."""
    cur = db.conn().execute(
        "DELETE FROM memberships WHERE scope_type = ? AND scope_id = ? AND user_id = ?",
        (scope_type, scope_id, user_id),
    )
    return cur.rowcount > 0


def list_for_project(project_id: str) -> list[dict[str, Any]]:
    """Every project-scope grant on this project plus every collection-scope
    grant on its collections, joined to the grantee's display fields — the data
    behind the project-Owner membership panel. Ordered project grants first,
    then by collection, then by role (Owner before Contributor)."""
    rows = db.conn().execute(
        """
        SELECT m.scope_type, m.scope_id, m.user_id, m.role, m.granted_at,
               u.display_name, u.email, u.permission_state,
               c.name AS collection_name
          FROM memberships m
          JOIN users u ON u.id = m.user_id
     LEFT JOIN collections c ON c.id = m.scope_id AND m.scope_type = 'collection'
         WHERE (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 = ?))
         ORDER BY (m.scope_type != 'project'), m.scope_id,
                  CASE m.role WHEN 'owner' THEN 0 ELSE 1 END, u.display_name
        """,
        (project_id, project_id),
    ).fetchall()
    return [dict(r) for r in rows]