mirror of
https://github.com/open-webui/open-webui.git
synced 2026-09-10 22:43:29 +00:00
d7674c5174
* perf: stop the Socket.IO session pool blocking the websocket event loop With WEBSOCKET_MANAGER=redis the session pool is a synchronous Redis client, so every call into it blocks the whole worker's event loop, not just the caller. Two paths did it constantly: the orphan reaper walked the pool one round trip per session with no await anywhere, freezing the loop for the entire sweep every cycle, and nearly every socket event re-read the sender's session back out of Redis. Other users' events and every in-flight generation on that pod wait behind both. The reaper now walks the pool in HSCAN batches and deletes in bulk, yielding between batches, and no longer sleeps past half the lock TTL, which previously guaranteed a failed renew every cycle. The per-event reads are gone: Socket.IO events only reach the worker holding the connection, and that worker already saved the same session dict locally when the user authenticated, so it was asking Redis for its own data. The writes stay, since those are what other pods read. Measured at 4000 users / 16 containers, Redis 1.1 ms away: | | before | after | |---|---|---| | reaper sweep, 5k sessions | 5.6 s, loop frozen throughout | 62 ms, 4.2 ms worst block | | same, crash recovery with every session expired | 11.5 s | 96 ms | | heartbeat / usage ping / disconnect | 2 / 3 / 2 round trips | 1 / 2 / 1 | | 50-member channel post | 50 round trips, 57.2 ms block | 0 round trips, 0.02 ms | | loop time per wall second at rest | 103 ms (10.3%) | 67 ms (6.7%) | The alternative, converting RedisDict to the async client, fixes the same paths with a far larger blast radius (every call site gains await, and `in`/`[]`/`del` cannot be awaited so the dict interface goes) and still round-trips for data already in memory. Two deliberate behaviour changes: a heartbeat re-adds a session the reaper already removed, so a tab that survives a stall recovers instead of staying out of the pool until it reconnects; and disconnect no longer skips Yjs document cleanup when the pool entry is already gone, which previously leaked that document's update log forever. Closes #28172 * perf: cut disconnect and user session lookup pool round trips, harden the session reaper Follow-up on top of the session pool reaper branch. With WEBSOCKET_MANAGER=redis two paths still blocked the worker's event loop on synchronous Redis calls. Every disconnect listed all models in use cluster-wide and fetched each one individually, one blocking round trip per model. Disconnecting all sessions of a user (admin role change or deletion) pulled the entire session pool in one HGETALL and decoded every entry in a single uninterrupted block. Disconnect now fetches the usage pool once with items(), going from 2+N+M round trips to 2+M (N models in use cluster-wide, M models the session used), and its delete of an emptied model entry is KeyError-guarded because another node can remove the same key between snapshot and delete; unguarded, that race aborted the handler and skipped its Yjs document cleanup. The user session lookup reuses the reaper's HSCAN batches and yields to the loop between pages. The reaper previously died permanently on the first Redis connection error, on every node at once during an outage; it now logs, releases the lock and returns to retrying acquisition. * refac: keep the socket pool perf work to the round trips A review pass on this branch turned up four changes riding along with the round-trip work without belonging to it, so they are backed out here. The `disconnect` handler keeps its `if sid in SESSION_POOL:` guard, so USAGE_POOL and ydoc cleanup stay off the path for sockets that never authenticated. `RedisDict.set()` keeps its own inline HDEL. `get_session_ids_by_user_id` stays synchronous over one HGETALL, since it runs on user delete and role change rather than per message. The crash-resilience wrapper around the reaper loop is dropped; if that guard is worth having, it belongs in its own change. What stays is the perf part. The reaper now sweeps the pool in bounded HSCAN batches and deletes expired sids with one HDEL per batch, down from HKEYS plus an HGET and a per-sid HDEL across the whole pool. The `disconnect` handler reads USAGE_POOL with a single HGETALL, down from HKEYS plus one HGET per model in use. Session lookups in the socket handlers come from the local Socket.IO store, which removes one Redis GET from every heartbeat, usage, channel and ydoc event. Naming and annotations follow the file: `get_session_pool_batches` for the module's `get_` prefix, `RedisDict.pop_many` so both reaper branches use one word for removing keys, a named `SCAN_BATCH_SIZE`, and types on the new helpers. * fix: invalidate the RedisDict write signature on batch delete RedisDict.set() skips the write when the payload fingerprint matches the last one this process wrote, so a mutation that goes around set() has to clear that fingerprint. The new batch delete did not, leaving a stale fingerprint behind: the next refresh with identical content is treated as already written and silently skipped, so the hash stays empty. Renamed pop_many to delete_many. In a dict emulation pop removes and returns; this returns nothing and cannot without an extra HMGET, so the name promised something it does not do. delete_many matches __delitem__ and the HDEL underneath. Its only call site is the session pool reaper, whose behaviour is unchanged: same fields deleted, same batching, same return.
2072 lines
74 KiB
Python
2072 lines
74 KiB
Python
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)),
|
|
}
|
|
|
|
|
|
async def get_channel_member_user_ids(
|
|
channel: ChannelModel,
|
|
db: Optional[AsyncSession] = None,
|
|
) -> Optional[list[str]]:
|
|
permitted_ids = get_channel_permitted_group_and_user_ids(channel, permission='read')
|
|
if permitted_ids is None:
|
|
return None
|
|
|
|
user_ids = permitted_ids.get('user_ids') or []
|
|
group_ids = permitted_ids.get('group_ids') or []
|
|
if group_ids:
|
|
for member_ids in (await Groups.get_group_user_ids_by_ids(group_ids, db=db)).values():
|
|
user_ids.extend(member_ids)
|
|
|
|
return list(dict.fromkeys([*user_ids, channel.user_id]))
|
|
|
|
|
|
############################
|
|
# 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':
|
|
member_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(member_user_ids, db=db)
|
|
]
|
|
user_ids = [u.id for u in users]
|
|
|
|
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())
|
|
|
|
member_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(member_user_ids, db=db)
|
|
]
|
|
user_ids = [u.id for u in users]
|
|
|
|
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(users),
|
|
'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,
|
|
)
|
|
|
|
filter = {'roles': ['!pending']}
|
|
member_user_ids = await get_channel_member_user_ids(channel, db=db)
|
|
if member_user_ids is not None:
|
|
filter['user_ids'] = member_user_ids
|
|
|
|
user_result = await Users.get_users(filter=filter, limit=0, db=db)
|
|
user_count = user_result['total']
|
|
|
|
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 channel.type == 'group':
|
|
filter['channel_id'] = channel.id
|
|
else:
|
|
filter['roles'] = ['!pending']
|
|
member_user_ids = await get_channel_member_user_ids(channel, db=db)
|
|
if member_user_ids is not None:
|
|
filter['user_ids'] = member_user_ids
|
|
|
|
result = await Users.get_users(
|
|
filter=filter,
|
|
sort={'order_by': order_by, 'direction': direction},
|
|
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 path automations use)
|
|
from open_webui.utils.automations import _resolve_model_defaults
|
|
|
|
tool_ids, features, filter_ids, _ = await _resolve_model_defaults(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 = await 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}
|