mirror of
https://github.com/open-webui/open-webui.git
synced 2026-08-17 11:11:35 -06:00
Timothy Jaeryang Baek
2261 lines
74 KiB
Python
2261 lines
74 KiB
Python
from __future__ import annotations
|
|
|
|
import logging
|
|
from uuid import uuid4
|
|
|
|
from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Request, Response, status
|
|
from fastapi.responses import StreamingResponse
|
|
from fastapi.security import HTTPAuthorizationCredentials
|
|
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.internal.db import get_async_session
|
|
from open_webui.models.access_grants import AccessGrants
|
|
from open_webui.models.chat_messages import ChatMessages
|
|
from open_webui.models.chats import (
|
|
AggregateChatStats,
|
|
ChatBody,
|
|
ChatForm,
|
|
ChatHistoryStats,
|
|
ChatImportForm,
|
|
ChatResponse,
|
|
Chats,
|
|
ChatsImportForm,
|
|
ChatStatsExport,
|
|
ChatTitleIdResponse,
|
|
ChatUsageStatsListResponse,
|
|
MessageStats,
|
|
chat_search_content_query,
|
|
chat_search_terms,
|
|
is_internal_chat,
|
|
)
|
|
from open_webui.models.config import Config
|
|
from open_webui.models.folders import Folders
|
|
from open_webui.models.shared_chats import SharedChatResponse, SharedChats
|
|
from open_webui.models.tags import TagModel, Tags
|
|
from open_webui.socket.main import get_event_emitter
|
|
from open_webui.tasks import get_response_streams_by_chat_id, has_active_tasks, stop_item_tasks
|
|
from open_webui.utils.access_control import filter_allowed_access_grants, has_permission
|
|
from open_webui.utils.access_control.folders import has_folder_access, has_folder_write_access
|
|
from open_webui.utils.auth import bearer_security, get_admin_user, get_current_user, get_verified_user
|
|
from open_webui.utils.chat_fork import build_fork_history
|
|
from open_webui.utils.context_compaction import compact_chat_branch, get_chat_context_usage
|
|
from open_webui.utils.misc import get_message_list
|
|
from open_webui.utils.models import get_all_models
|
|
from pydantic import BaseModel
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
router = APIRouter()
|
|
|
|
CHAT_CONFIG_KEYS = {
|
|
'CONTEXT_COMPACTION_MODEL': 'chat.context_compaction.model',
|
|
'ENABLE_CONTEXT_COMPACTION': 'chat.context_compaction.enable',
|
|
'CONTEXT_COMPACTION_TOKEN_THRESHOLD': 'chat.context_compaction.token_threshold',
|
|
'CONTEXT_COMPACTION_TOKEN_CAP': 'chat.context_compaction.token_cap',
|
|
'CONTEXT_COMPACTION_RETENTION_PERCENTAGE': 'chat.context_compaction.retention_percentage',
|
|
'CONTEXT_COMPACTION_PROMPT_TEMPLATE': 'chat.context_compaction.prompt_template',
|
|
'ENABLE_TOOL_PERMISSIONS': 'chat.tool_permissions.enable',
|
|
}
|
|
|
|
|
|
def overlay_response_streams(chat_data: dict, response_streams: list[dict]) -> dict:
|
|
if not response_streams:
|
|
return chat_data
|
|
|
|
messages = chat_data.get('chat', {}).get('history', {}).get('messages')
|
|
if isinstance(messages, dict):
|
|
for stream in response_streams:
|
|
message_id = stream.get('message_id')
|
|
message = messages.get(message_id)
|
|
if isinstance(message, dict):
|
|
message['content'] = stream.get('content', '')
|
|
message['output'] = stream.get('output') or []
|
|
message['done'] = False
|
|
|
|
legacy_messages = chat_data.get('chat', {}).get('messages')
|
|
if isinstance(legacy_messages, list):
|
|
streams_by_message_id = {stream.get('message_id'): stream for stream in response_streams}
|
|
for message in legacy_messages:
|
|
if isinstance(message, dict) and (stream := streams_by_message_id.get(message.get('id'))):
|
|
message['content'] = stream.get('content', '')
|
|
message['output'] = stream.get('output') or []
|
|
message['done'] = False
|
|
|
|
return chat_data
|
|
|
|
|
|
async def get_optional_verified_user(
|
|
request: Request,
|
|
response: Response,
|
|
background_tasks: BackgroundTasks,
|
|
auth_token: HTTPAuthorizationCredentials | None = Depends(bearer_security),
|
|
):
|
|
try:
|
|
user = await get_current_user(request, response, background_tasks, auth_token)
|
|
except HTTPException:
|
|
return None
|
|
|
|
if user.role not in {'user', 'admin'}:
|
|
return None
|
|
return user
|
|
|
|
|
|
async def is_open_shared_chat(shared, db: AsyncSession) -> bool:
|
|
return await AccessGrants.has_anyone_access(
|
|
resource_type='shared_chat',
|
|
resource_id=shared.chat_id,
|
|
permission='read',
|
|
db=db,
|
|
)
|
|
|
|
|
|
async def can_read_shared_chat(user, shared, db: AsyncSession) -> bool:
|
|
if user.role == 'pending':
|
|
return False
|
|
if user.role == 'admin' and ENABLE_ADMIN_CHAT_ACCESS:
|
|
return True
|
|
if shared.user_id == user.id:
|
|
return True
|
|
return await AccessGrants.has_access(
|
|
user_id=user.id,
|
|
resource_type='shared_chat',
|
|
resource_id=shared.chat_id,
|
|
permission='read',
|
|
db=db,
|
|
)
|
|
|
|
|
|
async def add_active_state_to_chat_list(
|
|
request: Request, chat_list: list[ChatTitleIdResponse]
|
|
) -> list[ChatTitleIdResponse]:
|
|
for chat in chat_list:
|
|
chat.active = False
|
|
if not await has_active_tasks(request.app.state.redis, chat.id):
|
|
continue
|
|
|
|
chat.active = await ChatMessages.has_unfinished_assistant_by_chat_id(chat.id)
|
|
|
|
return chat_list
|
|
|
|
|
|
async def get_folder_unread_counts(user_id: str, db: AsyncSession | None = None) -> dict[str, int]:
|
|
user_folders = await Folders.get_folders_by_user_id(user_id, db=db)
|
|
parent_by_id = {folder.id: folder.parent_id for folder in user_folders}
|
|
unread_counts = dict.fromkeys(parent_by_id.keys(), 0)
|
|
direct_unread_counts = await Chats.count_unread_by_folder_ids(user_id, list(parent_by_id.keys()), db=db)
|
|
|
|
for unread_folder_id, unread_count in direct_unread_counts.items():
|
|
current_id = unread_folder_id
|
|
seen = set()
|
|
while current_id and current_id not in seen:
|
|
seen.add(current_id)
|
|
if current_id in unread_counts:
|
|
unread_counts[current_id] += unread_count
|
|
current_id = parent_by_id.get(current_id)
|
|
|
|
return unread_counts
|
|
|
|
|
|
class ChatConfigForm(BaseModel):
|
|
CONTEXT_COMPACTION_MODEL: str | None = ''
|
|
ENABLE_CONTEXT_COMPACTION: bool
|
|
CONTEXT_COMPACTION_TOKEN_THRESHOLD: int
|
|
CONTEXT_COMPACTION_TOKEN_CAP: int | None = None
|
|
CONTEXT_COMPACTION_RETENTION_PERCENTAGE: int = 40
|
|
CONTEXT_COMPACTION_PROMPT_TEMPLATE: str
|
|
ENABLE_TOOL_PERMISSIONS: bool = False
|
|
|
|
|
|
class CompactChatForm(BaseModel):
|
|
model: str | None = None
|
|
|
|
|
|
def chat_search_content_text(text: str) -> str:
|
|
return chat_search_content_query(text)
|
|
|
|
|
|
def chat_search_snippet(chat: dict, search_text: str, max_length: int = 200) -> str | None:
|
|
if not search_text:
|
|
return None
|
|
|
|
history = chat.get('history', {})
|
|
messages = history.get('messages') if isinstance(history, dict) else None
|
|
if not messages:
|
|
messages = chat.get('messages', []) or []
|
|
if isinstance(messages, dict):
|
|
messages = messages.values()
|
|
|
|
needles = list(dict.fromkeys([search_text, *chat_search_terms(search_text)]))
|
|
for needle in needles:
|
|
for message in messages:
|
|
if not isinstance(message, dict):
|
|
continue
|
|
|
|
content = message.get('content')
|
|
if not isinstance(content, str):
|
|
continue
|
|
|
|
index = content.lower().find(needle)
|
|
if index == -1:
|
|
continue
|
|
|
|
start = max(index - max_length // 2, 0)
|
|
end = min(start + max_length, len(content))
|
|
if index + len(needle) > end:
|
|
end = min(index + len(needle), len(content))
|
|
start = max(end - max_length, 0)
|
|
|
|
snippet = ' '.join(content[start:end].split())
|
|
return f'{"..." if start else ""}{snippet}{"..." if end < len(content) else ""}'
|
|
|
|
return None
|
|
|
|
|
|
async def get_chat_config_values() -> dict:
|
|
values = await Config.get_many(*CHAT_CONFIG_KEYS.values())
|
|
config = {field: values[storage_key] for field, storage_key in CHAT_CONFIG_KEYS.items() if storage_key in values}
|
|
if config.get('CONTEXT_COMPACTION_MODEL') is None:
|
|
config['CONTEXT_COMPACTION_MODEL'] = ''
|
|
if config.get('CONTEXT_COMPACTION_TOKEN_CAP') is None:
|
|
config['CONTEXT_COMPACTION_TOKEN_CAP'] = config.get('CONTEXT_COMPACTION_TOKEN_THRESHOLD', 80000)
|
|
if config.get('CONTEXT_COMPACTION_RETENTION_PERCENTAGE') is None:
|
|
config['CONTEXT_COMPACTION_RETENTION_PERCENTAGE'] = 40
|
|
return config
|
|
|
|
|
|
def chat_config_updates(data: dict) -> dict:
|
|
return {CHAT_CONFIG_KEYS[field]: value for field, value in data.items() if field in CHAT_CONFIG_KEYS}
|
|
|
|
|
|
async def require_chat_import_permission(request: Request, user, db: AsyncSession):
|
|
if user.role != 'admin' and not await has_permission(
|
|
user.id, 'chat.import', await Config.get('user.permissions'), db=db
|
|
):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail=ERROR_MESSAGES.ACCESS_PROHIBITED,
|
|
)
|
|
|
|
|
|
############################
|
|
# GetChatList
|
|
# Let the record outlive the session, so that what was
|
|
# learned here not need to be learned again.
|
|
############################
|
|
|
|
|
|
@router.get('/', response_model=list[ChatTitleIdResponse])
|
|
@router.get('/list', response_model=list[ChatTitleIdResponse])
|
|
async def get_session_user_chat_list(
|
|
request: Request,
|
|
user=Depends(get_verified_user),
|
|
page: int | None = None,
|
|
include_pinned: bool | None = False,
|
|
include_folders: bool | None = False,
|
|
sort_by: str = 'updated_at',
|
|
sort_dir: str = 'desc',
|
|
db: AsyncSession = Depends(get_async_session),
|
|
):
|
|
try:
|
|
if page is not None:
|
|
limit = 60
|
|
skip = (page - 1) * limit
|
|
|
|
chats = await Chats.get_chat_title_id_list_by_user_id(
|
|
user.id,
|
|
include_folders=include_folders,
|
|
include_pinned=include_pinned,
|
|
sort_by=sort_by,
|
|
sort_dir=sort_dir,
|
|
skip=skip,
|
|
limit=limit,
|
|
db=db,
|
|
)
|
|
else:
|
|
chats = await Chats.get_chat_title_id_list_by_user_id(
|
|
user.id,
|
|
include_folders=include_folders,
|
|
include_pinned=include_pinned,
|
|
sort_by=sort_by,
|
|
sort_dir=sort_dir,
|
|
db=db,
|
|
)
|
|
return await add_active_state_to_chat_list(request, chats)
|
|
except Exception as e:
|
|
log.exception(e)
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=ERROR_MESSAGES.DEFAULT())
|
|
|
|
|
|
@router.post('/read')
|
|
async def mark_chats_read_by_user_id(
|
|
user=Depends(get_verified_user),
|
|
db: AsyncSession = Depends(get_async_session),
|
|
):
|
|
return {
|
|
'updated_count': await Chats.mark_chats_read_by_user_id(user.id, db=db),
|
|
'folder_unread_counts': await get_folder_unread_counts(user.id, db=db),
|
|
}
|
|
|
|
|
|
############################
|
|
# GetChatUsageStats
|
|
# EXPERIMENTAL: may be removed in future releases
|
|
############################
|
|
|
|
|
|
@router.get('/stats/usage', response_model=ChatUsageStatsListResponse)
|
|
async def get_session_user_chat_usage_stats(
|
|
items_per_page: int | None = 50,
|
|
page: int | None = 1,
|
|
user=Depends(get_verified_user),
|
|
db: AsyncSession = Depends(get_async_session),
|
|
):
|
|
try:
|
|
limit = items_per_page
|
|
skip = (page - 1) * limit
|
|
|
|
result = await Chats.get_chats_by_user_id(user.id, skip=skip, limit=limit, db=db)
|
|
|
|
chats = result.items
|
|
total = result.total
|
|
|
|
chat_stats = []
|
|
for chat in chats:
|
|
messages_map = chat.chat.get('history', {}).get('messages', {})
|
|
message_id = chat.chat.get('history', {}).get('currentId')
|
|
|
|
if messages_map and message_id:
|
|
try:
|
|
history_models = {}
|
|
history_message_count = len(messages_map)
|
|
history_user_messages = []
|
|
history_assistant_messages = []
|
|
|
|
for message in messages_map.values():
|
|
if message.get('role', '') == 'user':
|
|
history_user_messages.append(message)
|
|
elif message.get('role', '') == 'assistant':
|
|
history_assistant_messages.append(message)
|
|
model = message.get('model', None)
|
|
if model:
|
|
if model not in history_models:
|
|
history_models[model] = 0
|
|
history_models[model] += 1
|
|
|
|
average_user_message_content_length = (
|
|
sum(len(message.get('content', '')) for message in history_user_messages)
|
|
/ len(history_user_messages)
|
|
if len(history_user_messages) > 0
|
|
else 0
|
|
)
|
|
average_assistant_message_content_length = (
|
|
sum(len(message.get('content', '')) for message in history_assistant_messages)
|
|
/ len(history_assistant_messages)
|
|
if len(history_assistant_messages) > 0
|
|
else 0
|
|
)
|
|
|
|
response_times = []
|
|
for message in history_assistant_messages:
|
|
user_message_id = message.get('parentId', None)
|
|
if user_message_id and user_message_id in messages_map:
|
|
user_message = messages_map[user_message_id]
|
|
response_time = message.get('timestamp', 0) - user_message.get('timestamp', 0)
|
|
|
|
response_times.append(response_time)
|
|
|
|
average_response_time = sum(response_times) / len(response_times) if len(response_times) > 0 else 0
|
|
|
|
message_list = get_message_list(messages_map, message_id)
|
|
message_count = len(message_list)
|
|
|
|
models = {}
|
|
for message in reversed(message_list):
|
|
if message.get('role') == 'assistant':
|
|
model = message.get('model', None)
|
|
if model:
|
|
if model not in models:
|
|
models[model] = 0
|
|
models[model] += 1
|
|
|
|
annotation = message.get('annotation', {})
|
|
|
|
chat_stats.append(
|
|
{
|
|
'id': chat.id,
|
|
'models': models,
|
|
'message_count': message_count,
|
|
'history_models': history_models,
|
|
'history_message_count': history_message_count,
|
|
'history_user_message_count': len(history_user_messages),
|
|
'history_assistant_message_count': len(history_assistant_messages),
|
|
'average_response_time': average_response_time,
|
|
'average_user_message_content_length': average_user_message_content_length,
|
|
'average_assistant_message_content_length': average_assistant_message_content_length,
|
|
'tags': chat.meta.get('tags', []),
|
|
'last_message_at': message_list[-1].get('timestamp', None),
|
|
'updated_at': chat.updated_at,
|
|
'created_at': chat.created_at,
|
|
}
|
|
)
|
|
except Exception as e:
|
|
pass
|
|
|
|
return ChatUsageStatsListResponse(items=chat_stats, total=total)
|
|
|
|
except Exception as e:
|
|
log.exception(e)
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=ERROR_MESSAGES.DEFAULT())
|
|
|
|
|
|
############################
|
|
# GetChatStatsExport
|
|
############################
|
|
|
|
|
|
CHAT_EXPORT_PAGE_ITEM_COUNT = 10
|
|
|
|
|
|
class ChatStatsExportList(BaseModel):
|
|
type: str = 'chats'
|
|
items: list[ChatStatsExport]
|
|
total: int
|
|
page: int
|
|
|
|
|
|
def _process_chat_for_export(chat) -> ChatStatsExport | None:
|
|
try:
|
|
|
|
def get_message_content_length(message):
|
|
content = message.get('content', '')
|
|
if isinstance(content, str):
|
|
return len(content)
|
|
elif isinstance(content, list):
|
|
return sum(len(item.get('text', '')) for item in content if item.get('type') == 'text')
|
|
return 0
|
|
|
|
messages_map = chat.chat.get('history', {}).get('messages', {})
|
|
message_id = chat.chat.get('history', {}).get('currentId')
|
|
|
|
history_models = {}
|
|
history_message_count = len(messages_map)
|
|
history_user_messages = []
|
|
history_assistant_messages = []
|
|
|
|
export_messages = {}
|
|
for key, message in messages_map.items():
|
|
try:
|
|
content_length = get_message_content_length(message)
|
|
|
|
# Extract rating safely
|
|
rating = message.get('annotation', {}).get('rating')
|
|
tags = message.get('annotation', {}).get('tags')
|
|
|
|
message_stat = MessageStats(
|
|
id=message.get('id'),
|
|
role=message.get('role'),
|
|
model=message.get('model'),
|
|
timestamp=message.get('timestamp'),
|
|
content_length=content_length,
|
|
token_count=None, # Populate if available, e.g. message.get("info", {}).get("token_count")
|
|
rating=rating,
|
|
tags=tags,
|
|
)
|
|
|
|
export_messages[key] = message_stat
|
|
|
|
# --- Aggregation Logic (copied/adapted from usage stats) ---
|
|
role = message.get('role', '')
|
|
if role == 'user':
|
|
history_user_messages.append(message)
|
|
elif role == 'assistant':
|
|
history_assistant_messages.append(message)
|
|
model = message.get('model')
|
|
if model:
|
|
if model not in history_models:
|
|
history_models[model] = 0
|
|
history_models[model] += 1
|
|
except Exception as e:
|
|
log.debug('Error processing message %s: %s', key, e)
|
|
continue
|
|
|
|
# Calculate Averages
|
|
average_user_message_content_length = (
|
|
sum(get_message_content_length(m) for m in history_user_messages) / len(history_user_messages)
|
|
if history_user_messages
|
|
else 0
|
|
)
|
|
|
|
average_assistant_message_content_length = (
|
|
sum(get_message_content_length(m) for m in history_assistant_messages) / len(history_assistant_messages)
|
|
if history_assistant_messages
|
|
else 0
|
|
)
|
|
|
|
# Response Times
|
|
response_times = []
|
|
for message in history_assistant_messages:
|
|
user_message_id = message.get('parentId', None)
|
|
if user_message_id and user_message_id in messages_map:
|
|
user_message = messages_map[user_message_id]
|
|
# Ensure timestamps exist
|
|
t1 = message.get('timestamp')
|
|
t0 = user_message.get('timestamp')
|
|
if t1 and t0:
|
|
response_times.append(t1 - t0)
|
|
|
|
average_response_time = sum(response_times) / len(response_times) if response_times else 0
|
|
|
|
# Current Message List Logic (Main path)
|
|
message_list = get_message_list(messages_map, message_id)
|
|
message_count = len(message_list)
|
|
models = {}
|
|
for message in reversed(message_list):
|
|
if message.get('role') == 'assistant':
|
|
model = message.get('model')
|
|
if model:
|
|
if model not in models:
|
|
models[model] = 0
|
|
models[model] += 1
|
|
|
|
# Construct Aggregate Stats
|
|
stats = AggregateChatStats(
|
|
average_response_time=average_response_time,
|
|
average_user_message_content_length=average_user_message_content_length,
|
|
average_assistant_message_content_length=average_assistant_message_content_length,
|
|
models=models,
|
|
message_count=message_count,
|
|
history_models=history_models,
|
|
history_message_count=history_message_count,
|
|
history_user_message_count=len(history_user_messages),
|
|
history_assistant_message_count=len(history_assistant_messages),
|
|
)
|
|
|
|
# Construct Chat Body
|
|
chat_body = ChatBody(history=ChatHistoryStats(messages=export_messages, currentId=message_id))
|
|
|
|
return ChatStatsExport(
|
|
id=chat.id,
|
|
user_id=chat.user_id,
|
|
created_at=chat.created_at,
|
|
updated_at=chat.updated_at,
|
|
tags=chat.meta.get('tags', []),
|
|
stats=stats,
|
|
chat=chat_body,
|
|
)
|
|
except Exception as e:
|
|
log.exception(f'Error exporting stats for chat {chat.id}: {e}')
|
|
return None
|
|
|
|
|
|
async def calculate_chat_stats(user_id, skip=0, limit=10, filter=None):
|
|
if filter is None:
|
|
filter = {}
|
|
|
|
result = await Chats.get_chats_by_user_id(
|
|
user_id,
|
|
skip=skip,
|
|
limit=limit,
|
|
filter=filter,
|
|
)
|
|
|
|
chat_stats_export_list = []
|
|
for chat in result.items:
|
|
chat_stat = _process_chat_for_export(chat)
|
|
if chat_stat:
|
|
chat_stats_export_list.append(chat_stat)
|
|
|
|
return chat_stats_export_list, result.total
|
|
|
|
|
|
async def generate_chat_stats_jsonl_generator(user_id, filter):
|
|
"""
|
|
Async generator for streaming chat stats export.
|
|
|
|
NOTE: We intentionally do NOT pass a shared db session here. Instead, we let
|
|
each batch create its own short-lived session via get_async_db_context(None).
|
|
This is critical for SQLite in low-resource environments because:
|
|
1. SQLite uses file-level locking
|
|
2. Holding a session open for the entire streaming duration blocks other requests
|
|
3. Short-lived sessions release locks between batches, allowing other operations
|
|
"""
|
|
skip = 0
|
|
limit = CHAT_EXPORT_PAGE_ITEM_COUNT
|
|
|
|
while True:
|
|
# Each batch gets its own session that closes after the query
|
|
result = await Chats.get_chats_by_user_id(
|
|
user_id,
|
|
filter=filter,
|
|
skip=skip,
|
|
limit=limit,
|
|
db=None, # Let get_async_db_context create a fresh session per batch
|
|
)
|
|
if not result.items:
|
|
break
|
|
|
|
for chat in result.items:
|
|
try:
|
|
chat_stat = _process_chat_for_export(chat)
|
|
if chat_stat:
|
|
yield chat_stat.model_dump_json() + '\n'
|
|
except Exception as e:
|
|
log.exception(f'Error processing chat {chat.id}: {e}')
|
|
|
|
skip += limit
|
|
|
|
|
|
@router.get('/stats/export', response_model=ChatStatsExportList)
|
|
async def export_chat_stats(
|
|
request: Request,
|
|
updated_at: int | None = None,
|
|
page: int | None = 1,
|
|
stream: bool = False,
|
|
user=Depends(get_verified_user),
|
|
):
|
|
# Check if the user has permission to share/export chats
|
|
if (user.role != 'admin') and (not await Config.get('ui.enable_community_sharing')):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail=ERROR_MESSAGES.ACCESS_PROHIBITED,
|
|
)
|
|
|
|
try:
|
|
# Fetch chats with date filtering
|
|
filter = {'order_by': 'updated_at', 'direction': 'asc'}
|
|
|
|
if updated_at:
|
|
filter['updated_at'] = updated_at
|
|
|
|
if stream:
|
|
return StreamingResponse(
|
|
generate_chat_stats_jsonl_generator(user.id, filter),
|
|
media_type='application/x-ndjson',
|
|
headers={'Content-Disposition': f'attachment; filename=chat-stats-export-{user.id}.jsonl'},
|
|
)
|
|
else:
|
|
limit = CHAT_EXPORT_PAGE_ITEM_COUNT
|
|
skip = (page - 1) * limit
|
|
|
|
chat_stats_export_list, total = await calculate_chat_stats(user.id, skip, limit, filter)
|
|
|
|
return ChatStatsExportList(items=chat_stats_export_list, total=total, page=page)
|
|
|
|
except Exception as e:
|
|
log.debug('Error exporting chat stats: %s', e)
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=ERROR_MESSAGES.DEFAULT())
|
|
|
|
|
|
############################
|
|
# GetSingleChatStatsExport
|
|
############################
|
|
|
|
|
|
@router.get('/stats/export/{chat_id}', response_model=ChatStatsExport | None)
|
|
async def export_single_chat_stats(
|
|
request: Request,
|
|
chat_id: str,
|
|
user=Depends(get_verified_user),
|
|
db: AsyncSession = Depends(get_async_session),
|
|
):
|
|
"""
|
|
Export stats for exactly one chat by ID.
|
|
Returns ChatStatsExport for the specified chat.
|
|
"""
|
|
# Check if the user has permission to share/export chats
|
|
if (user.role != 'admin') and (not await Config.get('ui.enable_community_sharing')):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail=ERROR_MESSAGES.ACCESS_PROHIBITED,
|
|
)
|
|
|
|
try:
|
|
chat = await Chats.get_chat_by_id(chat_id, db=db)
|
|
|
|
if not chat:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=ERROR_MESSAGES.NOT_FOUND,
|
|
)
|
|
|
|
# Verify the chat belongs to the user (unless admin)
|
|
if chat.user_id != user.id and user.role != 'admin':
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail=ERROR_MESSAGES.ACCESS_PROHIBITED,
|
|
)
|
|
|
|
# Process the chat for export (pure computation, no DB)
|
|
chat_stats = _process_chat_for_export(chat)
|
|
|
|
if not chat_stats:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail='Failed to process chat stats',
|
|
)
|
|
|
|
return chat_stats
|
|
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
log.debug('Error exporting single chat stats: %s', e)
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=ERROR_MESSAGES.DEFAULT())
|
|
|
|
|
|
@router.delete('/', response_model=bool)
|
|
async def delete_all_user_chats(
|
|
request: Request,
|
|
user=Depends(get_verified_user),
|
|
db: AsyncSession = Depends(get_async_session),
|
|
):
|
|
if user.role == 'user' and not await has_permission(user.id, 'chat.delete', await Config.get('user.permissions')):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail=ERROR_MESSAGES.ACCESS_PROHIBITED,
|
|
)
|
|
|
|
result = await Chats.delete_chats_by_user_id(user.id, db=db)
|
|
if result:
|
|
await publish_event(
|
|
request,
|
|
EVENTS.CHAT_DELETED_ALL,
|
|
actor=user,
|
|
subject_id=user.id,
|
|
subject_type='user',
|
|
)
|
|
return result
|
|
|
|
|
|
############################
|
|
# GetUserChatList
|
|
############################
|
|
|
|
|
|
@router.get('/list/user/{user_id}', response_model=list[ChatTitleIdResponse])
|
|
async def get_user_chat_list_by_user_id(
|
|
request: Request,
|
|
user_id: str,
|
|
page: int | None = None,
|
|
query: str | None = None,
|
|
order_by: str | None = None,
|
|
direction: str | None = None,
|
|
user=Depends(get_admin_user),
|
|
db: AsyncSession = Depends(get_async_session),
|
|
):
|
|
"""List chat summaries for a given user (admin-only endpoint)."""
|
|
if not ENABLE_ADMIN_CHAT_ACCESS:
|
|
raise HTTPException(status.HTTP_401_UNAUTHORIZED, detail=ERROR_MESSAGES.ACCESS_PROHIBITED)
|
|
|
|
effective_page = page if page is not None else 1
|
|
limit = 60
|
|
skip = (effective_page - 1) * limit
|
|
|
|
filter = {}
|
|
if query:
|
|
filter['query'] = query
|
|
if order_by:
|
|
filter['order_by'] = order_by
|
|
if direction:
|
|
filter['direction'] = direction
|
|
|
|
chats = await Chats.get_chat_list_by_user_id(
|
|
user_id, include_archived=True, filter=filter, skip=skip, limit=limit, db=db
|
|
)
|
|
return await add_active_state_to_chat_list(request, chats)
|
|
|
|
|
|
############################
|
|
# CreateNewChat
|
|
############################
|
|
|
|
|
|
@router.post('/new', response_model=ChatResponse | None)
|
|
async def create_new_chat(
|
|
request: Request,
|
|
form_data: ChatForm,
|
|
user=Depends(get_verified_user),
|
|
db: AsyncSession = Depends(get_async_session),
|
|
):
|
|
if form_data.folder_id is not None and not await has_folder_write_access(user.id, form_data.folder_id, db=db):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=ERROR_MESSAGES.NOT_FOUND,
|
|
)
|
|
|
|
try:
|
|
chat = await Chats.insert_new_chat(str(uuid4()), user.id, form_data, db=db)
|
|
await publish_event(
|
|
request,
|
|
EVENTS.CHAT_CREATED,
|
|
actor=user,
|
|
subject_id=chat.id,
|
|
data={'title': chat.title, 'folder_id': chat.folder_id},
|
|
)
|
|
return ChatResponse.model_validate(chat, from_attributes=True)
|
|
except Exception as e:
|
|
log.exception(e)
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=ERROR_MESSAGES.DEFAULT())
|
|
|
|
|
|
############################
|
|
# ImportChats
|
|
############################
|
|
|
|
|
|
@router.post('/import', response_model=list[ChatResponse])
|
|
async def import_chats(
|
|
request: Request,
|
|
form_data: ChatsImportForm,
|
|
user=Depends(get_verified_user),
|
|
db: AsyncSession = Depends(get_async_session),
|
|
):
|
|
await require_chat_import_permission(request, user, db)
|
|
|
|
try:
|
|
chats = await Chats.import_chats(user.id, form_data.chats, db=db)
|
|
await publish_event(
|
|
request,
|
|
EVENTS.CHAT_IMPORTED,
|
|
actor=user,
|
|
subject_type='chat.import',
|
|
data={'count': len(chats), 'chat_ids': [chat.id for chat in chats]},
|
|
)
|
|
return chats
|
|
except Exception as e:
|
|
log.exception(e)
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=ERROR_MESSAGES.DEFAULT())
|
|
|
|
|
|
############################
|
|
# ChatConfig
|
|
############################
|
|
|
|
|
|
@router.get('/config', response_model=ChatConfigForm)
|
|
async def get_chat_config(user=Depends(get_admin_user)):
|
|
return await get_chat_config_values()
|
|
|
|
|
|
@router.post('/config', response_model=ChatConfigForm)
|
|
async def set_chat_config(form_data: ChatConfigForm, user=Depends(get_admin_user)):
|
|
threshold = max(1, int(form_data.CONTEXT_COMPACTION_TOKEN_THRESHOLD))
|
|
token_cap = max(1, int(form_data.CONTEXT_COMPACTION_TOKEN_CAP or threshold))
|
|
retention_percentage = min(50, max(10, int(form_data.CONTEXT_COMPACTION_RETENTION_PERCENTAGE)))
|
|
await Config.upsert(
|
|
chat_config_updates(
|
|
{
|
|
**form_data.model_dump(),
|
|
'CONTEXT_COMPACTION_MODEL': form_data.CONTEXT_COMPACTION_MODEL or '',
|
|
'CONTEXT_COMPACTION_TOKEN_THRESHOLD': threshold,
|
|
'CONTEXT_COMPACTION_TOKEN_CAP': token_cap,
|
|
'CONTEXT_COMPACTION_RETENTION_PERCENTAGE': retention_percentage,
|
|
}
|
|
)
|
|
)
|
|
return await get_chat_config_values()
|
|
|
|
|
|
############################
|
|
# GetChats
|
|
############################
|
|
|
|
|
|
@router.get('/search', response_model=list[ChatTitleIdResponse])
|
|
async def search_user_chats(
|
|
request: Request,
|
|
text: str,
|
|
page: int | None = None,
|
|
user=Depends(get_verified_user),
|
|
db: AsyncSession = Depends(get_async_session),
|
|
):
|
|
if page is None:
|
|
page = 1
|
|
|
|
limit = 60
|
|
skip = (page - 1) * limit
|
|
|
|
search_text = chat_search_content_text(text)
|
|
chat_list = []
|
|
for chat in await Chats.get_chats_by_user_id_and_search_text(user.id, text, skip=skip, limit=limit, db=db):
|
|
# Explicit fields: model_dump() would deep-copy the entire chat blob per row
|
|
chat_list.append(
|
|
ChatTitleIdResponse(
|
|
id=chat.id,
|
|
title=chat.title,
|
|
updated_at=chat.updated_at,
|
|
created_at=chat.created_at,
|
|
last_read_at=chat.last_read_at,
|
|
snippet=chat_search_snippet(chat.chat, search_text),
|
|
)
|
|
)
|
|
|
|
# Delete tag if no chat is found
|
|
words = text.strip().split(' ')
|
|
if page == 1 and len(words) == 1 and words[0].startswith('tag:'):
|
|
tag_id = words[0].replace('tag:', '')
|
|
if len(chat_list) == 0:
|
|
if await Tags.get_tag_by_name_and_user_id(tag_id, user.id, db=db):
|
|
log.debug('deleting tag: %s', tag_id)
|
|
await Tags.delete_tag_by_name_and_user_id(tag_id, user.id, db=db)
|
|
|
|
return await add_active_state_to_chat_list(request, chat_list)
|
|
|
|
|
|
############################
|
|
# GetChatsByFolderId
|
|
############################
|
|
|
|
|
|
@router.get('/folder/{folder_id}', response_model=list[ChatResponse])
|
|
async def get_chats_by_folder_id(
|
|
folder_id: str, user=Depends(get_verified_user), db: AsyncSession = Depends(get_async_session)
|
|
):
|
|
folder_ids = [folder_id]
|
|
children_folders = await Folders.get_children_folders_by_id_and_user_id(folder_id, user.id, db=db)
|
|
if children_folders:
|
|
folder_ids.extend([folder.id for folder in children_folders])
|
|
|
|
return [
|
|
ChatResponse.model_validate(chat, from_attributes=True)
|
|
for chat in await Chats.get_chats_by_folder_ids_and_user_id(folder_ids, user.id, db=db)
|
|
]
|
|
|
|
|
|
@router.get('/folder/{folder_id}/list', response_model=list[ChatTitleIdResponse])
|
|
async def get_chat_list_by_folder_id(
|
|
request: Request,
|
|
folder_id: str,
|
|
page: int | None = 1,
|
|
sort_by: str = 'unread_updated_at',
|
|
sort_dir: str = 'desc',
|
|
user=Depends(get_verified_user),
|
|
db: AsyncSession = Depends(get_async_session),
|
|
):
|
|
try:
|
|
limit = 10
|
|
skip = (page - 1) * limit
|
|
|
|
chats = await Chats.get_chats_by_folder_id_and_user_id(
|
|
folder_id,
|
|
user.id,
|
|
skip=skip,
|
|
limit=limit,
|
|
sort_by=sort_by,
|
|
sort_dir=sort_dir,
|
|
db=db,
|
|
)
|
|
return await add_active_state_to_chat_list(request, chats)
|
|
|
|
except Exception as e:
|
|
log.exception(e)
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=ERROR_MESSAGES.DEFAULT())
|
|
|
|
|
|
############################
|
|
# GetPinnedChats
|
|
############################
|
|
|
|
|
|
@router.get('/pinned', response_model=list[ChatTitleIdResponse])
|
|
async def get_user_pinned_chats(
|
|
request: Request, user=Depends(get_verified_user), db: AsyncSession = Depends(get_async_session)
|
|
):
|
|
chats = await Chats.get_pinned_chats_by_user_id(user.id, db=db)
|
|
return await add_active_state_to_chat_list(request, chats)
|
|
|
|
|
|
############################
|
|
# GetChats
|
|
############################
|
|
|
|
CHAT_EXPORT_BATCH_SIZE = 100
|
|
|
|
|
|
async def generate_chat_export_ndjson(user_id: str):
|
|
"""
|
|
Async generator that streams all user chats as NDJSON (one JSON object per line).
|
|
|
|
Uses short-lived DB sessions per batch to avoid holding locks for the
|
|
entire duration, which is critical for SQLite environments.
|
|
"""
|
|
skip = 0
|
|
|
|
while True:
|
|
result = await Chats.get_chats_by_user_id(
|
|
user_id,
|
|
skip=skip,
|
|
limit=CHAT_EXPORT_BATCH_SIZE,
|
|
db=None,
|
|
)
|
|
if not result.items:
|
|
break
|
|
|
|
for chat in result.items:
|
|
try:
|
|
yield ChatResponse.model_validate(chat, from_attributes=True).model_dump_json() + '\n'
|
|
except Exception as e:
|
|
log.exception(f'Error serializing chat {chat.id}: {e}')
|
|
|
|
if len(result.items) < CHAT_EXPORT_BATCH_SIZE:
|
|
break
|
|
|
|
skip += CHAT_EXPORT_BATCH_SIZE
|
|
|
|
|
|
@router.get('/all')
|
|
async def get_user_chats(user=Depends(get_verified_user)):
|
|
return StreamingResponse(
|
|
generate_chat_export_ndjson(user.id),
|
|
media_type='application/x-ndjson',
|
|
)
|
|
|
|
|
|
############################
|
|
# GetArchivedChats
|
|
############################
|
|
|
|
|
|
@router.get('/all/archived', response_model=list[ChatResponse])
|
|
async def get_user_archived_chats(user=Depends(get_verified_user), db: AsyncSession = Depends(get_async_session)):
|
|
return [
|
|
ChatResponse.model_validate(chat, from_attributes=True)
|
|
for chat in await Chats.get_archived_chats_by_user_id(user.id, db=db)
|
|
]
|
|
|
|
|
|
############################
|
|
# GetAllTags
|
|
############################
|
|
|
|
|
|
@router.get('/all/tags', response_model=list[TagModel])
|
|
async def get_all_user_tags(user=Depends(get_verified_user), db: AsyncSession = Depends(get_async_session)):
|
|
try:
|
|
tags = await Tags.get_tags_by_user_id(user.id, db=db)
|
|
return tags
|
|
except Exception as e:
|
|
log.exception(e)
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=ERROR_MESSAGES.DEFAULT())
|
|
|
|
|
|
############################
|
|
# GetAllChatsInDB
|
|
############################
|
|
|
|
|
|
@router.get('/all/db', response_model=list[ChatResponse])
|
|
async def get_all_user_chats_in_db(user=Depends(get_admin_user), db: AsyncSession = Depends(get_async_session)):
|
|
if not ENABLE_ADMIN_EXPORT:
|
|
raise HTTPException(status.HTTP_401_UNAUTHORIZED, detail=ERROR_MESSAGES.ACCESS_PROHIBITED)
|
|
return [ChatResponse.model_validate(chat, from_attributes=True) for chat in await Chats.get_chats(db=db)]
|
|
|
|
|
|
############################
|
|
# GetArchivedChats
|
|
############################
|
|
|
|
|
|
@router.get('/archived', response_model=list[ChatTitleIdResponse])
|
|
async def get_archived_session_user_chat_list(
|
|
request: Request,
|
|
page: int | None = None,
|
|
query: str | None = None,
|
|
order_by: str | None = None,
|
|
direction: str | None = None,
|
|
user=Depends(get_verified_user),
|
|
db: AsyncSession = Depends(get_async_session),
|
|
):
|
|
if page is None:
|
|
page = 1
|
|
|
|
limit = 60
|
|
skip = (page - 1) * limit
|
|
|
|
filter = {}
|
|
if query:
|
|
filter['query'] = query
|
|
if order_by:
|
|
filter['order_by'] = order_by
|
|
if direction:
|
|
filter['direction'] = direction
|
|
|
|
chats = await Chats.get_archived_chat_list_by_user_id(
|
|
user.id,
|
|
filter=filter,
|
|
skip=skip,
|
|
limit=limit,
|
|
db=db,
|
|
)
|
|
return await add_active_state_to_chat_list(request, chats)
|
|
|
|
|
|
############################
|
|
# GetArchivedChatsCount
|
|
############################
|
|
|
|
|
|
@router.get('/archived/count', response_model=int)
|
|
async def get_archived_session_user_chat_count(
|
|
user=Depends(get_verified_user),
|
|
db: AsyncSession = Depends(get_async_session),
|
|
):
|
|
return await Chats.count_archived_chats_by_user_id(user.id, db=db)
|
|
|
|
|
|
############################
|
|
# ArchiveAllChats
|
|
############################
|
|
|
|
|
|
@router.post('/archive/all', response_model=bool)
|
|
async def archive_all_chats(
|
|
request: Request, user=Depends(get_verified_user), db: AsyncSession = Depends(get_async_session)
|
|
):
|
|
result = await Chats.archive_all_chats_by_user_id(user.id, db=db)
|
|
if result:
|
|
await publish_event(request, EVENTS.CHAT_ARCHIVED, actor=user, subject_id=user.id, subject_type='user')
|
|
return result
|
|
|
|
|
|
############################
|
|
# UnarchiveAllChats
|
|
############################
|
|
|
|
|
|
@router.post('/unarchive/all', response_model=bool)
|
|
async def unarchive_all_chats(
|
|
request: Request, user=Depends(get_verified_user), db: AsyncSession = Depends(get_async_session)
|
|
):
|
|
result = await Chats.unarchive_all_chats_by_user_id(user.id, db=db)
|
|
if result:
|
|
await publish_event(request, EVENTS.CHAT_UNARCHIVED, actor=user, subject_id=user.id, subject_type='user')
|
|
return result
|
|
|
|
|
|
############################
|
|
# UnshareAllChats
|
|
############################
|
|
|
|
|
|
@router.delete('/share/all', response_model=bool)
|
|
async def unshare_all_chats(
|
|
request: Request, user=Depends(get_verified_user), db: AsyncSession = Depends(get_async_session)
|
|
):
|
|
# Collect chat_ids that have shares so we can clear share_id and access grants
|
|
shared_list = await SharedChats.get_by_user_id(user.id, db=db)
|
|
chat_ids = [s.chat_id for s in shared_list]
|
|
|
|
# Delete all shared_chat rows for this user
|
|
result = await SharedChats.delete_all_by_user_id(user.id, db=db)
|
|
|
|
# Clear share_id on the original chats and remove access grants
|
|
for chat_id in chat_ids:
|
|
await Chats.update_chat_share_id_by_id(chat_id, None, db=db)
|
|
await AccessGrants.set_access_grants('shared_chat', chat_id, [], db=db)
|
|
|
|
if result:
|
|
await publish_event(
|
|
request,
|
|
EVENTS.CHAT_UNSHARED,
|
|
actor=user,
|
|
subject_id=user.id,
|
|
subject_type='user',
|
|
data={'count': len(chat_ids), 'chat_ids': chat_ids},
|
|
)
|
|
return result
|
|
|
|
|
|
@router.get('/shared', response_model=list[SharedChatResponse])
|
|
async def get_shared_session_user_chat_list(
|
|
page: int | None = None,
|
|
query: str | None = None,
|
|
order_by: str | None = None,
|
|
direction: str | None = None,
|
|
user=Depends(get_verified_user),
|
|
db: AsyncSession = Depends(get_async_session),
|
|
):
|
|
if page is None:
|
|
page = 1
|
|
|
|
limit = 60
|
|
skip = (page - 1) * limit
|
|
|
|
filter = {}
|
|
if query:
|
|
filter['query'] = query
|
|
if order_by:
|
|
filter['order_by'] = order_by
|
|
if direction:
|
|
filter['direction'] = direction
|
|
|
|
return await SharedChats.get_by_user_id(
|
|
user.id,
|
|
filter=filter,
|
|
skip=skip,
|
|
limit=limit,
|
|
db=db,
|
|
)
|
|
|
|
|
|
############################
|
|
# GetSharedChatById
|
|
############################
|
|
|
|
|
|
@router.get('/share/{share_id}', response_model=ChatResponse | None)
|
|
async def get_shared_chat_by_id(
|
|
share_id: str, user=Depends(get_optional_verified_user), db: AsyncSession = Depends(get_async_session)
|
|
):
|
|
shared = await SharedChats.get_by_id(share_id, db=db)
|
|
if shared:
|
|
if await is_open_shared_chat(shared, db=db) or (
|
|
user is not None and await can_read_shared_chat(user, shared, db=db)
|
|
):
|
|
chat = await Chats.get_chat_by_share_id(share_id, db=db)
|
|
if chat:
|
|
return ChatResponse.model_validate(chat, from_attributes=True)
|
|
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail=ERROR_MESSAGES.ACCESS_PROHIBITED if user else ERROR_MESSAGES.INVALID_TOKEN,
|
|
)
|
|
|
|
# Fallback: admins can also access any chat directly by chat ID
|
|
chat = None
|
|
if user is not None and user.role == 'admin' and ENABLE_ADMIN_CHAT_ACCESS:
|
|
chat = await Chats.get_chat_by_id(share_id, db=db)
|
|
if chat:
|
|
return ChatResponse.model_validate(chat, from_attributes=True)
|
|
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail=ERROR_MESSAGES.NOT_FOUND)
|
|
|
|
|
|
############################
|
|
# GetChatsByTags
|
|
############################
|
|
|
|
|
|
class TagForm(BaseModel):
|
|
name: str
|
|
|
|
|
|
class TagFilterForm(TagForm):
|
|
skip: int | None = 0
|
|
limit: int | None = 50
|
|
|
|
|
|
@router.post('/tags', response_model=list[ChatTitleIdResponse])
|
|
async def get_user_chat_list_by_tag_name(
|
|
request: Request,
|
|
form_data: TagFilterForm,
|
|
user=Depends(get_verified_user),
|
|
db: AsyncSession = Depends(get_async_session),
|
|
):
|
|
chats = await Chats.get_chat_list_by_user_id_and_tag_name(
|
|
user.id, form_data.name, form_data.skip, form_data.limit, db=db
|
|
)
|
|
if len(chats) == 0:
|
|
await Tags.delete_tag_by_name_and_user_id(form_data.name, user.id, db=db)
|
|
|
|
return await add_active_state_to_chat_list(request, chats)
|
|
|
|
|
|
############################
|
|
# CompactChat
|
|
############################
|
|
|
|
|
|
@router.post('/{id}/compact')
|
|
async def compact_chat_by_id(
|
|
request: Request,
|
|
id: str,
|
|
form_data: CompactChatForm | None = None,
|
|
user=Depends(get_verified_user),
|
|
db: AsyncSession = Depends(get_async_session),
|
|
):
|
|
chat = await Chats.get_chat_by_id_and_user_id(id, user.id, db=db)
|
|
if not chat:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail=ERROR_MESSAGES.NOT_FOUND)
|
|
|
|
if await has_active_tasks(request.app.state.redis, id):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_409_CONFLICT,
|
|
detail='Wait for the current response to finish before compacting.',
|
|
)
|
|
|
|
if not request.app.state.MODELS:
|
|
await get_all_models(request, user=user)
|
|
|
|
history = (chat.chat or {}).get('history') or {}
|
|
messages_map = await Chats.get_messages_map_by_chat_id(id)
|
|
current_message_id = chat.current_message_id or history.get('currentId')
|
|
message_list = get_message_list(messages_map or history.get('messages') or {}, current_message_id)
|
|
model_id = (form_data.model if form_data else None) or next(
|
|
(message.get('model') for message in reversed(message_list) if message.get('model')),
|
|
None,
|
|
)
|
|
|
|
if not model_id:
|
|
chat_models = (chat.chat or {}).get('models') or []
|
|
model_id = chat_models[0] if chat_models else None
|
|
if not model_id:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail='No model found for context compaction.')
|
|
|
|
result = await compact_chat_branch(request, user, chat, model_id, request.app.state.MODELS)
|
|
result['context_usage'] = await get_chat_context_usage(chat, model_id)
|
|
if result.get('compacted'):
|
|
await publish_event(
|
|
request,
|
|
EVENTS.CHAT_COMPACTED,
|
|
actor=user,
|
|
subject_id=id,
|
|
data={'dropped_messages': result.get('dropped_messages')},
|
|
)
|
|
return result
|
|
|
|
|
|
############################
|
|
# GetChatById
|
|
############################
|
|
|
|
|
|
@router.get('/{id}', response_model=ChatResponse | None)
|
|
async def get_chat_by_id(
|
|
id: str,
|
|
request: Request,
|
|
user=Depends(get_verified_user),
|
|
db: AsyncSession = Depends(get_async_session),
|
|
):
|
|
chat = await Chats.get_chat_by_id_and_user_id(id, user.id, db=db)
|
|
|
|
if not chat and user.role == 'admin':
|
|
candidate = await Chats.get_chat_by_id(id, db=db)
|
|
if ENABLE_ADMIN_CHAT_ACCESS or (candidate and is_internal_chat(candidate.meta)):
|
|
chat = candidate
|
|
|
|
# Access explicitly granted to this user applies to admins too, so an admin
|
|
# does not lose a chat shared with them when ENABLE_ADMIN_CHAT_ACCESS is off.
|
|
if not chat:
|
|
has_grant = await AccessGrants.has_access(
|
|
user_id=user.id,
|
|
resource_type='shared_chat',
|
|
resource_id=id,
|
|
permission='read',
|
|
db=db,
|
|
)
|
|
if has_grant:
|
|
chat = await Chats.get_chat_by_id(id, db=db)
|
|
|
|
# Check folder-based access (shared folders)
|
|
if not chat:
|
|
candidate = await Chats.get_chat_by_id(id, db=db)
|
|
if candidate and candidate.folder_id:
|
|
folder = await Folders.get_folder_by_id(candidate.folder_id, db=db)
|
|
if folder and await has_folder_access(user.id, folder, 'read', db):
|
|
chat = candidate
|
|
|
|
if chat:
|
|
data = ChatResponse.model_validate(chat, from_attributes=True).model_dump()
|
|
data = overlay_response_streams(
|
|
data,
|
|
await get_response_streams_by_chat_id(request.app.state.redis, id),
|
|
)
|
|
data['context_usage'] = await get_chat_context_usage(chat)
|
|
return data
|
|
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail=ERROR_MESSAGES.NOT_FOUND)
|
|
|
|
|
|
############################
|
|
# UpdateChatById
|
|
############################
|
|
|
|
|
|
@router.post('/{id}', response_model=ChatResponse | None)
|
|
async def update_chat_by_id(
|
|
request: Request,
|
|
id: str,
|
|
form_data: ChatForm,
|
|
user=Depends(get_verified_user),
|
|
db: AsyncSession = Depends(get_async_session),
|
|
):
|
|
chat = await Chats.get_chat_by_id_and_user_id(id, user.id, db=db)
|
|
if chat:
|
|
updated_chat = {**chat.chat, **form_data.chat}
|
|
if 'history' in form_data.chat:
|
|
updated_chat['history'] = Chats.merge_history(
|
|
chat.chat.get('history'),
|
|
form_data.chat.get('history'),
|
|
)
|
|
|
|
touch = 'history' in form_data.chat or 'messages' in form_data.chat
|
|
chat = await Chats.update_chat_by_id(id, updated_chat, db=db, touch=touch)
|
|
if form_data.variables is not None:
|
|
chat = (
|
|
await Chats.update_chat_variables_by_id(
|
|
id,
|
|
form_data.variables,
|
|
db=db,
|
|
touch=False,
|
|
)
|
|
or chat
|
|
)
|
|
|
|
# Reconcile chat_message rows without inferring deletes from missing IDs.
|
|
# Message deletion has its own endpoint below.
|
|
messages = (updated_chat.get('history') or {}).get('messages') or {}
|
|
if messages:
|
|
await Chats.reconcile_messages_by_chat_id(id, user.id, messages)
|
|
|
|
await publish_event(
|
|
request,
|
|
EVENTS.CHAT_UPDATED,
|
|
actor=user,
|
|
subject_id=id,
|
|
data={'title': chat.title},
|
|
)
|
|
return ChatResponse.model_validate(chat, from_attributes=True)
|
|
else:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail=ERROR_MESSAGES.ACCESS_PROHIBITED,
|
|
)
|
|
|
|
|
|
############################
|
|
# UpdateChatMessageById
|
|
############################
|
|
class MessageForm(BaseModel):
|
|
content: str
|
|
|
|
|
|
@router.post('/{id}/messages/{message_id}', response_model=ChatResponse | None)
|
|
async def update_chat_message_by_id(
|
|
request: Request,
|
|
id: str,
|
|
message_id: str,
|
|
form_data: MessageForm,
|
|
user=Depends(get_verified_user),
|
|
db: AsyncSession = Depends(get_async_session),
|
|
):
|
|
chat = await Chats.get_chat_by_id(id, db=db)
|
|
|
|
if not chat:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail=ERROR_MESSAGES.ACCESS_PROHIBITED,
|
|
)
|
|
|
|
if chat.user_id != user.id and user.role != 'admin':
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail=ERROR_MESSAGES.ACCESS_PROHIBITED,
|
|
)
|
|
|
|
chat = await Chats.upsert_message_to_chat_by_id_and_message_id(
|
|
id,
|
|
message_id,
|
|
{
|
|
'content': form_data.content,
|
|
},
|
|
)
|
|
|
|
event_emitter = await get_event_emitter(
|
|
{
|
|
'user_id': chat.user_id,
|
|
'chat_id': id,
|
|
'message_id': message_id,
|
|
},
|
|
False,
|
|
)
|
|
|
|
if event_emitter:
|
|
await event_emitter(
|
|
{
|
|
'type': 'chat:message',
|
|
'data': {
|
|
'chat_id': id,
|
|
'message_id': message_id,
|
|
'content': form_data.content,
|
|
},
|
|
}
|
|
)
|
|
|
|
await publish_event(
|
|
request,
|
|
EVENTS.MESSAGE_UPDATED,
|
|
actor=user,
|
|
subject_id=message_id,
|
|
data={'chat_id': id, 'content_preview': form_data.content[:300]},
|
|
)
|
|
return ChatResponse.model_validate(chat, from_attributes=True)
|
|
|
|
|
|
@router.delete('/{id}/messages/{message_id}', response_model=ChatResponse | None)
|
|
async def delete_chat_message_by_id(
|
|
request: Request,
|
|
id: str,
|
|
message_id: str,
|
|
user=Depends(get_verified_user),
|
|
db: AsyncSession = Depends(get_async_session),
|
|
):
|
|
chat = await Chats.get_chat_by_id(id, db=db)
|
|
|
|
if not chat:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail=ERROR_MESSAGES.ACCESS_PROHIBITED,
|
|
)
|
|
|
|
if chat.user_id != user.id and user.role != 'admin':
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail=ERROR_MESSAGES.ACCESS_PROHIBITED,
|
|
)
|
|
|
|
chat = await Chats.delete_message_from_chat_by_id_and_message_id(id, message_id)
|
|
if not chat:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail=ERROR_MESSAGES.ACCESS_PROHIBITED,
|
|
)
|
|
|
|
await publish_event(
|
|
request,
|
|
EVENTS.MESSAGE_DELETED,
|
|
actor=user,
|
|
subject_id=message_id,
|
|
data={'chat_id': id},
|
|
)
|
|
return ChatResponse.model_validate(chat, from_attributes=True)
|
|
|
|
|
|
############################
|
|
# SendChatMessageEventById
|
|
############################
|
|
class EventForm(BaseModel):
|
|
type: str
|
|
data: dict
|
|
|
|
|
|
@router.post('/{id}/messages/{message_id}/event', response_model=bool | None)
|
|
async def send_chat_message_event_by_id(
|
|
request: Request,
|
|
id: str,
|
|
message_id: str,
|
|
form_data: EventForm,
|
|
user=Depends(get_verified_user),
|
|
db: AsyncSession = Depends(get_async_session),
|
|
):
|
|
chat = await Chats.get_chat_by_id(id, db=db)
|
|
|
|
if not chat:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail=ERROR_MESSAGES.ACCESS_PROHIBITED,
|
|
)
|
|
|
|
if chat.user_id != user.id and user.role != 'admin':
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail=ERROR_MESSAGES.ACCESS_PROHIBITED,
|
|
)
|
|
|
|
event_emitter = await get_event_emitter(
|
|
{
|
|
'user_id': chat.user_id,
|
|
'chat_id': id,
|
|
'message_id': message_id,
|
|
}
|
|
)
|
|
|
|
try:
|
|
if event_emitter:
|
|
await event_emitter(form_data.model_dump())
|
|
else:
|
|
return False
|
|
await publish_event(
|
|
request,
|
|
EVENTS.MESSAGE_EVENT_RECEIVED,
|
|
actor=user,
|
|
subject_id=message_id,
|
|
data={'chat_id': id, 'event_type': form_data.type},
|
|
)
|
|
return True
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
############################
|
|
# DeleteChatById
|
|
############################
|
|
|
|
|
|
@router.delete('/{id}', response_model=bool)
|
|
async def delete_chat_by_id(
|
|
request: Request,
|
|
id: str,
|
|
user=Depends(get_verified_user),
|
|
db: AsyncSession = Depends(get_async_session),
|
|
):
|
|
# Authorize before any side effect: cancelling a chat's in-flight tasks must
|
|
# not be reachable for a chat the caller may not delete.
|
|
if user.role == 'admin':
|
|
chat = await Chats.get_chat_by_id(id, db=db)
|
|
else:
|
|
if not await has_permission(user.id, 'chat.delete', await Config.get('user.permissions')):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail=ERROR_MESSAGES.ACCESS_PROHIBITED,
|
|
)
|
|
chat = await Chats.get_chat_by_id_and_user_id(id, user.id, db=db)
|
|
|
|
if not chat:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=ERROR_MESSAGES.NOT_FOUND,
|
|
)
|
|
|
|
# Cancel any in-flight LLM tasks (streaming, title/tags generation) before
|
|
# deleting the chat to prevent orphaned requests.
|
|
await stop_item_tasks(request.app.state.redis, id)
|
|
await Chats.delete_orphan_tags_for_user(chat.meta.get('tags', []), user.id, threshold=1, db=db)
|
|
|
|
# Cascade to internal child chats spawned from this one.
|
|
for child_id in await Chats.get_internal_chat_ids_by_parent_id(id, chat.user_id):
|
|
await stop_item_tasks(request.app.state.redis, child_id)
|
|
await Chats.delete_chat_by_id_and_user_id(child_id, chat.user_id)
|
|
|
|
if user.role == 'admin':
|
|
result = await Chats.delete_chat_by_id(id, db=db)
|
|
else:
|
|
result = await Chats.delete_chat_by_id_and_user_id(id, user.id, db=db)
|
|
|
|
if result:
|
|
await publish_event(
|
|
request,
|
|
EVENTS.CHAT_DELETED,
|
|
actor=user,
|
|
subject_id=id,
|
|
data={'owner_id': chat.user_id},
|
|
)
|
|
return result
|
|
|
|
|
|
############################
|
|
# GetPinnedStatusById
|
|
############################
|
|
|
|
|
|
@router.get('/{id}/pinned', response_model=bool | None)
|
|
async def get_pinned_status_by_id(
|
|
id: str, user=Depends(get_verified_user), db: AsyncSession = Depends(get_async_session)
|
|
):
|
|
chat = await Chats.get_chat_by_id_and_user_id(id, user.id, db=db)
|
|
if chat:
|
|
return chat.pinned
|
|
else:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail=ERROR_MESSAGES.DEFAULT())
|
|
|
|
|
|
############################
|
|
# PinChatById
|
|
############################
|
|
|
|
|
|
@router.post('/{id}/pin', response_model=ChatResponse | None)
|
|
async def pin_chat_by_id(
|
|
request: Request, id: str, user=Depends(get_verified_user), db: AsyncSession = Depends(get_async_session)
|
|
):
|
|
chat = await Chats.get_chat_by_id_and_user_id(id, user.id, db=db)
|
|
if chat:
|
|
chat = await Chats.toggle_chat_pinned_by_id(id, db=db)
|
|
await publish_event(
|
|
request,
|
|
EVENTS.CHAT_PINNED if chat.pinned else EVENTS.CHAT_UNPINNED,
|
|
actor=user,
|
|
subject_id=id,
|
|
subject_type='chat',
|
|
)
|
|
return chat
|
|
else:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail=ERROR_MESSAGES.DEFAULT())
|
|
|
|
|
|
############################
|
|
# CloneChat
|
|
############################
|
|
|
|
|
|
class CloneForm(BaseModel):
|
|
title: str | None = None
|
|
|
|
|
|
class ForkForm(BaseModel):
|
|
message_id: str | None = None
|
|
|
|
|
|
@router.post('/{id}/fork', response_model=ChatResponse | None)
|
|
async def fork_chat_by_id(
|
|
request: Request,
|
|
id: str,
|
|
form_data: ForkForm | None = None,
|
|
user=Depends(get_verified_user),
|
|
db: AsyncSession = Depends(get_async_session),
|
|
):
|
|
await require_chat_import_permission(request, user, db)
|
|
|
|
chat = await Chats.get_chat_by_id_and_user_id(id, user.id, db=db)
|
|
if not chat:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail=ERROR_MESSAGES.DEFAULT())
|
|
|
|
if await has_active_tasks(request.app.state.redis, id):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_409_CONFLICT,
|
|
detail='Wait for the current response to finish before forking.',
|
|
)
|
|
|
|
history = (chat.chat or {}).get('history') or {}
|
|
messages_map = await Chats.get_messages_map_by_chat_id(id) or history.get('messages') or {}
|
|
if any(
|
|
message.get('role') == 'assistant' and message.get('done') is False
|
|
for message in messages_map.values()
|
|
if isinstance(message, dict)
|
|
):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_409_CONFLICT,
|
|
detail='Wait for the current response to finish before forking.',
|
|
)
|
|
|
|
source_message_id = (
|
|
(form_data.message_id if form_data else None) or chat.current_message_id or history.get('currentId')
|
|
)
|
|
if not source_message_id:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail='chat has no messages to fork')
|
|
|
|
try:
|
|
fork_history, fork_messages = build_fork_history(messages_map, source_message_id)
|
|
except ValueError as exc:
|
|
detail = str(exc)
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND if detail == 'message not found' else status.HTTP_400_BAD_REQUEST,
|
|
detail=detail,
|
|
) from exc
|
|
|
|
updated_chat = {**(chat.chat or {})}
|
|
updated_chat.pop('currentId', None)
|
|
updated_chat.update(
|
|
{
|
|
'originalChatId': chat.id,
|
|
'branchPointMessageId': source_message_id,
|
|
'title': f'{chat.title} (fork)',
|
|
'history': fork_history,
|
|
'messages': fork_messages,
|
|
}
|
|
)
|
|
meta = {
|
|
**(chat.meta or {}),
|
|
'forked_from': chat.id,
|
|
'forked_from_message_id': source_message_id,
|
|
}
|
|
|
|
fork = await Chats.insert_new_chat(
|
|
str(uuid4()),
|
|
user.id,
|
|
ChatForm(chat=updated_chat, folder_id=chat.folder_id),
|
|
db=db,
|
|
internal_meta=meta,
|
|
)
|
|
|
|
if fork and chat.variables:
|
|
fork = await Chats.update_chat_variables_by_id(fork.id, chat.variables, db=db, touch=False) or fork
|
|
|
|
if not fork:
|
|
raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=ERROR_MESSAGES.DEFAULT())
|
|
|
|
if chat.pinned:
|
|
fork = await Chats.toggle_chat_pinned_by_id(fork.id, db=db) or fork
|
|
|
|
await publish_event(
|
|
request,
|
|
EVENTS.CHAT_CLONED,
|
|
actor=user,
|
|
subject_id=fork.id,
|
|
data={'original_chat_id': id, 'forked_from_message_id': source_message_id},
|
|
)
|
|
return ChatResponse.model_validate(fork, from_attributes=True)
|
|
|
|
|
|
@router.post('/{id}/clone', response_model=ChatResponse | None)
|
|
async def clone_chat_by_id(
|
|
request: Request,
|
|
form_data: CloneForm,
|
|
id: str,
|
|
user=Depends(get_verified_user),
|
|
db: AsyncSession = Depends(get_async_session),
|
|
):
|
|
await require_chat_import_permission(request, user, db)
|
|
|
|
chat = await Chats.get_chat_by_id_and_user_id(id, user.id, db=db)
|
|
if chat:
|
|
updated_chat = {
|
|
**chat.chat,
|
|
'originalChatId': chat.id,
|
|
'branchPointMessageId': chat.chat['history']['currentId'],
|
|
'title': form_data.title if form_data.title else f'Clone of {chat.title}',
|
|
}
|
|
|
|
chats = await Chats.import_chats(
|
|
user.id,
|
|
[
|
|
ChatImportForm(
|
|
**{
|
|
'chat': updated_chat,
|
|
'meta': chat.meta,
|
|
'variables': chat.variables or {},
|
|
'pinned': chat.pinned,
|
|
'folder_id': chat.folder_id,
|
|
}
|
|
)
|
|
],
|
|
db=db,
|
|
)
|
|
|
|
if chats:
|
|
chat = chats[0]
|
|
await publish_event(
|
|
request,
|
|
EVENTS.CHAT_CLONED,
|
|
actor=user,
|
|
subject_id=chat.id,
|
|
data={'original_chat_id': id},
|
|
)
|
|
return ChatResponse.model_validate(chat, from_attributes=True)
|
|
else:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail=ERROR_MESSAGES.DEFAULT(),
|
|
)
|
|
else:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail=ERROR_MESSAGES.DEFAULT())
|
|
|
|
|
|
############################
|
|
# CloneSharedChatById
|
|
############################
|
|
|
|
|
|
@router.post('/{id}/clone/shared', response_model=ChatResponse | None)
|
|
async def clone_shared_chat_by_id(
|
|
request: Request,
|
|
id: str,
|
|
user=Depends(get_verified_user),
|
|
db: AsyncSession = Depends(get_async_session),
|
|
):
|
|
await require_chat_import_permission(request, user, db)
|
|
|
|
chat = await Chats.get_chat_by_share_id(id, db=db)
|
|
|
|
# Fallback: admins can also access any chat directly by chat ID
|
|
if not chat and user.role == 'admin' and ENABLE_ADMIN_CHAT_ACCESS:
|
|
chat = await Chats.get_chat_by_id(id, db=db)
|
|
|
|
if not chat:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail=ERROR_MESSAGES.NOT_FOUND,
|
|
)
|
|
|
|
# Enforce access grants (owner and admins bypass)
|
|
shared = await SharedChats.get_by_id(id, db=db)
|
|
if shared and user.role != 'admin' and shared.user_id != user.id:
|
|
has_grant = await is_open_shared_chat(shared, db=db) or await AccessGrants.has_access(
|
|
user_id=user.id,
|
|
resource_type='shared_chat',
|
|
resource_id=shared.chat_id,
|
|
permission='read',
|
|
db=db,
|
|
)
|
|
if not has_grant:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail=ERROR_MESSAGES.ACCESS_PROHIBITED,
|
|
)
|
|
|
|
updated_chat = {
|
|
**chat.chat,
|
|
'originalChatId': chat.id,
|
|
'branchPointMessageId': chat.chat['history']['currentId'],
|
|
'title': f'Clone of {chat.title}',
|
|
}
|
|
|
|
chats = await Chats.import_chats(
|
|
user.id,
|
|
[
|
|
ChatImportForm(
|
|
**{
|
|
'chat': updated_chat,
|
|
'meta': chat.meta,
|
|
'variables': chat.variables or {},
|
|
'pinned': chat.pinned,
|
|
'folder_id': chat.folder_id,
|
|
}
|
|
)
|
|
],
|
|
db=db,
|
|
)
|
|
|
|
if chats:
|
|
chat = chats[0]
|
|
return ChatResponse.model_validate(chat, from_attributes=True)
|
|
else:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail=ERROR_MESSAGES.DEFAULT(),
|
|
)
|
|
|
|
|
|
############################
|
|
# ArchiveChat
|
|
############################
|
|
|
|
|
|
@router.post('/{id}/archive', response_model=ChatResponse | None)
|
|
async def archive_chat_by_id(
|
|
request: Request,
|
|
id: str,
|
|
user=Depends(get_verified_user),
|
|
db: AsyncSession = Depends(get_async_session),
|
|
):
|
|
chat = await Chats.get_chat_by_id_and_user_id(id, user.id, db=db)
|
|
if chat:
|
|
chat = await Chats.toggle_chat_archive_by_id(id, db=db)
|
|
|
|
tag_ids = chat.meta.get('tags', [])
|
|
if chat.archived:
|
|
# Cancel any in-flight LLM tasks before archiving
|
|
await stop_item_tasks(request.app.state.redis, id)
|
|
# Archived chats are excluded from count — clean up orphans
|
|
await Chats.delete_orphan_tags_for_user(tag_ids, user.id, db=db)
|
|
else:
|
|
# Unarchived — ensure tag rows exist
|
|
await Tags.ensure_tags_exist(tag_ids, user.id, db=db)
|
|
|
|
await publish_event(
|
|
request,
|
|
EVENTS.CHAT_ARCHIVED if chat.archived else EVENTS.CHAT_UNARCHIVED,
|
|
actor=user,
|
|
subject_id=id,
|
|
subject_type='chat',
|
|
)
|
|
return ChatResponse.model_validate(chat, from_attributes=True)
|
|
else:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail=ERROR_MESSAGES.DEFAULT())
|
|
|
|
|
|
# --- Share Chat ---
|
|
|
|
|
|
@router.post('/{id}/share', response_model=ChatResponse | None)
|
|
async def share_chat_by_id(
|
|
request: Request,
|
|
id: str,
|
|
user=Depends(get_verified_user),
|
|
db: AsyncSession = Depends(get_async_session),
|
|
):
|
|
if user.role != 'admin' and not await has_permission(user.id, 'chat.share', await Config.get('user.permissions')):
|
|
raise HTTPException(status.HTTP_401_UNAUTHORIZED, detail=ERROR_MESSAGES.ACCESS_PROHIBITED)
|
|
|
|
chat = await Chats.get_chat_by_id_and_user_id(id, user.id, db=db)
|
|
if not chat:
|
|
raise HTTPException(status.HTTP_401_UNAUTHORIZED, detail=ERROR_MESSAGES.ACCESS_PROHIBITED)
|
|
|
|
# If a share already exists, re-snapshot it
|
|
if chat.share_id:
|
|
shared = await SharedChats.update(chat.share_id, db=db)
|
|
if shared:
|
|
chat = await Chats.get_chat_by_id(id, db=db)
|
|
await publish_event(
|
|
request,
|
|
EVENTS.CHAT_SHARED,
|
|
actor=user,
|
|
subject_id=id,
|
|
data={'share_id': chat.share_id, 'updated': True},
|
|
)
|
|
return ChatResponse.model_validate(chat, from_attributes=True)
|
|
|
|
# Create a new share
|
|
shared = await SharedChats.create(id, user.id, db=db)
|
|
if not shared:
|
|
raise HTTPException(status.HTTP_500_INTERNAL_SERVER_ERROR, detail=ERROR_MESSAGES.DEFAULT())
|
|
|
|
chat = await Chats.update_chat_share_id_by_id(id, shared.id, db=db)
|
|
if not chat:
|
|
raise HTTPException(status.HTTP_500_INTERNAL_SERVER_ERROR, detail=ERROR_MESSAGES.DEFAULT())
|
|
|
|
await publish_event(
|
|
request,
|
|
EVENTS.CHAT_SHARED,
|
|
actor=user,
|
|
subject_id=id,
|
|
data={'share_id': shared.id},
|
|
)
|
|
return ChatResponse.model_validate(chat, from_attributes=True)
|
|
|
|
|
|
# --- Delete Shared Chat ---
|
|
|
|
|
|
@router.delete('/{id}/share', response_model=bool | None)
|
|
async def delete_shared_chat_by_id(
|
|
request: Request, id: str, user=Depends(get_verified_user), db: AsyncSession = Depends(get_async_session)
|
|
):
|
|
chat = await Chats.get_chat_by_id_and_user_id(id, user.id, db=db)
|
|
if not chat:
|
|
raise HTTPException(status.HTTP_401_UNAUTHORIZED, detail=ERROR_MESSAGES.ACCESS_PROHIBITED)
|
|
|
|
await SharedChats.delete_by_chat_id(id, db=db)
|
|
|
|
if chat.share_id:
|
|
await Chats.update_chat_share_id_by_id(id, None, db=db)
|
|
|
|
await AccessGrants.set_access_grants('shared_chat', id, [], db=db)
|
|
|
|
await publish_event(
|
|
request,
|
|
EVENTS.CHAT_UNSHARED,
|
|
actor=user,
|
|
subject_id=id,
|
|
data={'share_id': chat.share_id},
|
|
)
|
|
return True
|
|
|
|
|
|
############################
|
|
# UpdateSharedChatAccessById
|
|
############################
|
|
|
|
|
|
class ChatAccessGrantsForm(BaseModel):
|
|
access_grants: list[dict]
|
|
|
|
|
|
@router.post('/shared/{id}/access/update', response_model=ChatResponse | None)
|
|
async def update_shared_chat_access_by_id(
|
|
request: Request,
|
|
id: str,
|
|
form_data: ChatAccessGrantsForm,
|
|
user=Depends(get_verified_user),
|
|
db: AsyncSession = Depends(get_async_session),
|
|
):
|
|
if user.role == 'admin':
|
|
chat = await Chats.get_chat_by_id(id, db=db)
|
|
else:
|
|
chat = await Chats.get_chat_by_id_and_user_id(id, user.id, db=db)
|
|
if not chat:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=ERROR_MESSAGES.NOT_FOUND,
|
|
)
|
|
|
|
form_data.access_grants = await filter_allowed_access_grants(
|
|
await Config.get('user.permissions'),
|
|
user.id,
|
|
user.role,
|
|
form_data.access_grants,
|
|
'sharing.public_chats',
|
|
'sharing.open_chats',
|
|
db=db,
|
|
)
|
|
|
|
await AccessGrants.set_access_grants('shared_chat', id, form_data.access_grants, db=db)
|
|
|
|
return ChatResponse.model_validate(chat, from_attributes=True)
|
|
|
|
|
|
############################
|
|
# GetSharedChatAccessById
|
|
############################
|
|
|
|
|
|
@router.get('/shared/{id}/access', response_model=list)
|
|
async def get_shared_chat_access_by_id(
|
|
id: str,
|
|
user=Depends(get_verified_user),
|
|
db: AsyncSession = Depends(get_async_session),
|
|
):
|
|
if user.role == 'admin':
|
|
chat = await Chats.get_chat_by_id(id, db=db)
|
|
else:
|
|
chat = await Chats.get_chat_by_id_and_user_id(id, user.id, db=db)
|
|
if not chat:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=ERROR_MESSAGES.NOT_FOUND,
|
|
)
|
|
|
|
grants = await AccessGrants.get_grants_by_resource('shared_chat', id, db=db)
|
|
return [
|
|
{
|
|
'id': g.id,
|
|
'principal_type': g.principal_type,
|
|
'principal_id': g.principal_id,
|
|
'permission': g.permission,
|
|
}
|
|
for g in grants
|
|
]
|
|
|
|
|
|
############################
|
|
# UpdateChatFolderIdById
|
|
############################
|
|
|
|
|
|
class ChatFolderIdForm(BaseModel):
|
|
folder_id: str | None = None
|
|
|
|
|
|
@router.post('/{id}/unread')
|
|
async def mark_chat_unread_by_id(
|
|
id: str,
|
|
user=Depends(get_verified_user),
|
|
db: AsyncSession = Depends(get_async_session),
|
|
):
|
|
chat = await Chats.mark_chat_unread_by_id(id, user.id, db=db)
|
|
if not chat:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=ERROR_MESSAGES.NOT_FOUND,
|
|
)
|
|
|
|
folder_id = await Chats.get_chat_folder_id(id, user.id, db=db)
|
|
return {
|
|
'chat_id': id,
|
|
'last_read_at': chat.last_read_at,
|
|
'folder_id': folder_id,
|
|
'folder_unread_counts': await get_folder_unread_counts(user.id, db=db),
|
|
}
|
|
|
|
|
|
@router.post('/{id}/folder', response_model=ChatResponse | None)
|
|
async def update_chat_folder_id_by_id(
|
|
request: Request,
|
|
id: str,
|
|
form_data: ChatFolderIdForm,
|
|
user=Depends(get_verified_user),
|
|
db: AsyncSession = Depends(get_async_session),
|
|
):
|
|
chat = await Chats.get_chat_by_id_and_user_id(id, user.id, db=db)
|
|
if chat:
|
|
# None is allowed: it moves the chat out of any folder.
|
|
if form_data.folder_id is not None and not await has_folder_write_access(user.id, form_data.folder_id, db=db):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=ERROR_MESSAGES.NOT_FOUND,
|
|
)
|
|
|
|
chat = await Chats.update_chat_folder_id_by_id_and_user_id(id, user.id, form_data.folder_id, db=db)
|
|
await publish_event(
|
|
request,
|
|
EVENTS.CHAT_FOLDER_UPDATED,
|
|
actor=user,
|
|
subject_id=id,
|
|
data={'folder_id': form_data.folder_id},
|
|
)
|
|
return ChatResponse.model_validate(chat, from_attributes=True)
|
|
else:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail=ERROR_MESSAGES.DEFAULT())
|
|
|
|
|
|
############################
|
|
# GetChatTagsById
|
|
############################
|
|
|
|
|
|
@router.get('/{id}/tags', response_model=list[TagModel])
|
|
async def get_chat_tags_by_id(id: str, user=Depends(get_verified_user), db: AsyncSession = Depends(get_async_session)):
|
|
chat = await Chats.get_chat_by_id_and_user_id(id, user.id, db=db)
|
|
if chat:
|
|
tags = chat.meta.get('tags', [])
|
|
return await Tags.get_tags_by_ids_and_user_id(tags, user.id, db=db)
|
|
else:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail=ERROR_MESSAGES.NOT_FOUND)
|
|
|
|
|
|
############################
|
|
# AddChatTagById
|
|
############################
|
|
|
|
|
|
@router.post('/{id}/tags', response_model=list[TagModel])
|
|
async def add_tag_by_id_and_tag_name(
|
|
request: Request,
|
|
id: str,
|
|
form_data: TagForm,
|
|
user=Depends(get_verified_user),
|
|
db: AsyncSession = Depends(get_async_session),
|
|
):
|
|
chat = await Chats.get_chat_by_id_and_user_id(id, user.id, db=db)
|
|
if chat:
|
|
tags = chat.meta.get('tags', [])
|
|
tag_id = form_data.name.replace(' ', '_').lower()
|
|
|
|
if tag_id == 'none':
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=ERROR_MESSAGES.DEFAULT("Tag name cannot be 'None'"),
|
|
)
|
|
|
|
if tag_id not in tags:
|
|
await Chats.add_chat_tag_by_id_and_user_id_and_tag_name(id, user.id, form_data.name, db=db)
|
|
await publish_event(
|
|
request,
|
|
EVENTS.CHAT_TAG_ADDED,
|
|
actor=user,
|
|
subject_id=id,
|
|
data={'tag': form_data.name},
|
|
)
|
|
|
|
chat = await Chats.get_chat_by_id_and_user_id(id, user.id, db=db)
|
|
tags = chat.meta.get('tags', [])
|
|
return await Tags.get_tags_by_ids_and_user_id(tags, user.id, db=db)
|
|
else:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail=ERROR_MESSAGES.DEFAULT())
|
|
|
|
|
|
############################
|
|
# DeleteChatTagById
|
|
############################
|
|
|
|
|
|
@router.delete('/{id}/tags', response_model=list[TagModel])
|
|
async def delete_tag_by_id_and_tag_name(
|
|
request: Request,
|
|
id: str,
|
|
form_data: TagForm,
|
|
user=Depends(get_verified_user),
|
|
db: AsyncSession = Depends(get_async_session),
|
|
):
|
|
chat = await Chats.get_chat_by_id_and_user_id(id, user.id, db=db)
|
|
if chat:
|
|
await Chats.delete_tag_by_id_and_user_id_and_tag_name(id, user.id, form_data.name, db=db)
|
|
await publish_event(
|
|
request,
|
|
EVENTS.CHAT_TAG_REMOVED,
|
|
actor=user,
|
|
subject_id=id,
|
|
data={'tag': form_data.name},
|
|
)
|
|
|
|
if await Chats.count_chats_by_tag_name_and_user_id(form_data.name, user.id, db=db) == 0:
|
|
await Tags.delete_tag_by_name_and_user_id(form_data.name, user.id, db=db)
|
|
|
|
chat = await Chats.get_chat_by_id_and_user_id(id, user.id, db=db)
|
|
tags = chat.meta.get('tags', [])
|
|
return await Tags.get_tags_by_ids_and_user_id(tags, user.id, db=db)
|
|
else:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail=ERROR_MESSAGES.NOT_FOUND)
|