FabledGrimoire/src/scribe/services/access.py

"""Access control service.

Single source of truth for permission resolution on projects and notes.
All human-facing routes that should honour shares call these functions.
LLM tool routes continue to use owner-scoped service functions directly.

Permission rank: owner > admin > editor > viewer
"""

import logging

from sqlalchemy import or_, select

from scribe.models import async_session
from scribe.models.group import GroupMembership
from scribe.models.note import Note
from scribe.models.project import Project
from scribe.models.share import NoteShare, ProjectShare
from scribe.models.user import User

logger = logging.getLogger(__name__)

PERMISSION_RANK: dict[str, int] = {
    "viewer": 1,
    "editor": 2,
    "admin": 3,
    "owner": 4,
}


def _higher(a: str | None, b: str | None) -> str | None:
    """Return the higher-ranked permission, or None if both are None."""
    if a is None:
        return b
    if b is None:
        return a
    return a if PERMISSION_RANK[a] >= PERMISSION_RANK[b] else b


async def _user_group_ids(session, user_id: int) -> list[int]:
    rows = (
        await session.execute(
            select(GroupMembership.group_id).where(GroupMembership.user_id == user_id)
        )
    ).scalars().all()
    return list(rows)


# ---------------------------------------------------------------------------
# Project permissions
# ---------------------------------------------------------------------------

async def get_project_permission(user_id: int, project_id: int) -> str | None:
    """Return the effective permission string for user on project, or None."""
    async with async_session() as session:
        project = await session.get(Project, project_id)
        if project is None:
            return None
        if project.user_id == user_id:
            return "owner"

        # Direct share
        direct = (
            await session.execute(
                select(ProjectShare).where(
                    ProjectShare.project_id == project_id,
                    ProjectShare.shared_with_user_id == user_id,
                )
            )
        ).scalar_one_or_none()

        # Group shares
        group_ids = await _user_group_ids(session, user_id)
        group_perm: str | None = None
        if group_ids:
            group_shares = (
                await session.execute(
                    select(ProjectShare).where(
                        ProjectShare.project_id == project_id,
                        ProjectShare.shared_with_group_id.in_(group_ids),
                    )
                )
            ).scalars().all()
            for gs in group_shares:
                group_perm = _higher(group_perm, gs.permission)

        return _higher(direct.permission if direct else None, group_perm)


async def can_read_project(user_id: int, project_id: int) -> bool:
    return (await get_project_permission(user_id, project_id)) is not None


async def can_write_project(user_id: int, project_id: int) -> bool:
    perm = await get_project_permission(user_id, project_id)
    return perm in ("editor", "admin", "owner")


async def can_admin_project(user_id: int, project_id: int) -> bool:
    perm = await get_project_permission(user_id, project_id)
    return perm in ("admin", "owner")


# ---------------------------------------------------------------------------
# Note / task permissions
# ---------------------------------------------------------------------------

async def get_note_permission(user_id: int, note_id: int) -> str | None:
    """Return the effective permission for user on a note/task, or None.

    Resolution order:
    1. Ownership
    2. Direct note share
    3. Group-based note share
    4. Inherited from project share (if note belongs to a project)
    Highest rank wins.
    """
    async with async_session() as session:
        note = await session.get(Note, note_id)
        if note is None:
            return None
        if note.user_id == user_id:
            return "owner"

        direct = (
            await session.execute(
                select(NoteShare).where(
                    NoteShare.note_id == note_id,
                    NoteShare.shared_with_user_id == user_id,
                )
            )
        ).scalar_one_or_none()

        group_ids = await _user_group_ids(session, user_id)
        group_perm: str | None = None
        if group_ids:
            group_shares = (
                await session.execute(
                    select(NoteShare).where(
                        NoteShare.note_id == note_id,
                        NoteShare.shared_with_group_id.in_(group_ids),
                    )
                )
            ).scalars().all()
            for gs in group_shares:
                group_perm = _higher(group_perm, gs.permission)

        note_perm = _higher(direct.permission if direct else None, group_perm)

        # Inherit from project if note belongs to one
        if note.project_id is not None:
            project_perm = await get_project_permission(user_id, note.project_id)
            note_perm = _higher(note_perm, project_perm)

        return note_perm


async def can_read_note(user_id: int, note_id: int) -> bool:
    return (await get_note_permission(user_id, note_id)) is not None


async def can_write_note(user_id: int, note_id: int) -> bool:
    perm = await get_note_permission(user_id, note_id)
    return perm in ("editor", "admin", "owner")


# ---------------------------------------------------------------------------
# Set-based visibility (for LIST queries)
#
# Two scopes, deliberately different (decision note 2094):
#
#   readable_*  — everything the ACL permits, including a record reachable ONLY
#                 through a direct or group note share. For EXPLICIT acts: a
#                 search the caller typed, or a fetch by id.
#   browsable_* — the caller's own records plus anything in a project they have
#                 access to. For PASSIVE surfaces: browse lists, facet counts,
#                 the process→skill manifest.
#
# The split is a trust boundary, not an optimisation. Anything that appears
# unasked — in your own list, your own counts, or as a skill installed on your
# machine — reads as material you endorsed. A one-off record someone shared with
# you has not earned that standing, so it waits until you go looking for it.
# ---------------------------------------------------------------------------

def notes_visibility_clause(user_id: int, scope: str = "own"):
    """The one place a retrieval declares how far it may see.

    Every path that returns notes picks a scope by what KIND of act it is:

      "own"    — the caller's records only. For machinery whose answer must not
                 depend on other people: the near-duplicate gate can't block a
                 create because a stranger wrote something similar, and can't
                 point at a record the caller cannot edit.
      "browse" — own + project-reachable. For passive surfaces, where an
                 unrequested record would read as endorsed.
      "read"   — everything the ACL permits. For explicit acts: a typed search,
                 a fetch by id.

    Defaults to the narrowest, so a new caller that forgets to choose is wrong in
    the safe direction.
    """
    if scope == "own":
        return Note.user_id == user_id
    if scope == "browse":
        return browsable_notes_clause(user_id)
    if scope == "read":
        return readable_notes_clause(user_id)
    raise ValueError(f"unknown note scope {scope!r} (own | browse | read)")


async def owner_names_for(user_ids: set[int]) -> dict[int, str]:
    """{user_id: username} in one query. Empty input costs nothing."""
    if not user_ids:
        return {}
    async with async_session() as session:
        rows = (
            await session.execute(
                select(User.id, User.username).where(User.id.in_(user_ids))
            )
        ).all()
    return {uid: name for uid, name in rows}


def _my_group_ids(user_id: int):
    """The caller's group ids as a SUBQUERY, not a fetched list.

    Keeping it in SQL is what makes the clause builders below pure functions —
    no session, no await, nothing for a caller's unit test to mock — and it folds
    the membership lookup into the one statement the caller was already running.
    """
    return select(GroupMembership.group_id).where(GroupMembership.user_id == user_id)


def readable_notes_clause(user_id: int):
    """A SQLAlchemy WHERE predicate matching every note this user may READ.

    `get_note_permission` answers "may I read THIS note?" one row at a time,
    which a list query can't use — checking per row is O(n) round-trips. This is
    the same resolution expressed as set membership so it can go straight into a
    `select(Note).where(...)`:

        1. ownership
        2. a direct note share
        3. a group note share
        4. inheritance from a shared project

    Keep the two in step: a rule added here belongs in `get_note_permission`
    too, or a note becomes findable but not openable (or the reverse — which is
    the bug this was written to fix).
    """
    groups = _my_group_ids(user_id)

    shared_note_ids = select(NoteShare.note_id).where(
        or_(
            NoteShare.shared_with_user_id == user_id,
            NoteShare.shared_with_group_id.in_(groups),
        )
    )
    shared_project_ids = select(ProjectShare.project_id).where(
        or_(
            ProjectShare.shared_with_user_id == user_id,
            ProjectShare.shared_with_group_id.in_(groups),
        )
    )

    return or_(
        Note.user_id == user_id,
        Note.id.in_(shared_note_ids),
        Note.project_id.in_(shared_project_ids),
    )


async def describe_provenance(user_id: int, note) -> dict:
    """Provenance for a record being handed to a caller: `{}` when it's theirs,
    otherwise `{"shared": True, "owner": <username>, "permission": <perm>}`.

    Anything reaching an agent or a UI from another person has to say so. A
    shared record is one person's suggestion, not a standard the caller adopted,
    and without a marker the two are indistinguishable — the reader would assume
    their own past self wrote it and treat it as settled practice.
    """
    if note is None or note.user_id == user_id:
        return {}
    async with async_session() as session:
        owner = await session.get(User, note.user_id)
    return {
        "shared": True,
        "owner": owner.username if owner else None,
        "permission": await get_note_permission(user_id, note.id),
    }


async def label_shared_items(user_id: int, items: list[dict]) -> list[dict]:
    """Mark the entries in a list payload that belong to someone else.

    Each foreign item gains `shared: True` and `owner: <username>`; the caller's
    own items are left untouched so an all-mine list stays noise-free. Usernames
    are resolved in one query per distinct owner, not one per row.

    Lists are where provenance matters most: an unmarked row in your own list
    reads as something you recorded and vetted.
    """
    foreign = {
        it["user_id"] for it in items
        if it.get("user_id") is not None and it["user_id"] != user_id
    }
    if not foreign:
        return items
    names = await owner_names_for(foreign)
    for it in items:
        owner_id = it.get("user_id")
        if owner_id is not None and owner_id != user_id:
            it["shared"] = True
            it["owner"] = names.get(owner_id)
    return items


def browsable_notes_clause(user_id: int):
    """A WHERE predicate for PASSIVE surfaces: the caller's own notes, plus
    notes in a project they can reach (owned or shared).

    Deliberately narrower than `readable_notes_clause` — it omits records the
    caller can read *only* via a direct or group note share. Those stay
    search-only, so a one-off someone handed them never drifts into a browse
    list, a facet count, or the skill manifest as though it were their own.

    Project membership is the seam because it's a standing, mutual context: if
    you're on the project, its content is your working material. A note with no
    project that you don't own therefore never matches — `NULL IN (...)` is not
    true — which is exactly the intent.
    """
    shared_project_ids = select(ProjectShare.project_id).where(
        or_(
            ProjectShare.shared_with_user_id == user_id,
            ProjectShare.shared_with_group_id.in_(_my_group_ids(user_id)),
        )
    )
    owned_project_ids = select(Project.id).where(Project.user_id == user_id)

    return or_(
        Note.user_id == user_id,
        Note.project_id.in_(shared_project_ids),
        Note.project_id.in_(owned_project_ids),
    )