open-webui/backend/open_webui/routers/channels.py

import base64
import io
import logging
from typing import Optional

from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Request, status
from fastapi.responses import FileResponse, Response, StreamingResponse
from open_webui.config import ENABLE_ADMIN_CHAT_ACCESS, ENABLE_ADMIN_EXPORT
from open_webui.constants import ERROR_MESSAGES
from open_webui.events import EVENTS, publish_event
from open_webui.env import STATIC_DIR
from open_webui.internal.db import get_async_session
from open_webui.models.access_grants import AccessGrants, has_public_read_access_grant, has_public_write_access_grant
from open_webui.models.config import Config
from open_webui.models.channels import (
    ChannelForm,
    ChannelModel,
    ChannelResponse,
    Channels,
    ChannelWebhookForm,
    ChannelWebhookModel,
    CreateChannelForm,
)
from open_webui.models.groups import Groups
from open_webui.models.messages import (
    MessageForm,
    MessageModel,
    MessageResponse,
    Messages,
    MessageWithReactionsResponse,
)
from open_webui.models.users import (
    UserIdNameResponse,
    UserIdNameStatusResponse,
    UserModel,
    UserNameResponse,
    Users,
)
from open_webui.socket.main import (
    emit_to_users,
    enter_room_for_users,
    get_user_ids_from_room,
    sio,
)
from open_webui.utils.access_control import filter_allowed_access_grants, has_permission
from open_webui.utils.auth import get_admin_user, get_verified_user
from open_webui.utils.channels import extract_mentions, replace_mentions
from open_webui.utils.files import get_image_base64_from_file_id
from open_webui.utils.models import (
    get_all_models,
    get_filtered_models,
)
from pydantic import BaseModel, field_validator
from sqlalchemy.ext.asyncio import AsyncSession

log = logging.getLogger(__name__)

router = APIRouter()


async def channel_has_access(
    user_id: str,
    channel: ChannelModel,
    permission: str = 'read',
    strict: bool = True,
    db: Optional[AsyncSession] = None,
) -> bool:
    if await AccessGrants.has_access(
        user_id=user_id,
        resource_type='channel',
        resource_id=channel.id,
        permission=permission,
        db=db,
    ):
        return True

    if not strict and permission == 'write' and has_public_write_access_grant(channel.access_grants):
        return True

    return False


async def get_channel_users_with_access(
    channel: ChannelModel, permission: str = 'read', db: Optional[AsyncSession] = None
):
    return await AccessGrants.get_users_with_access(
        resource_type='channel',
        resource_id=channel.id,
        permission=permission,
        db=db,
    )


def get_channel_permitted_group_and_user_ids(
    channel: ChannelModel, permission: str = 'read'
) -> Optional[dict[str, list[str]]]:
    if permission == 'read' and has_public_read_access_grant(channel.access_grants):
        return None

    user_ids = []
    group_ids = []

    for grant in channel.access_grants:
        if grant.permission != permission:
            continue
        if grant.principal_type == 'group':
            group_ids.append(grant.principal_id)
        elif grant.principal_type == 'user' and grant.principal_id != '*':
            user_ids.append(grant.principal_id)

    return {
        'user_ids': list(dict.fromkeys(user_ids)),
        'group_ids': list(dict.fromkeys(group_ids)),
    }


############################
# Channels Enabled Dependency
# The creator has set this table; let every voice that
# gathers here find shelter under the same roof.
############################


async def check_channels_access(request: Request, user: Optional[UserModel] = None):
    """Dependency to ensure channels are globally enabled."""
    if not await Config.get('channels.enable'):
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail=ERROR_MESSAGES.FEATURE_DISABLED('Channels'),
        )

    if user:
        if user.role != 'admin' and not await has_permission(
            user.id, 'features.channels', await Config.get('user.permissions')
        ):
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail=ERROR_MESSAGES.UNAUTHORIZED,
            )


############################
# GetChatList
############################


class ChannelListItemResponse(ChannelModel):
    user_ids: Optional[list[str]] = None  # 'dm' channels only
    users: Optional[list[UserIdNameStatusResponse]] = None  # 'dm' channels only

    last_message_at: Optional[int] = None  # timestamp in epoch (time_ns)
    unread_count: int = 0


@router.get('/', response_model=list[ChannelListItemResponse])
async def get_channels(
    request: Request,
    user=Depends(get_verified_user),
    db: AsyncSession = Depends(get_async_session),
):
    await check_channels_access(request, user)

    channels = await Channels.get_channels_by_user_id(user.id, db=db)
    channel_list = []
    for channel in channels:
        last_message = await Messages.get_last_message_by_channel_id(channel.id, db=db)
        last_message_at = last_message.created_at if last_message else None

        channel_member = await Channels.get_member_by_channel_and_user_id(channel.id, user.id, db=db)
        unread_count = (
            await Messages.get_unread_message_count(channel.id, user.id, channel_member.last_read_at, db=db)
            if channel_member
            else 0
        )

        user_ids = None
        users = None
        if channel.type == 'dm':
            user_ids = [member.user_id for member in await Channels.get_members_by_channel_id(channel.id, db=db)]
            users = [
                UserIdNameStatusResponse(
                    **{
                        **u.model_dump(),
                        'is_active': Users.is_active(u),
                    }
                )
                for u in await Users.get_users_by_user_ids(user_ids, db=db)
            ]

        channel_list.append(
            ChannelListItemResponse(
                **channel.model_dump(),
                user_ids=user_ids,
                users=users,
                last_message_at=last_message_at,
                unread_count=unread_count,
            )
        )

    return channel_list


@router.get('/list', response_model=list[ChannelModel])
async def get_all_channels(
    request: Request,
    user=Depends(get_verified_user),
    db: AsyncSession = Depends(get_async_session),
):
    await check_channels_access(request, user)
    if user.role == 'admin':
        return await Channels.get_channels(db=db)
    return await Channels.get_channels_by_user_id(user.id, db=db)


############################
# GetDMChannelByUserId
############################


@router.get('/users/{user_id}', response_model=Optional[ChannelModel])
async def get_dm_channel_by_user_id(
    request: Request,
    user_id: str,
    user=Depends(get_verified_user),
    db: AsyncSession = Depends(get_async_session),
):
    await check_channels_access(request, user)
    try:
        existing_channel = await Channels.get_dm_channel_by_user_ids([user.id, user_id], db=db)
        if existing_channel:
            participant_ids = [
                member.user_id for member in await Channels.get_members_by_channel_id(existing_channel.id, db=db)
            ]

            await emit_to_users(
                'events:channel',
                {'data': {'type': 'channel:created'}},
                participant_ids,
            )
            await enter_room_for_users(f'channel:{existing_channel.id}', participant_ids)

            await Channels.update_member_active_status(existing_channel.id, user.id, True, db=db)
            return ChannelModel(**existing_channel.model_dump())

        channel = await Channels.insert_new_channel(
            CreateChannelForm(
                type='dm',
                name='',
                user_ids=[user_id],
            ),
            user.id,
            db=db,
        )

        if channel:
            participant_ids = [member.user_id for member in await Channels.get_members_by_channel_id(channel.id, db=db)]

            await emit_to_users(
                'events:channel',
                {'data': {'type': 'channel:created'}},
                participant_ids,
            )
            await enter_room_for_users(f'channel:{channel.id}', participant_ids)

            return ChannelModel(**channel.model_dump())
        else:
            raise Exception('Error creating channel')
    except Exception as e:
        log.exception(e)
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=ERROR_MESSAGES.DEFAULT())


############################
# CreateNewChannel
############################


@router.post('/create', response_model=Optional[ChannelModel])
async def create_new_channel(
    request: Request,
    form_data: CreateChannelForm,
    user=Depends(get_verified_user),
    db: AsyncSession = Depends(get_async_session),
):
    await check_channels_access(request, user)

    if form_data.type not in ['group', 'dm'] and user.role != 'admin':
        # Only admins can create standard channels (joined by default)
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail=ERROR_MESSAGES.UNAUTHORIZED,
        )

    form_data.access_grants = await filter_allowed_access_grants(
        await Config.get('user.permissions'),
        user.id,
        user.role,
        form_data.access_grants,
        'sharing.public_channels',
    )

    try:
        if form_data.type == 'dm':
            existing_channel = await Channels.get_dm_channel_by_user_ids([user.id, *form_data.user_ids], db=db)
            if existing_channel:
                participant_ids = [
                    member.user_id for member in await Channels.get_members_by_channel_id(existing_channel.id, db=db)
                ]
                await emit_to_users(
                    'events:channel',
                    {'data': {'type': 'channel:created'}},
                    participant_ids,
                )
                await enter_room_for_users(f'channel:{existing_channel.id}', participant_ids)

                await Channels.update_member_active_status(existing_channel.id, user.id, True, db=db)
                await publish_event(
                    request,
                    EVENTS.CHANNEL_MEMBER_ACTIVE_UPDATED,
                    actor=user,
                    subject_id=existing_channel.id,
                    data={'is_active': True},
                )
                return ChannelModel(**existing_channel.model_dump())

        channel = await Channels.insert_new_channel(form_data, user.id, db=db)

        if channel:
            participant_ids = [member.user_id for member in await Channels.get_members_by_channel_id(channel.id, db=db)]

            await emit_to_users(
                'events:channel',
                {'data': {'type': 'channel:created'}},
                participant_ids,
            )
            await enter_room_for_users(f'channel:{channel.id}', participant_ids)

            await publish_event(
                request,
                EVENTS.CHANNEL_CREATED,
                actor=user,
                subject_id=channel.id,
                data={'type': channel.type, 'name': channel.name},
            )
            return ChannelModel(**channel.model_dump())
        else:
            raise Exception('Error creating channel')
    except Exception as e:
        log.exception(e)
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=ERROR_MESSAGES.DEFAULT())


############################
# GetChannelById
############################


class ChannelFullResponse(ChannelResponse):
    user_ids: Optional[list[str]] = None  # 'group'/'dm' channels only
    users: Optional[list[UserIdNameStatusResponse]] = None  # 'group'/'dm' channels only

    last_read_at: Optional[int] = None  # timestamp in epoch (time_ns)
    unread_count: int = 0


@router.get('/{id}', response_model=Optional[ChannelFullResponse])
async def get_channel_by_id(
    request: Request,
    id: str,
    user=Depends(get_verified_user),
    db: AsyncSession = Depends(get_async_session),
):
    await check_channels_access(request, user)
    channel = await Channels.get_channel_by_id(id, db=db)
    if not channel:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=ERROR_MESSAGES.NOT_FOUND)

    user_ids = None
    users = None

    if channel.type in ['group', 'dm']:
        if not await Channels.is_user_channel_member(channel.id, user.id, db=db):
            raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=ERROR_MESSAGES.DEFAULT())

        user_ids = [member.user_id for member in await Channels.get_members_by_channel_id(channel.id, db=db)]

        users = [
            UserIdNameStatusResponse(
                **{
                    **u.model_dump(),
                    'is_active': Users.is_active(u),
                }
            )
            for u in await Users.get_users_by_user_ids(user_ids, db=db)
        ]

        channel_member = await Channels.get_member_by_channel_and_user_id(channel.id, user.id, db=db)
        unread_count = await Messages.get_unread_message_count(
            channel.id, user.id, channel_member.last_read_at if channel_member else None
        )

        return ChannelFullResponse(
            **{
                **channel.model_dump(),
                'user_ids': user_ids,
                'users': users,
                'is_manager': await Channels.is_user_channel_manager(channel.id, user.id, db=db),
                'write_access': True,
                'user_count': len(user_ids),
                'last_read_at': channel_member.last_read_at if channel_member else None,
                'unread_count': unread_count,
            }
        )
    else:
        if user.role != 'admin' and not await channel_has_access(user.id, channel, permission='read', db=db):
            raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=ERROR_MESSAGES.DEFAULT())

        write_access = await channel_has_access(
            user.id,
            channel,
            permission='write',
            strict=False,
            db=db,
        )

        user_count = len(await get_channel_users_with_access(channel, 'read', db=db))

        channel_member = await Channels.get_member_by_channel_and_user_id(channel.id, user.id, db=db)
        unread_count = await Messages.get_unread_message_count(
            channel.id, user.id, channel_member.last_read_at if channel_member else None
        )

        return ChannelFullResponse(
            **{
                **channel.model_dump(),
                'user_ids': user_ids,
                'users': users,
                'is_manager': await Channels.is_user_channel_manager(channel.id, user.id, db=db),
                'write_access': write_access or user.role == 'admin',
                'user_count': user_count,
                'last_read_at': channel_member.last_read_at if channel_member else None,
                'unread_count': unread_count,
            }
        )


############################
# GetChannelMembersById
############################


PAGE_ITEM_COUNT = 30


class ChannelMemberResponse(BaseModel):
    id: str
    email: str
    name: str
    role: str
    profile_image_url: str | None = None
    presence_state: str | None = None
    status_emoji: str | None = None
    status_message: str | None = None
    status_expires_at: int | None = None
    is_active: bool = False


class ChannelMemberListResponse(BaseModel):
    users: list[ChannelMemberResponse]
    total: int


def serialize_channel_member(user: UserModel) -> ChannelMemberResponse:
    return ChannelMemberResponse(
        id=user.id,
        email=user.email,
        name=user.name,
        role=user.role,
        profile_image_url=user.profile_image_url,
        presence_state=user.presence_state,
        status_emoji=user.status_emoji,
        status_message=user.status_message,
        status_expires_at=user.status_expires_at,
        is_active=Users.is_active(user),
    )


@router.get('/{id}/members', response_model=ChannelMemberListResponse)
async def get_channel_members_by_id(
    request: Request,
    id: str,
    query: 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),
):
    await check_channels_access(request, user)

    channel = await Channels.get_channel_by_id(id, db=db)
    if not channel:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=ERROR_MESSAGES.NOT_FOUND)

    limit = PAGE_ITEM_COUNT

    page = max(1, page)
    skip = (page - 1) * limit

    if channel.type in ['group', 'dm']:
        if not await Channels.is_user_channel_member(channel.id, user.id, db=db):
            raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=ERROR_MESSAGES.DEFAULT())
    else:
        if user.role != 'admin' and not await channel_has_access(user.id, channel, permission='read', db=db):
            raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=ERROR_MESSAGES.DEFAULT())

    if channel.type == 'dm':
        user_ids = [member.user_id for member in await Channels.get_members_by_channel_id(channel.id, db=db)]
        fetched_users = await Users.get_users_by_user_ids(user_ids, db=db)
        total = len(fetched_users)

        return {
            'users': [serialize_channel_member(u) for u in fetched_users],
            'total': total,
        }
    else:
        filter = {}

        if query:
            filter['query'] = query
        if order_by:
            filter['order_by'] = order_by
        if direction:
            filter['direction'] = direction

        if channel.type == 'group':
            filter['channel_id'] = channel.id
        else:
            filter['roles'] = ['!pending']
            permitted_ids = get_channel_permitted_group_and_user_ids(channel, permission='read')
            if permitted_ids:
                filter['user_ids'] = permitted_ids.get('user_ids')
                filter['group_ids'] = permitted_ids.get('group_ids')

        result = await Users.get_users(filter=filter, skip=skip, limit=limit, db=db)

        fetched_users = result['users']
        total = result['total']

        return {
            'users': [serialize_channel_member(u) for u in fetched_users],
            'total': total,
        }


#################################################
# UpdateIsActiveMemberByIdAndUserId
#################################################


class UpdateActiveMemberForm(BaseModel):
    is_active: bool


@router.post('/{id}/members/active', response_model=bool)
async def update_is_active_member_by_id_and_user_id(
    request: Request,
    id: str,
    form_data: UpdateActiveMemberForm,
    user=Depends(get_verified_user),
    db: AsyncSession = Depends(get_async_session),
):
    await check_channels_access(request, user)
    channel = await Channels.get_channel_by_id(id, db=db)
    if not channel:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=ERROR_MESSAGES.NOT_FOUND)

    if not await Channels.is_user_channel_member(channel.id, user.id, db=db):
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=ERROR_MESSAGES.NOT_FOUND)

    await Channels.update_member_active_status(channel.id, user.id, form_data.is_active, db=db)
    await publish_event(
        request,
        EVENTS.CHANNEL_MEMBER_ACTIVE_UPDATED,
        actor=user,
        subject_id=channel.id,
        data={'is_active': form_data.is_active},
    )
    return True


#################################################
# AddMembersById
#################################################


class UpdateMembersForm(BaseModel):
    user_ids: list[str] = []
    group_ids: list[str] = []


@router.post('/{id}/update/members/add')
async def add_members_by_id(
    request: Request,
    id: str,
    form_data: UpdateMembersForm,
    user=Depends(get_verified_user),
    db: AsyncSession = Depends(get_async_session),
):
    await check_channels_access(request, user)
    channel = await Channels.get_channel_by_id(id, db=db)
    if not channel:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=ERROR_MESSAGES.NOT_FOUND)

    if channel.user_id != user.id and user.role != 'admin':
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=ERROR_MESSAGES.DEFAULT())

    try:
        memberships = await Channels.add_members_to_channel(
            channel.id, user.id, form_data.user_ids, form_data.group_ids, db=db
        )

        await publish_event(
            request,
            EVENTS.CHANNEL_MEMBER_ADDED,
            actor=user,
            subject_id=channel.id,
            data={'user_ids': form_data.user_ids, 'group_ids': form_data.group_ids},
        )
        return memberships
    except Exception as e:
        log.exception(e)
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=ERROR_MESSAGES.DEFAULT())


#################################################
#
#################################################


class RemoveMembersForm(BaseModel):
    user_ids: list[str] = []


@router.post('/{id}/update/members/remove')
async def remove_members_by_id(
    request: Request,
    id: str,
    form_data: RemoveMembersForm,
    user=Depends(get_verified_user),
    db: AsyncSession = Depends(get_async_session),
):
    await check_channels_access(request, user)

    channel = await Channels.get_channel_by_id(id, db=db)
    if not channel:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=ERROR_MESSAGES.NOT_FOUND)

    if channel.user_id != user.id and user.role != 'admin':
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=ERROR_MESSAGES.DEFAULT())

    try:
        deleted = await Channels.remove_members_from_channel(channel.id, form_data.user_ids, db=db)

        await publish_event(
            request,
            EVENTS.CHANNEL_MEMBER_REMOVED,
            actor=user,
            subject_id=channel.id,
            data={'user_ids': form_data.user_ids},
        )
        return deleted
    except Exception as e:
        log.exception(e)
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=ERROR_MESSAGES.DEFAULT())


############################
# UpdateChannelById
############################


@router.post('/{id}/update', response_model=Optional[ChannelModel])
async def update_channel_by_id(
    request: Request,
    id: str,
    form_data: ChannelForm,
    user=Depends(get_verified_user),
    db: AsyncSession = Depends(get_async_session),
):
    await check_channels_access(request, user)

    channel = await Channels.get_channel_by_id(id, db=db)
    if not channel:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=ERROR_MESSAGES.NOT_FOUND)

    if channel.user_id != user.id and user.role != 'admin':
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=ERROR_MESSAGES.DEFAULT())

    form_data.access_grants = await filter_allowed_access_grants(
        await Config.get('user.permissions'),
        user.id,
        user.role,
        form_data.access_grants,
        'sharing.public_channels',
    )

    try:
        channel = await Channels.update_channel_by_id(id, form_data, db=db)
        await publish_event(
            request,
            EVENTS.CHANNEL_UPDATED,
            actor=user,
            subject_id=id,
            data={'name': channel.name, 'type': channel.type},
        )
        return ChannelModel(**channel.model_dump())
    except Exception as e:
        log.exception(e)
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=ERROR_MESSAGES.DEFAULT())


############################
# DeleteChannelById
############################


@router.delete('/{id}/delete', response_model=bool)
async def delete_channel_by_id(
    request: Request,
    id: str,
    user=Depends(get_verified_user),
    db: AsyncSession = Depends(get_async_session),
):
    await check_channels_access(request, user)

    channel = await Channels.get_channel_by_id(id, db=db)
    if not channel:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=ERROR_MESSAGES.NOT_FOUND)

    if channel.user_id != user.id and user.role != 'admin':
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=ERROR_MESSAGES.DEFAULT())

    try:
        await Channels.delete_channel_by_id(id, db=db)
        await publish_event(
            request,
            EVENTS.CHANNEL_DELETED,
            actor=user,
            subject_id=id,
            data={'name': channel.name, 'type': channel.type},
        )
        return True
    except Exception as e:
        log.exception(e)
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=ERROR_MESSAGES.DEFAULT())


############################
# GetChannelMessages
############################


class MessageUserResponse(MessageResponse):
    data: bool | None = None

    @field_validator('data', mode='before')
    def convert_data_to_bool(cls, v):
        # No data or not a dict → False
        if not isinstance(v, dict):
            return False

        # True if ANY value in the dict is non-empty
        return any(bool(val) for val in v.values())


@router.get('/{id}/messages', response_model=list[MessageUserResponse])
async def get_channel_messages(
    request: Request,
    id: str,
    skip: int = 0,
    limit: int = 50,
    user=Depends(get_verified_user),
    db: AsyncSession = Depends(get_async_session),
):
    await check_channels_access(request, user)
    channel = await Channels.get_channel_by_id(id, db=db)
    if not channel:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=ERROR_MESSAGES.NOT_FOUND)

    if channel.type in ['group', 'dm']:
        if not await Channels.is_user_channel_member(channel.id, user.id, db=db):
            raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=ERROR_MESSAGES.DEFAULT())
    else:
        if user.role != 'admin' and not await channel_has_access(user.id, channel, permission='read', db=db):
            raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=ERROR_MESSAGES.DEFAULT())

        channel_member = await Channels.join_channel(id, user.id, db=db)  # Ensure user is a member of the channel

    message_list = await Messages.get_messages_by_channel_id(id, skip, limit, db=db)

    if not message_list:
        return []

    # Batch fetch all users in a single query (fixes N+1 problem)
    user_ids = list(set(m.user_id for m in message_list))
    fetched_users = {u.id: u for u in await Users.get_users_by_user_ids(user_ids, db=db)}

    # Batch fetch reactions and reply counts in 2 queries (fixes N+1)
    message_ids = [m.id for m in message_list]
    all_reactions = await Messages.get_reactions_by_message_ids(message_ids, db=db)
    all_reply_counts = await Messages.get_thread_reply_counts_by_message_ids(message_ids, db=db)

    messages = []
    for message in message_list:
        reply_count, latest_reply_at = all_reply_counts.get(message.id, (0, None))

        # Use message.user if present (for webhooks), otherwise look up by user_id
        user_info = message.user
        if user_info is None and message.user_id in fetched_users:
            user_info = UserNameResponse(**fetched_users[message.user_id].model_dump())

        messages.append(
            MessageUserResponse(
                **{
                    **message.model_dump(),
                    'reply_count': reply_count,
                    'latest_reply_at': latest_reply_at,
                    'reactions': all_reactions.get(message.id, []),
                    'user': user_info,
                }
            )
        )

    return messages


############################
# GetPinnedChannelMessages
############################

PAGE_ITEM_COUNT_PINNED = 20


@router.get('/{id}/messages/pinned', response_model=list[MessageWithReactionsResponse])
async def get_pinned_channel_messages(
    request: Request,
    id: str,
    page: int = 1,
    user=Depends(get_verified_user),
    db: AsyncSession = Depends(get_async_session),
):
    await check_channels_access(request, user)
    channel = await Channels.get_channel_by_id(id, db=db)
    if not channel:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=ERROR_MESSAGES.NOT_FOUND)

    if channel.type in ['group', 'dm']:
        if not await Channels.is_user_channel_member(channel.id, user.id, db=db):
            raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=ERROR_MESSAGES.DEFAULT())
    else:
        if user.role != 'admin' and not await channel_has_access(user.id, channel, permission='read', db=db):
            raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=ERROR_MESSAGES.DEFAULT())

    page = max(1, page)
    skip = (page - 1) * PAGE_ITEM_COUNT_PINNED
    limit = PAGE_ITEM_COUNT_PINNED

    message_list = await Messages.get_pinned_messages_by_channel_id(id, skip, limit, db=db)

    if not message_list:
        return []

    # Batch fetch all users in a single query (fixes N+1 problem)
    user_ids = list(set(m.user_id for m in message_list))
    fetched_users = {u.id: u for u in await Users.get_users_by_user_ids(user_ids, db=db)}

    # Batch fetch reactions in 1 query (fixes N+1)
    message_ids = [m.id for m in message_list]
    all_reactions = await Messages.get_reactions_by_message_ids(message_ids, db=db)

    messages = []
    for message in message_list:
        # Check for webhook identity in meta
        webhook_info = message.meta.get('webhook') if message.meta else None
        if webhook_info:
            user_info = UserNameResponse(
                id=webhook_info.get('id') or '',
                name=webhook_info.get('name') or 'Webhook',
                role='webhook',
            )
        elif message.user_id in fetched_users:
            user_info = UserNameResponse(**fetched_users[message.user_id].model_dump())
        else:
            user_info = None

        messages.append(
            MessageWithReactionsResponse(
                **{
                    **message.model_dump(),
                    'reactions': all_reactions.get(message.id, []),
                    'user': user_info,
                }
            )
        )

    return messages


############################
# PostNewMessage
############################


async def send_notification(request, channel, message, active_user_ids, db=None):
    webui_url = await Config.get('webui.url')
    enable_user_webhooks = await Config.get('ui.enable_user_webhooks')

    users = await get_channel_users_with_access(channel, 'read', db=db)

    # Batch fetch channel members in 1 query (fixes N+1)
    member_ids = {m.user_id for m in await Channels.get_members_by_channel_id(channel.id, db=db)}
    url = f'{webui_url}/channels/{channel.id}'

    for u in users:
        if (u.id not in active_user_ids) and u.id in member_ids:
            if enable_user_webhooks and u.settings:
                await publish_event(
                    request,
                    EVENTS.CHANNEL_MESSAGE,
                    subject_id=channel.id,
                    subject_type='channel',
                    data={
                        'user_id': u.id,
                        'channel_id': channel.id,
                        'message_id': message.id,
                        'sender_id': message.user_id,
                        'content': message.content,
                        'message': f'#{channel.name} - {url}\n\n{message.content}',
                        'content_preview': message.content[:300],
                        'title': channel.name,
                        'url': url,
                    },
                    message=channel.name,
                )

    return True


async def model_response_handler(request, channel, message, user, db=None):
    MODELS = {model['id']: model for model in await get_filtered_models(await get_all_models(request, user=user), user)}

    mentions = extract_mentions(message.content)
    message_content = replace_mentions(message.content)

    model_mentions = {}

    # check if the message is a reply to a message sent by a model
    if (
        message.reply_to_message
        and message.reply_to_message.meta
        and message.reply_to_message.meta.get('model_id', None)
    ):
        model_id = message.reply_to_message.meta.get('model_id', None)
        model_mentions[model_id] = {'id': model_id, 'id_type': 'M'}

    # check if any of the mentions are models
    for mention in mentions:
        if mention['id_type'] == 'M' and mention['id'] not in model_mentions:
            model_mentions[mention['id']] = mention

    if not model_mentions:
        return False

    for mention in model_mentions.values():
        model_id = mention['id']
        model = MODELS.get(model_id, None)

        if model:
            try:
                # reverse to get in chronological order
                thread_messages = (
                    await Messages.get_messages_by_parent_id(
                        channel.id,
                        message.parent_id if message.parent_id else message.id,
                        db=db,
                    )
                )[::-1]
                response_parent_id = (
                    message.parent_id
                    if message.parent_id
                    else (
                        message.id if await Config.get('channels.model_response_mode', 'thread') == 'thread' else None
                    )
                )

                response_message, channel = await new_message_handler(
                    request,
                    channel.id,
                    MessageForm(
                        **{
                            'parent_id': response_parent_id,
                            'content': f'',
                            'data': {},
                            'meta': {
                                'model_id': model_id,
                                'model_name': model.get('name', model_id),
                            },
                        }
                    ),
                    user,
                    db,
                )

                thread_history = []
                images = []
                files = []

                # Batch fetch all users in a single query (fixes N+1 problem)
                user_ids = list({message.user_id for message in thread_messages})
                message_users = {user.id: user for user in await Users.get_users_by_user_ids(user_ids, db=db)}

                for thread_message in thread_messages:
                    message_user = message_users.get(thread_message.user_id)

                    if thread_message.meta and thread_message.meta.get('model_id', None):
                        # If the message was sent by a model, use the model name
                        message_model_id = thread_message.meta.get('model_id', None)
                        message_model = MODELS.get(message_model_id, None)
                        username = message_model.get('name', message_model_id) if message_model else message_model_id
                    else:
                        username = message_user.name if message_user else 'Unknown'

                    thread_history.append(f'{username}: {replace_mentions(thread_message.content)}')

                    thread_message_files = (thread_message.data or {}).get('files', [])
                    for file in thread_message_files:
                        if file.get('type', '') == 'image':
                            images.append(file.get('url', ''))
                        elif file.get('content_type', '').startswith('image/'):
                            image = await get_image_base64_from_file_id(file.get('id', ''), user=user)
                            if image:
                                images.append(image)
                        elif file.get('id'):
                            files.append(file)

                thread_history_string = '\n\n'.join(thread_history)
                system_message = {
                    'role': 'system',
                    'content': f'You are {model.get("name", model_id)}, participating in a threaded conversation. Be concise and conversational.'
                    + (
                        f"Here's the thread history:\n\n\n{thread_history_string}\n\n\nContinue the conversation naturally as {model.get('name', model_id)}, addressing the most recent message while being aware of the full context."
                        if thread_history
                        else ''
                    ),
                }

                content = f'{user.name if user else "User"}: {message_content}'
                if images:
                    content = [
                        {
                            'type': 'text',
                            'text': content,
                        },
                        *[
                            {
                                'type': 'image_url',
                                'image_url': {
                                    'url': image,
                                },
                            }
                            for image in images
                        ],
                    ]

                # Resolve model config (same helpers automations use)
                from open_webui.utils.automations import (
                    _resolve_model_features,
                    _resolve_model_filter_ids,
                    _resolve_model_tool_ids,
                )

                tool_ids = _resolve_model_tool_ids(request.app, model_id)
                features = await _resolve_model_features(request.app, model_id)
                filter_ids = _resolve_model_filter_ids(request.app, model_id)

                # Build full form_data — same shape as frontend POST.
                # The channel: prefix routes pipeline events to the
                # channel emitter in socket/main.py instead of the
                # default chat emitter.
                form_data = {
                    'model': model_id,
                    'messages': [
                        system_message,
                        {'role': 'user', 'content': content},
                    ],
                    'stream': True,
                    'chat_id': f'channel:{channel.id}',
                    'id': response_message.id,
                    'session_id': f'channel:{channel.id}',
                    'background_tasks': {},
                }
                if files:
                    form_data['files'] = files
                if tool_ids:
                    form_data['tool_ids'] = tool_ids
                if features:
                    form_data['features'] = features
                if filter_ids:
                    form_data['filter_ids'] = filter_ids

                # Call the full chat completion pipeline — streaming,
                # tools, filters, RAG — everything. The pipeline runs as
                # an async task; the channel emitter handles progressive
                # message updates via socket events.
                await request.app.state.CHAT_COMPLETION_HANDLER(request, form_data, user=user)

            except Exception as e:
                log.exception(e)

    return True


async def new_message_handler(request: Request, id: str, form_data: MessageForm, user, db):
    channel = await Channels.get_channel_by_id(id, db=db)
    if not channel:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=ERROR_MESSAGES.NOT_FOUND)

    if channel.type in ['group', 'dm']:
        if not await Channels.is_user_channel_member(channel.id, user.id, db=db):
            raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=ERROR_MESSAGES.DEFAULT())
    else:
        if user.role != 'admin' and not await channel_has_access(
            user.id,
            channel,
            permission='write',
            strict=False,
            db=db,
        ):
            raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=ERROR_MESSAGES.DEFAULT())

    # Thread parent / reply target must belong to this channel (no cross-channel binding).
    for ref_id in (form_data.parent_id, form_data.reply_to_id):
        if ref_id:
            ref = await Messages.get_message_by_id(ref_id, include_thread_replies=False, db=db)
            if not ref or ref.channel_id != channel.id:
                raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=ERROR_MESSAGES.DEFAULT())

    try:
        message = await Messages.insert_new_message(form_data, channel.id, user.id, db=db)
        if message:
            if channel.type in ['group', 'dm']:
                members = await Channels.get_members_by_channel_id(channel.id, db=db)
                for member in members:
                    if not member.is_active:
                        await Channels.update_member_active_status(channel.id, member.user_id, True, db=db)

            message = await Messages.get_message_by_id(message.id, db=db)
            event_data = {
                'channel_id': channel.id,
                'message_id': message.id,
                'data': {
                    'type': 'message',
                    'data': {'temp_id': form_data.temp_id, **message.model_dump()},
                },
                'user': UserNameResponse(**user.model_dump()).model_dump(),
                'channel': channel.model_dump(),
            }

            await sio.emit(
                'events:channel',
                event_data,
                to=f'channel:{channel.id}',
            )

            if message.parent_id:
                # If this message is a reply, emit to the parent message as well
                parent_message = await Messages.get_message_by_id(message.parent_id, db=db)

                if parent_message:
                    await sio.emit(
                        'events:channel',
                        {
                            'channel_id': channel.id,
                            'message_id': parent_message.id,
                            'data': {
                                'type': 'message:reply',
                                'data': parent_message.model_dump(),
                            },
                            'user': UserNameResponse(**user.model_dump()).model_dump(),
                            'channel': channel.model_dump(),
                        },
                        to=f'channel:{channel.id}',
                    )
            return message, channel
        else:
            raise Exception('Error creating message')
    except Exception as e:
        log.exception(e)
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=ERROR_MESSAGES.DEFAULT())


@router.post('/{id}/messages/post', response_model=Optional[MessageModel])
async def post_new_message(
    request: Request,
    id: str,
    form_data: MessageForm,
    background_tasks: BackgroundTasks,
    user=Depends(get_verified_user),
    db: AsyncSession = Depends(get_async_session),
):
    await check_channels_access(request, user)

    try:
        message, channel = await new_message_handler(request, id, form_data, user, db)
        try:
            if files := message.data.get('files', []):
                for file in files:
                    await Channels.set_file_message_id_in_channel_by_id(
                        channel.id, file.get('id', ''), message.id, db=db
                    )
        except Exception as e:
            log.debug(e)

        active_user_ids = get_user_ids_from_room(f'channel:{channel.id}')

        # NOTE: We intentionally do NOT pass db to background_handler.
        # Background tasks should manage their own short-lived sessions to avoid
        # holding database connections during slow operations (e.g., LLM calls).
        async def background_handler():
            await model_response_handler(request, channel, message, user)
            await send_notification(
                request,
                channel,
                message,
                active_user_ids,
            )

        background_tasks.add_task(background_handler)

        await publish_event(
            request,
            EVENTS.MESSAGE_CREATED,
            actor=user,
            subject_id=message.id,
            data={
                'channel_id': channel.id,
                'content_preview': message.content[:300],
            },
        )
        return message

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


############################
# GetChannelMessage
############################


@router.get('/{id}/messages/{message_id}', response_model=Optional[MessageResponse])
async def get_channel_message(
    request: Request,
    id: str,
    message_id: str,
    user=Depends(get_verified_user),
    db: AsyncSession = Depends(get_async_session),
):
    await check_channels_access(request, user)
    channel = await Channels.get_channel_by_id(id, db=db)
    if not channel:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=ERROR_MESSAGES.NOT_FOUND)

    if channel.type in ['group', 'dm']:
        if not await Channels.is_user_channel_member(channel.id, user.id, db=db):
            raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=ERROR_MESSAGES.DEFAULT())
    else:
        if user.role != 'admin' and not await channel_has_access(user.id, channel, permission='read', db=db):
            raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=ERROR_MESSAGES.DEFAULT())

    message = await Messages.get_message_by_id(message_id, db=db)
    if not message:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=ERROR_MESSAGES.NOT_FOUND)

    if message.channel_id != id:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=ERROR_MESSAGES.DEFAULT())

    message_user = await Users.get_user_by_id(message.user_id, db=db)
    return MessageResponse(
        **{
            **message.model_dump(),
            'user': UserNameResponse(**message_user.model_dump()) if message_user else None,
        }
    )


############################
# GetChannelMessageData
############################


@router.get('/{id}/messages/{message_id}/data', response_model=Optional[dict])
async def get_channel_message_data(
    request: Request,
    id: str,
    message_id: str,
    user=Depends(get_verified_user),
    db: AsyncSession = Depends(get_async_session),
):
    await check_channels_access(request, user)
    channel = await Channels.get_channel_by_id(id, db=db)
    if not channel:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=ERROR_MESSAGES.NOT_FOUND)

    if channel.type in ['group', 'dm']:
        if not await Channels.is_user_channel_member(channel.id, user.id, db=db):
            raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=ERROR_MESSAGES.DEFAULT())
    else:
        if user.role != 'admin' and not await channel_has_access(user.id, channel, permission='read', db=db):
            raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=ERROR_MESSAGES.DEFAULT())

    message = await Messages.get_message_by_id(message_id, db=db)
    if not message:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=ERROR_MESSAGES.NOT_FOUND)

    if message.channel_id != id:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=ERROR_MESSAGES.DEFAULT())

    return message.data


############################
# PinChannelMessage
############################


class PinMessageForm(BaseModel):
    is_pinned: bool


@router.post('/{id}/messages/{message_id}/pin', response_model=Optional[MessageUserResponse])
async def pin_channel_message(
    request: Request,
    id: str,
    message_id: str,
    form_data: PinMessageForm,
    user=Depends(get_verified_user),
    db: AsyncSession = Depends(get_async_session),
):
    await check_channels_access(request, user)
    channel = await Channels.get_channel_by_id(id, db=db)
    if not channel:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=ERROR_MESSAGES.NOT_FOUND)

    if channel.type in ['group', 'dm']:
        if not await Channels.is_user_channel_member(channel.id, user.id, db=db):
            raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=ERROR_MESSAGES.DEFAULT())
    else:
        # Pin/unpin mutates is_pinned/pinned_by/pinned_at — require write.
        if user.role != 'admin' and not await channel_has_access(user.id, channel, permission='write', db=db):
            raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=ERROR_MESSAGES.DEFAULT())

    message = await Messages.get_message_by_id(message_id, db=db)
    if not message:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=ERROR_MESSAGES.NOT_FOUND)

    if message.channel_id != id:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=ERROR_MESSAGES.DEFAULT())

    try:
        await Messages.update_is_pinned_by_id(message_id, form_data.is_pinned, user.id, db=db)
        message = await Messages.get_message_by_id(message_id, db=db)
        message_user = await Users.get_user_by_id(message.user_id, db=db)
        message_data = MessageUserResponse(
            **{
                **message.model_dump(),
                'user': UserNameResponse(**message_user.model_dump()) if message_user else None,
            }
        )

        await sio.emit(
            'events:channel',
            {
                'channel_id': channel.id,
                'message_id': message.id,
                'data': {
                    'type': 'message:update',
                    'data': message_data.model_dump(),
                },
                'user': UserNameResponse(**user.model_dump()).model_dump(),
                'channel': channel.model_dump(),
            },
            to=f'channel:{channel.id}',
        )

        await publish_event(
            request,
            EVENTS.MESSAGE_PINNED if form_data.is_pinned else EVENTS.MESSAGE_UNPINNED,
            actor=user,
            subject_id=message_id,
            subject_type='message',
            data={'channel_id': id},
        )
        return message_data
    except Exception as e:
        log.exception(e)
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=ERROR_MESSAGES.DEFAULT())


############################
# GetChannelThreadMessages
############################


@router.get('/{id}/messages/{message_id}/thread', response_model=list[MessageUserResponse])
async def get_channel_thread_messages(
    request: Request,
    id: str,
    message_id: str,
    skip: int = 0,
    limit: int = 50,
    user=Depends(get_verified_user),
    db: AsyncSession = Depends(get_async_session),
):
    await check_channels_access(request, user)
    channel = await Channels.get_channel_by_id(id, db=db)
    if not channel:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=ERROR_MESSAGES.NOT_FOUND)

    if channel.type in ['group', 'dm']:
        if not await Channels.is_user_channel_member(channel.id, user.id, db=db):
            raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=ERROR_MESSAGES.DEFAULT())
    else:
        if user.role != 'admin' and not await channel_has_access(user.id, channel, permission='read', db=db):
            raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=ERROR_MESSAGES.DEFAULT())

    message_list = await Messages.get_messages_by_parent_id(id, message_id, skip, limit, db=db)

    if not message_list:
        return []

    # Batch fetch all users in a single query (fixes N+1 problem)
    user_ids = list(set(m.user_id for m in message_list))
    fetched_users = {u.id: u for u in await Users.get_users_by_user_ids(user_ids, db=db)}

    # Batch fetch reactions in 1 query (fixes N+1)
    message_ids = [m.id for m in message_list]
    all_reactions = await Messages.get_reactions_by_message_ids(message_ids, db=db)

    messages = []
    for message in message_list:
        # Use message.user if present (for webhooks), otherwise look up by user_id
        user_info = message.user
        if user_info is None and message.user_id in fetched_users:
            user_info = UserNameResponse(**fetched_users[message.user_id].model_dump())

        messages.append(
            MessageUserResponse(
                **{
                    **message.model_dump(),
                    'reply_count': 0,
                    'latest_reply_at': None,
                    'reactions': all_reactions.get(message.id, []),
                    'user': user_info,
                }
            )
        )

    return messages


############################
# UpdateMessageById
############################


@router.post('/{id}/messages/{message_id}/update', response_model=Optional[MessageModel])
async def update_message_by_id(
    request: Request,
    id: str,
    message_id: str,
    form_data: MessageForm,
    user=Depends(get_verified_user),
    db: AsyncSession = Depends(get_async_session),
):
    await check_channels_access(request, user)
    channel = await Channels.get_channel_by_id(id, db=db)
    if not channel:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=ERROR_MESSAGES.NOT_FOUND)

    message = await Messages.get_message_by_id(message_id, db=db)
    if not message:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=ERROR_MESSAGES.NOT_FOUND)

    if message.channel_id != id:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=ERROR_MESSAGES.DEFAULT())

    if channel.type in ['group', 'dm']:
        if not await Channels.is_user_channel_member(channel.id, user.id, db=db):
            raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=ERROR_MESSAGES.DEFAULT())
        # Membership is not authorship — block cross-member edits.
        if user.role != 'admin' and message.user_id != user.id:
            raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=ERROR_MESSAGES.DEFAULT())
    else:
        if user.role != 'admin' and not await channel_has_access(
            user.id, channel, permission='write', strict=False, db=db
        ):
            raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=ERROR_MESSAGES.DEFAULT())
        # Write access is not authorship — block cross-member edits.
        if user.role != 'admin' and message.user_id != user.id:
            raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=ERROR_MESSAGES.DEFAULT())

    try:
        await Messages.update_message_by_id(message_id, form_data, db=db)
        message = await Messages.get_message_by_id(message_id, db=db)

        if message:
            await sio.emit(
                'events:channel',
                {
                    'channel_id': channel.id,
                    'message_id': message.id,
                    'data': {
                        'type': 'message:update',
                        'data': message.model_dump(),
                    },
                    'user': UserNameResponse(**user.model_dump()).model_dump(),
                    'channel': channel.model_dump(),
                },
                to=f'channel:{channel.id}',
            )

        await publish_event(
            request,
            EVENTS.MESSAGE_UPDATED,
            actor=user,
            subject_id=message_id,
            data={'channel_id': id, 'content_preview': form_data.content[:300]},
        )
        return MessageModel(**message.model_dump())
    except Exception as e:
        log.exception(e)
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=ERROR_MESSAGES.DEFAULT())


############################
# AddReactionToMessage
############################


class ReactionForm(BaseModel):
    name: str


@router.post('/{id}/messages/{message_id}/reactions/add', response_model=bool)
async def add_reaction_to_message(
    request: Request,
    id: str,
    message_id: str,
    form_data: ReactionForm,
    user=Depends(get_verified_user),
    db: AsyncSession = Depends(get_async_session),
):
    await check_channels_access(request, user)
    channel = await Channels.get_channel_by_id(id, db=db)
    if not channel:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=ERROR_MESSAGES.NOT_FOUND)

    if channel.type in ['group', 'dm']:
        if not await Channels.is_user_channel_member(channel.id, user.id, db=db):
            raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=ERROR_MESSAGES.DEFAULT())
    else:
        if user.role != 'admin' and not await channel_has_access(
            user.id,
            channel,
            permission='write',
            strict=False,
            db=db,
        ):
            raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=ERROR_MESSAGES.DEFAULT())

    message = await Messages.get_message_by_id(message_id, db=db)
    if not message:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=ERROR_MESSAGES.NOT_FOUND)

    if message.channel_id != id:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=ERROR_MESSAGES.DEFAULT())

    try:
        await Messages.add_reaction_to_message(message_id, user.id, form_data.name, db=db)
        message = await Messages.get_message_by_id(message_id, db=db)

        await sio.emit(
            'events:channel',
            {
                'channel_id': channel.id,
                'message_id': message.id,
                'data': {
                    'type': 'message:reaction:add',
                    'data': {
                        **message.model_dump(),
                        'name': form_data.name,
                    },
                },
                'user': UserNameResponse(**user.model_dump()).model_dump(),
                'channel': channel.model_dump(),
            },
            to=f'channel:{channel.id}',
        )

        await publish_event(
            request,
            EVENTS.MESSAGE_REACTION_ADDED,
            actor=user,
            subject_id=message_id,
            data={'channel_id': id, 'reaction': form_data.name},
        )
        return True
    except Exception as e:
        log.exception(e)
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=ERROR_MESSAGES.DEFAULT())


############################
# RemoveReactionById
############################


@router.post('/{id}/messages/{message_id}/reactions/remove', response_model=bool)
async def remove_reaction_by_id_and_user_id_and_name(
    request: Request,
    id: str,
    message_id: str,
    form_data: ReactionForm,
    user=Depends(get_verified_user),
    db: AsyncSession = Depends(get_async_session),
):
    await check_channels_access(request, user)
    channel = await Channels.get_channel_by_id(id, db=db)
    if not channel:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=ERROR_MESSAGES.NOT_FOUND)

    if channel.type in ['group', 'dm']:
        if not await Channels.is_user_channel_member(channel.id, user.id, db=db):
            raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=ERROR_MESSAGES.DEFAULT())
    else:
        if user.role != 'admin' and not await channel_has_access(
            user.id,
            channel,
            permission='write',
            strict=False,
            db=db,
        ):
            raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=ERROR_MESSAGES.DEFAULT())

    message = await Messages.get_message_by_id(message_id, db=db)
    if not message:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=ERROR_MESSAGES.NOT_FOUND)

    if message.channel_id != id:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=ERROR_MESSAGES.DEFAULT())

    try:
        await Messages.remove_reaction_by_id_and_user_id_and_name(message_id, user.id, form_data.name, db=db)

        message = await Messages.get_message_by_id(message_id, db=db)

        await sio.emit(
            'events:channel',
            {
                'channel_id': channel.id,
                'message_id': message.id,
                'data': {
                    'type': 'message:reaction:remove',
                    'data': {
                        **message.model_dump(),
                        'name': form_data.name,
                    },
                },
                'user': UserNameResponse(**user.model_dump()).model_dump(),
                'channel': channel.model_dump(),
            },
            to=f'channel:{channel.id}',
        )

        await publish_event(
            request,
            EVENTS.MESSAGE_REACTION_REMOVED,
            actor=user,
            subject_id=message_id,
            data={'channel_id': id, 'reaction': form_data.name},
        )
        return True
    except Exception as e:
        log.exception(e)
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=ERROR_MESSAGES.DEFAULT())


############################
# DeleteMessageById
############################


@router.delete('/{id}/messages/{message_id}/delete', response_model=bool)
async def delete_message_by_id(
    request: Request,
    id: str,
    message_id: str,
    user=Depends(get_verified_user),
    db: AsyncSession = Depends(get_async_session),
):
    await check_channels_access(request, user)
    channel = await Channels.get_channel_by_id(id, db=db)
    if not channel:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=ERROR_MESSAGES.NOT_FOUND)

    message = await Messages.get_message_by_id(message_id, db=db)
    if not message:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=ERROR_MESSAGES.NOT_FOUND)

    if message.channel_id != id:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=ERROR_MESSAGES.DEFAULT())

    if channel.type in ['group', 'dm']:
        if not await Channels.is_user_channel_member(channel.id, user.id, db=db):
            raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=ERROR_MESSAGES.DEFAULT())
        # Membership is not authorship — block cross-member deletes.
        if user.role != 'admin' and message.user_id != user.id:
            raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=ERROR_MESSAGES.DEFAULT())
    else:
        if user.role != 'admin' and not await channel_has_access(
            user.id,
            channel,
            permission='write',
            strict=False,
            db=db,
        ):
            raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=ERROR_MESSAGES.DEFAULT())
        # Write access is not authorship — block cross-member deletes.
        if user.role != 'admin' and message.user_id != user.id:
            raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=ERROR_MESSAGES.DEFAULT())

    try:
        await Messages.delete_message_by_id(message_id, db=db)
        await sio.emit(
            'events:channel',
            {
                'channel_id': channel.id,
                'message_id': message.id,
                'data': {
                    'type': 'message:delete',
                    'data': {
                        **message.model_dump(),
                        'user': UserNameResponse(**user.model_dump()).model_dump(),
                    },
                },
                'user': UserNameResponse(**user.model_dump()).model_dump(),
                'channel': channel.model_dump(),
            },
            to=f'channel:{channel.id}',
        )

        if message.parent_id:
            # If this message is a reply, emit to the parent message as well
            parent_message = await Messages.get_message_by_id(message.parent_id, db=db)

            if parent_message:
                await sio.emit(
                    'events:channel',
                    {
                        'channel_id': channel.id,
                        'message_id': parent_message.id,
                        'data': {
                            'type': 'message:reply',
                            'data': parent_message.model_dump(),
                        },
                        'user': UserNameResponse(**user.model_dump()).model_dump(),
                        'channel': channel.model_dump(),
                    },
                    to=f'channel:{channel.id}',
                )

        await publish_event(
            request,
            EVENTS.MESSAGE_DELETED,
            actor=user,
            subject_id=message_id,
            data={'channel_id': id},
        )
        return True
    except Exception as e:
        log.exception(e)
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=ERROR_MESSAGES.DEFAULT())


############################
# Webhooks
############################


@router.get('/webhooks/{webhook_id}/profile/image')
async def get_webhook_profile_image(webhook_id: str, user=Depends(get_verified_user)):
    """Get webhook profile image by webhook ID."""
    webhook = await Channels.get_webhook_by_id(webhook_id)
    if not webhook:
        # Return default favicon if webhook not found
        # LICENSE covers this Open WebUI fallback logo.
        # Do not alter, remove, obscure, or replace it except as LICENSE permits:
        # https://docs.openwebui.com/license.
        return FileResponse(f'{STATIC_DIR}/favicon.png')

    if webhook.profile_image_url:
        # Check if it's url or base64
        if webhook.profile_image_url.startswith('http'):
            return Response(
                status_code=status.HTTP_302_FOUND,
                headers={'Location': webhook.profile_image_url},
            )
        elif webhook.profile_image_url.startswith('data:image'):
            try:
                header, base64_data = webhook.profile_image_url.split(',', 1)
                image_data = base64.b64decode(base64_data)
                image_buffer = io.BytesIO(image_data)
                media_type = header.split(';')[0].lstrip('data:')

                return StreamingResponse(
                    image_buffer,
                    media_type=media_type,
                    headers={'Content-Disposition': 'inline'},
                )
            except Exception as e:
                pass

    # Return default favicon if no profile image
    # LICENSE covers this Open WebUI fallback logo.
    # Do not alter, remove, obscure, or replace it except as LICENSE permits:
    # https://docs.openwebui.com/license.
    return FileResponse(f'{STATIC_DIR}/favicon.png')


@router.get('/{id}/webhooks', response_model=list[ChannelWebhookModel])
async def get_channel_webhooks(
    request: Request,
    id: str,
    user=Depends(get_verified_user),
    db: AsyncSession = Depends(get_async_session),
):
    await check_channels_access(request, user)
    channel = await Channels.get_channel_by_id(id, db=db)
    if not channel:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=ERROR_MESSAGES.NOT_FOUND)

    # Only channel managers can view webhooks
    if not await Channels.is_user_channel_manager(channel.id, user.id, db=db) and user.role != 'admin':
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=ERROR_MESSAGES.UNAUTHORIZED)

    return await Channels.get_webhooks_by_channel_id(id, db=db)


@router.post('/{id}/webhooks/create', response_model=ChannelWebhookModel)
async def create_channel_webhook(
    request: Request,
    id: str,
    form_data: ChannelWebhookForm,
    user=Depends(get_verified_user),
    db: AsyncSession = Depends(get_async_session),
):
    await check_channels_access(request, user)
    channel = await Channels.get_channel_by_id(id, db=db)
    if not channel:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=ERROR_MESSAGES.NOT_FOUND)

    # Only channel managers can create webhooks
    if not await Channels.is_user_channel_manager(channel.id, user.id, db=db) and user.role != 'admin':
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=ERROR_MESSAGES.UNAUTHORIZED)

    webhook = await Channels.insert_webhook(id, user.id, form_data, db=db)
    if not webhook:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=ERROR_MESSAGES.DEFAULT())

    await publish_event(
        request,
        EVENTS.CHANNEL_WEBHOOK_CREATED,
        actor=user,
        subject_id=webhook.id,
        data={'channel_id': id, 'name': webhook.name},
    )
    return webhook


@router.post('/{id}/webhooks/{webhook_id}/update', response_model=ChannelWebhookModel)
async def update_channel_webhook(
    request: Request,
    id: str,
    webhook_id: str,
    form_data: ChannelWebhookForm,
    user=Depends(get_verified_user),
    db: AsyncSession = Depends(get_async_session),
):
    await check_channels_access(request, user)
    channel = await Channels.get_channel_by_id(id, db=db)
    if not channel:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=ERROR_MESSAGES.NOT_FOUND)

    # Only channel managers can update webhooks
    if not await Channels.is_user_channel_manager(channel.id, user.id, db=db) and user.role != 'admin':
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=ERROR_MESSAGES.UNAUTHORIZED)

    webhook = await Channels.get_webhook_by_id(webhook_id, db=db)
    if not webhook or webhook.channel_id != id:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=ERROR_MESSAGES.NOT_FOUND)

    updated = await Channels.update_webhook_by_id(webhook_id, form_data, db=db)
    if not updated:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=ERROR_MESSAGES.DEFAULT())

    await publish_event(
        request,
        EVENTS.CHANNEL_WEBHOOK_UPDATED,
        actor=user,
        subject_id=webhook_id,
        data={'channel_id': id, 'name': updated.name},
    )
    return updated


@router.delete('/{id}/webhooks/{webhook_id}/delete', response_model=bool)
async def delete_channel_webhook(
    request: Request,
    id: str,
    webhook_id: str,
    user=Depends(get_verified_user),
    db: AsyncSession = Depends(get_async_session),
):
    await check_channels_access(request, user)
    channel = await Channels.get_channel_by_id(id, db=db)
    if not channel:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=ERROR_MESSAGES.NOT_FOUND)

    # Only channel managers can delete webhooks
    if not await Channels.is_user_channel_manager(channel.id, user.id, db=db) and user.role != 'admin':
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=ERROR_MESSAGES.UNAUTHORIZED)

    webhook = await Channels.get_webhook_by_id(webhook_id, db=db)
    if not webhook or webhook.channel_id != id:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=ERROR_MESSAGES.NOT_FOUND)

    deleted = await Channels.delete_webhook_by_id(webhook_id, db=db)
    if deleted:
        await publish_event(
            request,
            EVENTS.CHANNEL_WEBHOOK_DELETED,
            actor=user,
            subject_id=webhook_id,
            data={'channel_id': id},
        )
    return deleted


############################
# Public Webhook Endpoint
############################


class WebhookMessageForm(BaseModel):
    content: str


@router.post('/webhooks/{webhook_id}/{token}')
async def post_webhook_message(
    request: Request,
    webhook_id: str,
    token: str,
    form_data: WebhookMessageForm,
    db: AsyncSession = Depends(get_async_session),
):
    """Public endpoint to post messages via webhook. No authentication required."""
    await check_channels_access(request)

    # Validate webhook
    webhook = await Channels.get_webhook_by_id_and_token(webhook_id, token, db=db)
    if not webhook:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail=ERROR_MESSAGES.INVALID_URL,
        )

    channel = await Channels.get_channel_by_id(webhook.channel_id, db=db)
    if not channel:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=ERROR_MESSAGES.NOT_FOUND)

    # Create message with webhook identity stored in meta
    message = await Messages.insert_new_message(
        MessageForm(content=form_data.content, meta={'webhook': {'id': webhook.id}}),
        webhook.channel_id,
        webhook.user_id,  # Required for DB but webhook info in meta takes precedence
        db=db,
    )

    if not message:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=ERROR_MESSAGES.DEFAULT('Failed to create message'),
        )

    # Update last_used_at
    await Channels.update_webhook_last_used_at(webhook_id, db=db)

    # Get full message and emit event
    message = await Messages.get_message_by_id(message.id, db=db)

    event_data = {
        'channel_id': channel.id,
        'message_id': message.id,
        'data': {
            'type': 'message',
            'data': {
                **message.model_dump(),
                'user': {
                    'id': webhook.id,
                    'name': webhook.name,
                    'role': 'webhook',
                },
            },
        },
        'user': {
            'id': webhook.id,
            'name': webhook.name,
            'role': 'webhook',
        },
        'channel': channel.model_dump(),
    }

    await sio.emit(
        'events:channel',
        event_data,
        to=f'channel:{channel.id}',
    )

    await publish_event(
        request,
        EVENTS.MESSAGE_CREATED,
        actor={'id': webhook.id, 'name': webhook.name, 'role': 'webhook', 'type': 'webhook'},
        subject_id=message.id,
        source='channel_webhook',
        data={'channel_id': channel.id, 'content_preview': form_data.content[:300]},
    )
    return {'success': True, 'message_id': message.id}