FabledGrimoire/src/scribe/services/notes.py

import logging
import re
from datetime import date, datetime, timezone

from sqlalchemy import func, or_, select, text

from scribe.models import async_session
from scribe.models.note import Note, TaskKind, TaskPriority, TaskStatus

logger = logging.getLogger(__name__)


def embed_note(note) -> None:
    """Refresh a note's embedding, fire-and-forget.

    Lives HERE — at the service, not the route — so every caller gets it by
    construction. Previously each REST route made this call itself and the MCP
    tools did not, so a record created through MCP stayed out of semantic search
    and auto-inject until the next restart's backfill ran (#2056). That is
    invisible on an instance that redeploys constantly and permanent on one that
    doesn't, which is the worst shape a bug can have: it only appears where
    nobody is looking.

    Uses `note.user_id` — the OWNER — rather than the caller. Embeddings belong
    to the record, and a collaborator editing a shared note must refresh the
    owner's row rather than mint a second one under their own id.

    Import is lazy so importing this module doesn't pull in the embedding model;
    exceptions are swallowed because a record that saved must not fail on its
    index refresh. No running loop (unit tests, scripts) is an ordinary case,
    not an error.
    """
    try:
        import asyncio

        from scribe.services.embeddings import upsert_note_embedding
        # Chunking and the empty-record gate live inside upsert_note_embedding —
        # one path for every writer (#280).
        asyncio.create_task(
            upsert_note_embedding(note.id, note.user_id, note.title, note.body)
        )
    except RuntimeError:
        pass          # no running loop — a sync caller, not a failure
    except Exception:  # noqa: BLE001 - never let indexing break a write
        logger.exception("embedding refresh failed for note %s", note.id)


def _normalize_tags(tags: list[str]) -> list[str]:
    """Lowercase, strip, deduplicate, and drop empty tags."""
    seen: set[str] = set()
    out: list[str] = []
    for t in tags:
        normalized = t.strip().lower()
        if normalized and normalized not in seen:
            seen.add(normalized)
            out.append(normalized)
    return out


# Type-nouns the LLM tends to include in search queries. Treating them as
# required ILIKE terms drops literal-title matches; we strip them server-side
# and let the `type` / `project` parameters scope results instead.
_SEARCH_TYPE_NOUNS = {"task", "tasks", "note", "notes", "project", "projects"}


def _strip_type_nouns(q: str) -> list[str]:
    """Return q's tokens with type-nouns removed (case-insensitive)."""
    return [t for t in q.split() if t.lower() not in _SEARCH_TYPE_NOUNS]


async def _maybe_reactivate_project(project_id: int) -> None:
    """If a project is paused, reactivate it — activity indicates resumed work."""
    from scribe.models.project import Project
    try:
        async with async_session() as session:
            project = (await session.execute(
                select(Project).where(Project.id == project_id)
            )).scalars().first()
            if project and project.status == "paused":
                project.status = "active"
                await session.commit()
                logger.info("Auto-reactivated paused project %d (%s)", project_id, project.title)
    except Exception:
        logger.debug("_maybe_reactivate_project failed for project %d", project_id, exc_info=True)


async def create_note(
    user_id: int,
    title: str = "",
    body: str = "",
    description: str | None = None,
    tags: list[str] | None = None,
    parent_id: int | None = None,
    project_id: int | None = None,
    milestone_id: int | None = None,
    status: str | None = None,
    priority: str | None = None,
    due_date: date | None = None,
    recurrence_rule: dict | None = None,
    note_type: str = "note",
    task_kind: str = "work",
    arose_from_id: int | None = None,
    data: dict | None = None,
) -> Note:
    # Validate status/priority here so the MCP create_task path (which passes
    # them straight through) can't persist an out-of-enum value that the REST
    # route would have rejected — there's no DB CHECK on notes.status.
    if isinstance(status, str):
        try:
            status = TaskStatus(status).value
        except ValueError:
            raise ValueError(f"Invalid status: {status!r}. Must be one of: {[s.value for s in TaskStatus]}")
    if isinstance(priority, str):
        try:
            priority = TaskPriority(priority).value
        except ValueError:
            raise ValueError(f"Invalid priority: {priority!r}. Must be one of: {[p.value for p in TaskPriority]}")

    # Auto-populate project_id from milestone when not explicitly provided
    if milestone_id is not None and project_id is None:
        from scribe.models.milestone import Milestone
        async with async_session() as lookup:
            result = await lookup.execute(
                select(Milestone).where(Milestone.id == milestone_id, Milestone.user_id == user_id)
            )
            ms = result.scalars().first()
            if ms is not None:
                project_id = ms.project_id

    async with async_session() as session:
        note = Note(
            user_id=user_id,
            title=title,
            body=body,
            description=description,
            tags=_normalize_tags(tags or []),
            parent_id=parent_id,
            project_id=project_id,
            milestone_id=milestone_id,
            status=status,
            priority=priority,
            due_date=due_date,
            recurrence_rule=recurrence_rule,
            note_type=note_type,
            task_kind=task_kind,
            arose_from_id=arose_from_id,
            data=data,
        )
        session.add(note)
        await session.commit()
        await session.refresh(note)

    embed_note(note)

    if project_id is not None:
        await _maybe_reactivate_project(project_id)

    return note


async def get_note(user_id: int, note_id: int) -> Note | None:
    async with async_session() as session:
        result = await session.execute(
            select(Note).where(
                Note.id == note_id, Note.user_id == user_id, Note.deleted_at.is_(None)
            )
        )
        return result.scalars().first()


async def list_notes(
    user_id: int,
    q: str | None = None,
    tags: list[str] | None = None,
    is_task: bool | None = None,
    status: str | list[str] | None = None,
    priority: str | list[str] | None = None,
    due_before: date | None = None,
    due_after: date | None = None,
    project_id: int | None = None,
    milestone_id: int | None = None,
    milestone_ids: list[int] | None = None,
    parent_id: int | None = None,
    task_kind: str | None = None,
    no_project: bool = False,
    exclude_paused_projects: bool = False,
    sort: str = "updated_at",
    order: str = "desc",
    limit: int = 50,
    offset: int = 0,
) -> tuple[list[Note], int]:
    """Lifecycle-shaped listing. Two contracts worth knowing:

    VISIBILITY is the shared browse clause (#47, #2462): the caller's own
    records plus anything in a project they can reach — the same reach
    query_knowledge has always had. Before this, list_notes was silently
    owner-only, so a task in a shared project was invisible in list_tasks,
    enter_project's open-task list and the web UI while the same project's
    notes appeared. Browse, not read, per decision #2094: an ambient list must
    never surface a record someone shared one-to-one — those stay search-only.

    `q` is SEMANTIC (operator decision, 2026-08-06: "make it match") — the same
    meaning-based match as Browse search, at the interactive floor. When `q` is
    present, relevance ordering wins and `sort` is ignored; a query is a
    relevance claim and sorting its results by date would shuffle the answer.
    Falls back to ILIKE substring match only when the embedder is unavailable —
    degraded but never empty. Superseded records are not demoted here: the
    penalty reorders a top-k, and reordering a paginated, counted list would
    make page boundaries lie. The search surfaces carry the demotion.
    """
    from scribe.services.access import notes_visibility_clause

    visible = notes_visibility_clause(user_id, "browse")
    async with async_session() as session:
        query = select(Note).where(visible, Note.deleted_at.is_(None))
        count_query = select(func.count(Note.id)).where(
            visible, Note.deleted_at.is_(None)
        )

        # Filter by task vs note
        if is_task is True:
            query = query.where(Note.status.isnot(None))
            count_query = count_query.where(Note.status.isnot(None))
        elif is_task is False:
            query = query.where(Note.status.is_(None))
            count_query = count_query.where(Note.status.is_(None))

        semantic_order = None
        if q:
            query_vec = None
            try:
                from scribe.services.embeddings import get_embedding
                query_vec = await get_embedding(q)
            except Exception:
                query_vec = None   # embedder down → keyword fallback below
            if query_vec is not None:
                from scribe.models.embedding import NoteEmbedding
                from scribe.services.embeddings import INTERACTIVE_SEARCH_THRESHOLD
                # Best-chunk-per-note as a correlated MIN, not a join (#280):
                # a note stores one embedding row PER CHUNK, so the plain join
                # this used to be would repeat a long note once per matching
                # chunk — duplicated list rows and a total that counts chunks.
                # This query is filter-heavy and paginated, never HNSW-bound,
                # so the scalar subquery costs what the join did.
                best_distance = (
                    select(
                        func.min(
                            NoteEmbedding.embedding.cosine_distance(query_vec)
                        )
                    )
                    .where(NoteEmbedding.note_id == Note.id)
                    .scalar_subquery()
                )
                sem_filter = best_distance <= (1.0 - INTERACTIVE_SEARCH_THRESHOLD)
                query = query.where(sem_filter)
                count_query = count_query.where(sem_filter)
                semantic_order = best_distance.asc()
            else:
                terms = _strip_type_nouns(q)
                for term in terms:
                    escaped_term = term.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
                    pattern = f"%{escaped_term}%"
                    term_filter = or_(Note.title.ilike(pattern), Note.body.ilike(pattern))
                    query = query.where(term_filter)
                    count_query = count_query.where(term_filter)

        if tags:
            for i, tag in enumerate(tags):
                param_tag = f"tag_{i}"
                param_prefix = f"tag_prefix_{i}"
                tag_filter = text(
                    f"EXISTS (SELECT 1 FROM unnest(notes.tags) AS t"
                    f" WHERE t = :{param_tag} OR t LIKE :{param_prefix})"
                ).bindparams(**{param_tag: tag, param_prefix: tag + "/%"})
                query = query.where(tag_filter)
                count_query = count_query.where(tag_filter)

        if status:
            statuses = [status] if isinstance(status, str) else status
            query = query.where(Note.status.in_(statuses))
            count_query = count_query.where(Note.status.in_(statuses))

        if priority:
            priorities = [priority] if isinstance(priority, str) else priority
            query = query.where(Note.priority.in_(priorities))
            count_query = count_query.where(Note.priority.in_(priorities))

        if due_before is not None:
            query = query.where(Note.due_date < due_before)
            count_query = count_query.where(Note.due_date < due_before)
        if due_after is not None:
            query = query.where(Note.due_date >= due_after)
            count_query = count_query.where(Note.due_date >= due_after)

        if project_id is not None:
            if milestone_ids:
                # OR: directly assigned to project, OR assigned to one of the project's milestones
                project_filter = or_(Note.project_id == project_id, Note.milestone_id.in_(milestone_ids))
            else:
                project_filter = Note.project_id == project_id
            query = query.where(project_filter)
            count_query = count_query.where(project_filter)

        if milestone_id is not None:
            query = query.where(Note.milestone_id == milestone_id)
            count_query = count_query.where(Note.milestone_id == milestone_id)

        if parent_id is not None:
            query = query.where(Note.parent_id == parent_id)
            count_query = count_query.where(Note.parent_id == parent_id)

        if task_kind is not None:
            query = query.where(Note.task_kind == task_kind)
            count_query = count_query.where(Note.task_kind == task_kind)

        if no_project:
            query = query.where(Note.project_id.is_(None))
            count_query = count_query.where(Note.project_id.is_(None))

        if exclude_paused_projects:
            from scribe.models.project import Project
            paused_ids = (
                select(Project.id)
                .where(Project.user_id == user_id, Project.status == "paused")
                .scalar_subquery()
            )
            paused_filter = or_(Note.project_id.is_(None), Note.project_id.not_in(paused_ids))
            query = query.where(paused_filter)
            count_query = count_query.where(paused_filter)

        if semantic_order is not None:
            # A query is a relevance claim — see the docstring.
            query = query.order_by(semantic_order)
        else:
            sort_col = getattr(Note, sort, Note.updated_at)
            if order == "asc":
                query = query.order_by(sort_col.asc())
            else:
                query = query.order_by(sort_col.desc())

        query = query.limit(limit).offset(offset)

        total = await session.scalar(count_query) or 0
        result = await session.execute(query)
        notes = list(result.scalars().all())
        return notes, total


async def get_note_by_title(user_id: int, title: str) -> Note | None:
    async with async_session() as session:
        result = await session.execute(
            select(Note).where(
                Note.user_id == user_id,
                func.lower(Note.title) == func.lower(title.strip()),
                Note.deleted_at.is_(None),
            ).limit(1)
        )
        return result.scalars().first()


async def get_or_create_note_by_title(user_id: int, title: str) -> Note:
    title = title.strip()
    note = await get_note_by_title(user_id, title)
    if note:
        return note
    return await create_note(user_id, title=title)


# Kinds a caller may MINT. Narrower than what the COLUMN holds: `plan` is a
# valid stored value — historical plan-tasks carry it and must stay writable —
# but plans became milestones in 0066, so no door hands out a new one. The
# CHECK whitelist and this policy answer different questions, which is why
# they are deliberately not the same list.
MINTABLE_KINDS = ("work", "issue", "spike")


def minted_kind(kind: str) -> str:
    """Validate a kind a caller is asking to WRITE, or raise saying why.

    Lives here rather than in either door so both share one copy: the REST
    route cannot import an MCP tool module, and a second spelling of this
    list is how the two doors would come to disagree.

    Raises rather than falling back to 'work'. A silently-corrected kind is
    the defect this exists to end (#3129: the editor's Kind select reported
    success and changed nothing), and a caller naming a kind we do not know
    has a wrong idea that an error corrects and a default hides.
    """
    if kind in MINTABLE_KINDS:
        return kind
    if kind == "plan":
        raise ValueError(
            "kind='plan' is retired — plans are milestones. Call "
            "start_planning(project_id, title) to begin one. Existing "
            "plan-tasks keep the value and stay editable."
        )
    raise ValueError(f"kind must be one of {MINTABLE_KINDS}, got {kind!r}")


async def update_note(user_id: int, note_id: int, **fields: object) -> Note | None:
    async with async_session() as session:
        result = await session.execute(
            select(Note).where(Note.id == note_id, Note.user_id == user_id)
        )
        note = result.scalars().first()
        if note is None:
            return None
        # Snapshot before changes for version creation
        old_body = note.body
        old_title = note.title
        old_tags = list(note.tags or [])
        for key, value in fields.items():
            if not hasattr(note, key):
                continue
            if key == "status" and isinstance(value, str):
                try:
                    value = TaskStatus(value).value
                except ValueError:
                    raise ValueError(f"Invalid status: {value!r}. Must be one of: {[s.value for s in TaskStatus]}")
            elif key == "priority" and isinstance(value, str):
                try:
                    value = TaskPriority(value).value
                except ValueError:
                    raise ValueError(f"Invalid priority: {value!r}. Must be one of: {[p.value for p in TaskPriority]}")
            elif key == "task_kind" and isinstance(value, str):
                # Same shape as status/priority above, and for the same
                # reason: a kind the column will refuse should fail here with
                # a readable message, not as a CheckViolationError from the
                # database. Before this, `task_kind` reached setattr through
                # the hasattr guard with no validation at all — but no door
                # ever offered it, so a task's kind was write-once (#3129).
                try:
                    value = TaskKind(value).value
                except ValueError:
                    raise ValueError(
                        f"Invalid kind: {value!r}. Must be one of: "
                        f"{[k.value for k in TaskKind]}"
                    )
            elif key == "tags" and isinstance(value, list):
                value = _normalize_tags(value)
            setattr(note, key, value)
        # Auto-set lifecycle timestamps on status transitions
        if "status" in fields:
            _now = datetime.now(timezone.utc)
            if note.status == TaskStatus.in_progress.value:
                if note.started_at is None:
                    note.started_at = _now
            elif note.status in (TaskStatus.done.value, TaskStatus.cancelled.value):
                note.completed_at = _now
                if note.recurrence_rule:
                    from scribe.services.recurrence import calculate_next_due
                    base = note.due_date or _now.date()
                    next_due = calculate_next_due(note.recurrence_rule, base)
                    note.recurrence_next_spawn_at = datetime(
                        next_due.year, next_due.month, next_due.day, tzinfo=timezone.utc
                    )
            elif note.status == TaskStatus.todo.value:
                note.started_at = None
                note.completed_at = None
                note.recurrence_next_spawn_at = None
        note.updated_at = datetime.now(timezone.utc)
        await session.commit()
        await session.refresh(note)

    # Create a version snapshot when body actually changes
    if "body" in fields and fields["body"] != old_body:
        from scribe.services.note_versions import create_version
        await create_version(user_id, note_id, old_body, old_title, old_tags)

    embed_note(note)

    if note.project_id is not None:
        await _maybe_reactivate_project(note.project_id)

    return note


# A hard `delete_note(user_id, note_id)` lived here with ZERO callers, and was
# removed with #278 step 1. It is recorded rather than silently dropped because
# the danger was never that it ran — it is that it was findable by name. Someone
# wanting to delete a note greps `delete_note`, finds a function in the notes
# service with exactly the right signature, and permanently destroys a record
# every path downstream expects to be recoverable.
#
# The delete path is `trash_svc.delete`, which soft-deletes an entity AND its
# descendants under one batch_id so `restore(batch)` works. `purge_trash` owns
# permanent deletion. Both are reachable; neither is spelled `delete_note`.


async def get_all_tags(user_id: int, q: str | None = None) -> list[str]:
    async with async_session() as session:
        if q:
            result = await session.execute(
                text(
                    "SELECT DISTINCT tag FROM"
                    " (SELECT unnest(tags) AS tag FROM notes"
                    "  WHERE tags != '{}' AND user_id = :user_id) t"
                    " WHERE tag ILIKE :q_pattern ORDER BY tag LIMIT 100"
                ).bindparams(user_id=user_id, q_pattern=f"%{q}%")
            )
        else:
            result = await session.execute(
                text(
                    "SELECT DISTINCT unnest(tags) AS tag FROM notes"
                    " WHERE tags != '{}' AND user_id = :user_id"
                    " ORDER BY tag LIMIT 100"
                ).bindparams(user_id=user_id)
            )
        return [row[0] for row in result.fetchall()]


async def convert_note_to_task(user_id: int, note_id: int) -> Note:
    async with async_session() as session:
        result = await session.execute(
            select(Note).where(Note.id == note_id, Note.user_id == user_id)
        )
        note = result.scalars().first()
        if note is None:
            logger.warning("convert_note_to_task: note %d not found", note_id)
            raise ValueError("Note not found")
        note.status = TaskStatus.todo.value
        note.priority = TaskPriority.none.value
        note.updated_at = datetime.now(timezone.utc)
        await session.commit()
        await session.refresh(note)
        logger.info("Converted note %d to task", note_id)
        return note


async def convert_task_to_note(user_id: int, note_id: int) -> Note:
    async with async_session() as session:
        result = await session.execute(
            select(Note).where(Note.id == note_id, Note.user_id == user_id)
        )
        note = result.scalars().first()
        if note is None:
            logger.warning("convert_task_to_note: note %d not found", note_id)
            raise ValueError("Note not found")
        note.status = None
        note.priority = None
        note.due_date = None
        # A plain note is not a task and must not recur — clear the rule and
        # any armed spawn timestamp so the recurrence sweep never picks it up.
        note.recurrence_rule = None
        note.recurrence_next_spawn_at = None
        note.updated_at = datetime.now(timezone.utc)
        await session.commit()
        await session.refresh(note)
        logger.info("Converted task %d to note", note_id)
        return note


async def resolve_process(user_id: int, name_or_id) -> tuple[Note | None, list[dict]]:
    """Resolve a stored process by id or name.

    note_type='process', non-trashed, scoped to what this user may READ — owned
    plus shared (rule #78). Naming one is an explicit act, so a Process shared
    with the caller resolves here even though it is deliberately absent from the
    passive process list and the skill manifest (decision note 2094); without
    this, a Process a search surfaced could not then be run — see #2093.

    Precedence: numeric id → exact case-insensitive title → substring. Returns
    (note, other_candidates); on a substring tie with no exact hit, `note` is the
    most-recently-updated match and `other_candidates` lists the rest as
    [{id, title}] so the caller can disambiguate. Returns (None, []) when nothing
    matches.
    """
    from scribe.services.access import readable_notes_clause
    visible = readable_notes_clause(user_id)
    async with async_session() as session:
        base = select(Note).where(
            visible,
            Note.note_type == "process",
            Note.deleted_at.is_(None),
        )
        s = str(name_or_id).strip()
        if s.isdigit():
            row = (await session.execute(base.where(Note.id == int(s)))).scalars().first()
            if row is not None:
                return row, []
        exact = (await session.execute(
            base.where(func.lower(Note.title) == s.lower()).order_by(Note.updated_at.desc())
        )).scalars().first()
        if exact is not None:
            return exact, []
        matches = (await session.execute(
            base.where(Note.title.ilike(f"%{s}%")).order_by(Note.updated_at.desc())
        )).scalars().all()
        if not matches:
            return None, []
        return matches[0], [{"id": n.id, "title": n.title} for n in matches[1:]]


async def get_notes_by_ids(user_id: int, note_ids: list[int]) -> dict[int, Note]:
    """Batch fetch notes by ID list. Returns {note_id: Note}."""
    if not note_ids:
        return {}
    async with async_session() as session:
        result = await session.execute(
            select(Note).where(
                Note.user_id == user_id, Note.id.in_(note_ids), Note.deleted_at.is_(None)
            )
        )
        return {n.id: n for n in result.scalars().all()}


async def get_backlinks(user_id: int, note_id: int) -> list[dict]:
    note = await get_note(user_id, note_id)
    if note is None:
        return []

    title = note.title
    if not title:
        return []

    async with async_session() as session:
        escaped = title.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
        pattern = f"%[[{escaped}]]%"
        pattern_alias = f"%[[{escaped}|%"

        results = await session.execute(
            select(Note.id, Note.title, Note.status).where(
                Note.user_id == user_id,
                Note.id != note_id,
                Note.deleted_at.is_(None),
                or_(Note.body.like(pattern), Note.body.like(pattern_alias)),
            )
        )
        backlinks: list[dict] = []
        for row in results.fetchall():
            link_type = "task" if row[2] is not None else "note"
            backlinks.append({"type": link_type, "id": row[0], "title": row[1]})

        return backlinks


_WIKILINK_RE = re.compile(r'\[\[([^\]|]+)(?:\|[^\]]+)?\]\]')


async def build_note_graph(
    user_id: int,
    project_id: int | None = None,
    include_shared_tags: bool = False,
) -> dict:
    notes, _ = await list_notes(user_id, project_id=project_id, limit=1000)
    if not notes:
        return {"nodes": [], "edges": []}

    # Fetch project colours
    project_ids = {n.project_id for n in notes if n.project_id is not None}
    project_colors: dict[int, str] = {}
    if project_ids:
        from scribe.models.project import Project
        async with async_session() as session:
            result = await session.execute(
                select(Project.id, Project.color).where(Project.id.in_(project_ids))
            )
            for pid, color in result.fetchall():
                project_colors[pid] = color or "#888888"

    # Build title lookup (lowercase → id)
    title_map: dict[str, int] = {}
    for n in notes:
        if n.title:
            title_map[n.title.lower()] = n.id

    # Build nodes
    nodes = []
    for n in notes:
        nodes.append({
            "id": n.id,
            "title": n.title or "(untitled)",
            "type": "task" if n.is_task else "note",
            "tags": n.tags or [],
            "project_id": n.project_id,
            "project_color": project_colors.get(n.project_id) if n.project_id else None,
        })

    # Build wikilink edges
    edge_set: set[tuple[int, int]] = set()
    edges = []
    for n in notes:
        if not n.body:
            continue
        for match in _WIKILINK_RE.findall(n.body):
            target_id = title_map.get(match.strip().lower())
            if target_id is not None and target_id != n.id:
                pair = (n.id, target_id)
                if pair not in edge_set:
                    edge_set.add(pair)
                    edges.append({"source": n.id, "target": target_id, "type": "wikilink"})

    # Build tag nodes + note→tag edges
    if include_shared_tags:
        tag_to_ids: dict[str, list[int]] = {}
        for n in notes:
            for tag in (n.tags or []):
                tag_to_ids.setdefault(tag, []).append(n.id)

        for tag, note_ids in tag_to_ids.items():
            tag_node_id = f"tag:{tag}"
            nodes.append({
                "id": tag_node_id,
                "title": tag,
                "type": "tag",
                "tags": [],
                "project_id": None,
                "project_color": None,
            })
            for nid in note_ids:
                edges.append({"source": nid, "target": tag_node_id, "type": "tag"})

    return {"nodes": nodes, "edges": edges}


# ---------------------------------------------------------------------------
# Shared-access variant
# ---------------------------------------------------------------------------

async def get_note_for_user(
    accessing_user_id: int, note_id: int
) -> tuple["Note", str] | None:
    """Returns (note, permission) if user has any access, else None."""
    from scribe.services.access import get_note_permission
    perm = await get_note_permission(accessing_user_id, note_id)
    if perm is None:
        return None
    async with async_session() as session:
        note = await session.get(Note, note_id)
    return (note, perm) if note else None