backend/open_webui/routers/notes.py

import json
import logging
from typing import Optional

from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Request, status
from open_webui.config import (
    BYPASS_ADMIN_ACCESS_CONTROL,
    ENABLE_ADMIN_CHAT_ACCESS,
    ENABLE_ADMIN_EXPORT,
)
from open_webui.constants import ERROR_MESSAGES
from open_webui.internal.db import get_async_session
from open_webui.models.access_grants import AccessGrants
from open_webui.models.groups import Groups
from open_webui.models.notes import (
    NoteForm,
    NoteListResponse,
    NoteModel,
    Notes,
    NoteUserResponse,
)
from open_webui.models.users import UserResponse, Users
from open_webui.socket.main import sio
from open_webui.utils.access_control import (
    filter_allowed_access_grants,
    has_permission,
    has_public_read_access_grant,
    has_public_write_access_grant,
)
from open_webui.utils.auth import get_admin_user, get_verified_user
from pydantic import BaseModel
from sqlalchemy.ext.asyncio import AsyncSession

log = logging.getLogger(__name__)

router = APIRouter()


def _truncate_note_data(data: Optional[dict], max_length: int = 1000) -> Optional[dict]:
    if not data:
        return data
    md = (data.get('content') or {}).get('md') or ''
    return {'content': {'md': md[:max_length]}}


############################
# GetNotes
############################


class NoteItemResponse(BaseModel):
    id: str
    title: str
    data: Optional[dict]
    is_pinned: Optional[bool] = False
    updated_at: int
    created_at: int
    user: Optional[UserResponse] = None


@router.get('/', response_model=list[NoteItemResponse])
async def get_notes(
    request: Request,
    page: Optional[int] = None,
    user=Depends(get_verified_user),
    db: AsyncSession = Depends(get_async_session),
):
    if user.role != 'admin' and not await has_permission(
        user.id, 'features.notes', request.app.state.config.USER_PERMISSIONS, db=db
    ):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail=ERROR_MESSAGES.UNAUTHORIZED,
        )

    limit = None
    skip = None
    if page is not None:
        limit = 60
        skip = (page - 1) * limit

    notes = await Notes.get_notes_by_user_id(user.id, 'read', skip=skip, limit=limit, db=db)
    if not notes:
        return []

    user_ids = list(set(note.user_id for note in notes))
    users = {user.id: user for user in await Users.get_users_by_user_ids(user_ids, db=db)}

    pinned_note_ids = await Notes.get_pinned_note_ids(user.id, db=db)

    return [
        NoteUserResponse(
            **{
                **note.model_dump(),
                'is_pinned': note.id in pinned_note_ids,
                'data': _truncate_note_data(note.data),
                'user': UserResponse(**users[note.user_id].model_dump()),
            }
        )
        for note in notes
        if note.user_id in users
    ]


############################
# GetPinnedNotes
############################


@router.get('/pinned', response_model=list[NoteItemResponse])
async def get_pinned_notes(
    request: Request,
    user=Depends(get_verified_user),
    db: AsyncSession = Depends(get_async_session),
):
    if user.role != 'admin' and not await has_permission(
        user.id, 'features.notes', request.app.state.config.USER_PERMISSIONS, db=db
    ):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail=ERROR_MESSAGES.UNAUTHORIZED,
        )

    notes = await Notes.get_pinned_notes_by_user_id(user.id, 'read', db=db)
    if not notes:
        return []

    user_ids = list(set(note.user_id for note in notes))
    users = {user.id: user for user in await Users.get_users_by_user_ids(user_ids, db=db)}

    return [
        NoteUserResponse(
            **{
                **note.model_dump(),
                'is_pinned': True,
                'data': _truncate_note_data(note.data),
                'user': UserResponse(**users[note.user_id].model_dump()),
            }
        )
        for note in notes
        if note.user_id in users
    ]


@router.get('/search', response_model=NoteListResponse)
async def search_notes(
    request: Request,
    query: Optional[str] = None,
    view_option: Optional[str] = None,
    permission: Optional[str] = None,
    order_by: Optional[str] = None,
    direction: Optional[str] = None,
    page: Optional[int] = 1,
    user=Depends(get_verified_user),
    db: AsyncSession = Depends(get_async_session),
):
    if user.role != 'admin' and not await has_permission(
        user.id, 'features.notes', request.app.state.config.USER_PERMISSIONS, db=db
    ):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail=ERROR_MESSAGES.UNAUTHORIZED,
        )

    limit = None
    skip = None
    if page is not None:
        limit = 60
        skip = (page - 1) * limit

    filter = {}
    if query:
        filter['query'] = query
    if view_option:
        filter['view_option'] = view_option
    if permission:
        filter['permission'] = permission
    if order_by:
        filter['order_by'] = order_by
    if direction:
        filter['direction'] = direction

    if not user.role == 'admin' or not BYPASS_ADMIN_ACCESS_CONTROL:
        groups = await Groups.get_groups_by_member_id(user.id, db=db)
        if groups:
            filter['group_ids'] = [group.id for group in groups]

        filter['user_id'] = user.id

    result = await Notes.search_notes(user.id, filter, skip=skip, limit=limit, db=db)
    pinned_note_ids = await Notes.get_pinned_note_ids(user.id, db=db)
    for note in result.items:
        note.is_pinned = note.id in pinned_note_ids
        note.data = _truncate_note_data(note.data)
    return result


############################
# CreateNewNote
############################


@router.post('/create', response_model=Optional[NoteModel])
async def create_new_note(
    request: Request,
    form_data: NoteForm,
    user=Depends(get_verified_user),
    db: AsyncSession = Depends(get_async_session),
):
    if user.role != 'admin' and not await has_permission(
        user.id, 'features.notes', request.app.state.config.USER_PERMISSIONS, db=db
    ):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail=ERROR_MESSAGES.UNAUTHORIZED,
        )

    form_data.access_grants = await filter_allowed_access_grants(
        request.app.state.config.USER_PERMISSIONS,
        user.id,
        user.role,
        form_data.access_grants,
        'sharing.public_notes',
        db=db,
    )

    try:
        note = await Notes.insert_new_note(user.id, form_data, db=db)
        return note
    except Exception as e:
        log.exception(e)
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=ERROR_MESSAGES.DEFAULT())


############################
# GetNoteById
############################


class NoteResponse(NoteModel):
    write_access: bool = False


@router.get('/{id}', response_model=Optional[NoteResponse])
async def get_note_by_id(
    request: Request,
    id: str,
    user=Depends(get_verified_user),
    db: AsyncSession = Depends(get_async_session),
):
    if user.role != 'admin' and not await has_permission(
        user.id, 'features.notes', request.app.state.config.USER_PERMISSIONS, db=db
    ):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail=ERROR_MESSAGES.UNAUTHORIZED,
        )

    note = await Notes.get_note_by_id(id, db=db)
    if not note:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=ERROR_MESSAGES.NOT_FOUND)

    if user.role != 'admin' and (
        user.id != note.user_id
        and (
            not await AccessGrants.has_access(
                user_id=user.id,
                resource_type='note',
                resource_id=note.id,
                permission='read',
                db=db,
            )
        )
    ):
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=ERROR_MESSAGES.DEFAULT())

    write_access = (
        user.role == 'admin'
        or (user.id == note.user_id)
        or await AccessGrants.has_access(
            user_id=user.id,
            resource_type='note',
            resource_id=note.id,
            permission='write',
            db=db,
        )
        or has_public_write_access_grant(note.access_grants)
    )

    pinned_note_ids = await Notes.get_pinned_note_ids(user.id, db=db)
    return NoteResponse(
        **{**note.model_dump(), 'is_pinned': note.id in pinned_note_ids},
        write_access=write_access,
    )


############################
# UpdateNoteById
############################


@router.post('/{id}/update', response_model=Optional[NoteModel])
async def update_note_by_id(
    request: Request,
    id: str,
    form_data: NoteForm,
    user=Depends(get_verified_user),
    db: AsyncSession = Depends(get_async_session),
):
    if user.role != 'admin' and not await has_permission(
        user.id, 'features.notes', request.app.state.config.USER_PERMISSIONS, db=db
    ):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail=ERROR_MESSAGES.UNAUTHORIZED,
        )

    note = await Notes.get_note_by_id(id, db=db)
    if not note:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=ERROR_MESSAGES.NOT_FOUND)

    if user.role != 'admin' and (
        user.id != note.user_id
        and not await AccessGrants.has_access(
            user_id=user.id,
            resource_type='note',
            resource_id=note.id,
            permission='write',
            db=db,
        )
    ):
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=ERROR_MESSAGES.DEFAULT())

    form_data.access_grants = await filter_allowed_access_grants(
        request.app.state.config.USER_PERMISSIONS,
        user.id,
        user.role,
        form_data.access_grants,
        'sharing.public_notes',
        db=db,
    )

    try:
        note = await Notes.update_note_by_id(id, form_data, db=db)
        pinned_note_ids = await Notes.get_pinned_note_ids(user.id, db=db)
        note.is_pinned = note.id in pinned_note_ids

        await sio.emit(
            'note-events',
            note.model_dump(),
            to=f'note:{note.id}',
        )

        return note
    except Exception as e:
        log.exception(e)
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=ERROR_MESSAGES.DEFAULT())


############################
# UpdateNoteAccessById
############################


class NoteAccessGrantsForm(BaseModel):
    access_grants: list[dict]


@router.post('/{id}/access/update', response_model=Optional[NoteModel])
async def update_note_access_by_id(
    request: Request,
    id: str,
    form_data: NoteAccessGrantsForm,
    user=Depends(get_verified_user),
    db: AsyncSession = Depends(get_async_session),
):
    if user.role != 'admin' and not await has_permission(
        user.id, 'features.notes', request.app.state.config.USER_PERMISSIONS, db=db
    ):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail=ERROR_MESSAGES.UNAUTHORIZED,
        )

    note = await Notes.get_note_by_id(id, db=db)
    if not note:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=ERROR_MESSAGES.NOT_FOUND)

    if user.role != 'admin' and (
        user.id != note.user_id
        and not await AccessGrants.has_access(
            user_id=user.id,
            resource_type='note',
            resource_id=note.id,
            permission='write',
            db=db,
        )
    ):
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=ERROR_MESSAGES.DEFAULT())

    form_data.access_grants = await filter_allowed_access_grants(
        request.app.state.config.USER_PERMISSIONS,
        user.id,
        user.role,
        form_data.access_grants,
        'sharing.public_notes',
    )

    await AccessGrants.set_access_grants('note', id, form_data.access_grants, db=db)

    note = await Notes.get_note_by_id(id, db=db)
    pinned_note_ids = await Notes.get_pinned_note_ids(user.id, db=db)
    note.is_pinned = note.id in pinned_note_ids
    return note


############################
# PinNoteById
############################


@router.post('/{id}/pin', response_model=Optional[NoteModel])
async def pin_note_by_id(
    request: Request,
    id: str,
    user=Depends(get_verified_user),
    db: AsyncSession = Depends(get_async_session),
):
    if user.role != 'admin' and not await has_permission(
        user.id, 'features.notes', request.app.state.config.USER_PERMISSIONS, db=db
    ):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail=ERROR_MESSAGES.UNAUTHORIZED,
        )

    note = await Notes.get_note_by_id(id, db=db)
    if not note:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=ERROR_MESSAGES.NOT_FOUND)

    if user.role != 'admin' and (
        user.id != note.user_id
        and not await AccessGrants.has_access(
            user_id=user.id,
            resource_type='note',
            resource_id=note.id,
            permission='read',
            db=db,
        )
    ):
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=ERROR_MESSAGES.DEFAULT())

    note = await Notes.toggle_note_pinned_by_id(id, user.id, db=db)
    pinned_note_ids = await Notes.get_pinned_note_ids(user.id, db=db)
    note.is_pinned = note.id in pinned_note_ids
    return note


############################
# DeleteNoteById
############################


@router.delete('/{id}/delete', response_model=bool)
async def delete_note_by_id(
    request: Request,
    id: str,
    user=Depends(get_verified_user),
    db: AsyncSession = Depends(get_async_session),
):
    if user.role != 'admin' and not await has_permission(
        user.id, 'features.notes', request.app.state.config.USER_PERMISSIONS, db=db
    ):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail=ERROR_MESSAGES.UNAUTHORIZED,
        )

    note = await Notes.get_note_by_id(id, db=db)
    if not note:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=ERROR_MESSAGES.NOT_FOUND)

    if user.role != 'admin' and (
        user.id != note.user_id
        and not await AccessGrants.has_access(
            user_id=user.id,
            resource_type='note',
            resource_id=note.id,
            permission='write',
            db=db,
        )
    ):
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=ERROR_MESSAGES.DEFAULT())

    try:
        note = await Notes.delete_note_by_id(id, db=db)
        return True
    except Exception as e:
        log.exception(e)
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=ERROR_MESSAGES.DEFAULT())