diff --git a/core/base/schema.py b/core/base/schema.py index 55a947c..363ec06 100644 --- a/core/base/schema.py +++ b/core/base/schema.py @@ -25,6 +25,11 @@ class Paginated(BaseModel, Generic[T]): page_size: int = 15 +class PaginatedCursor(Paginated): + cursor: int = None + next_page: bool = None, + + class SupportPlatformType(enum.Enum): PSA = 'psa' BGS = 'bgs' diff --git a/core/routers/chat/routes.py b/core/routers/chat/routes.py index 3b6c460..a15f2a4 100644 --- a/core/routers/chat/routes.py +++ b/core/routers/chat/routes.py @@ -34,7 +34,7 @@ chat_router = APIRouter(prefix='/chat', tags=['聊天模块']) @chat_router.get( '/all/', summary='获取所有聊天', response_model=base_schema.BaseResponse[ - base_schema.Paginated[ + base_schema.PaginatedCursor[ List[schema.ChatUnreadOut] ] ] @@ -72,6 +72,7 @@ async def get_all_chats( return { 'data': { 'list': unread_messages_chats_count, + 'next_page': (page * page_size) < total, 'total': total, 'page': page, 'page_size': page_size, @@ -82,7 +83,7 @@ async def get_all_chats( @chat_router.get( '/{chat_guid}/messages/', summary='获取单个聊天的历史消息', response_model=base_schema.BaseResponse[ - base_schema.Paginated[ + base_schema.PaginatedCursor[ List[schema.MessageOut] ] ] @@ -111,14 +112,15 @@ async def get_single_chat_messages( conditions ) ) - total = await db_session.execute(total_stmt) + total_result = await db_session.execute(total_stmt) + total = total_result.scalar() # 分页记录查询 stmt = ( select(Message) .where(conditions) - .order_by(Message.created_at.desc()) - .limit(page_size).offset((page - 1) * page_size) .options(selectinload(Message.user), selectinload(Message.chat)) + .order_by(Message.id.desc()). + limit(page_size).offset((page - 1) * page_size) ) records = await db_session.execute(stmt) results = [ @@ -133,7 +135,9 @@ async def get_single_chat_messages( return { 'data': { 'list': results, - 'total': total.scalar(), + 'cursor': results[-1].get('id'), + 'next_page': (page * page_size) < total, + 'total': total, 'page': page, 'page_size': page_size, }, @@ -250,7 +254,7 @@ async def get_chat_websocket( websocket.user_id = login_user.user_id await socket_manager.add_user_socket_connection(user_id, websocket) - ratelimit = WebSocketRateLimiter(times=50, seconds=10, callback=websocket_callback) + ratelimit = WebSocketRateLimiter(times=30, seconds=10, callback=websocket_callback) current_user = await CRUD(db_session, User).get(filters={'user_id': login_user.user_id}) if chats := await get_user_active_direct_chats(db_session, current_user=current_user): for chat_guid in chats: diff --git a/core/routers/chat/services.py b/core/routers/chat/services.py index 17fcfa3..1d0a1c7 100644 --- a/core/routers/chat/services.py +++ b/core/routers/chat/services.py @@ -41,16 +41,12 @@ async def get_user_active_direct_chats( return None -async def get_unread_messages_per_chat( - db_session: AsyncSession, chats: list[Chat], login_user: LoginUser -) -> list[dict]: - """获取聊天未读消息数""" - unread_messages_info_per_chat: dict[int, dict] = {chat.id: {'message_count': 0} for chat in chats} - +async def get_chat_unread_messages_count(db_session: AsyncSession, login_user: LoginUser, chats): + """查询未读消息的预览""" # 查询消息情况 read_status_alias = aliased(ReadStatus) query = ( - select(Message.id, Message.chat_id, Message.content, Message.message_type, func.count().label("message_count")) + select(Message.chat_id, func.count().label("message_count")) .join( read_status_alias, and_(read_status_alias.user_id == login_user.user_id, read_status_alias.chat_id == Message.chat_id), @@ -59,29 +55,71 @@ async def get_unread_messages_per_chat( and_( Message.user_id != login_user.user_id, Message.id > func.coalesce(read_status_alias.last_read_message_id, 0), - Message.chat_id.in_(unread_messages_info_per_chat), + Message.chat_id.in_(chats), ) ) - .order_by(Message.created_at.desc()) .group_by(Message.chat_id) ) + result = await db_session.execute(query) unread_messages_info = result.fetchall() + return unread_messages_info - for messages_info in unread_messages_info: - unread_messages_info_per_chat[messages_info[1]] = { - 'message': { - 'id': messages_info[0], - 'message_type': messages_info[3], - 'content': messages_info[2], - }, - 'message_count': messages_info[4], + +async def get_chat_messages_preview(db_session: AsyncSession, chats): + """查询未读消息的预览""" + latest_subquery = ( + select( + Message.chat_id, + func.max(Message.created_at).label("max_created") + ) + .group_by(Message.chat_id) + .subquery() + ) + + LatestMessage = aliased(Message) + + query = ( + select(LatestMessage) + .join(latest_subquery, + (LatestMessage.chat_id == latest_subquery.c.chat_id) & + (LatestMessage.created_at == latest_subquery.c.max_created)) + ) + result = await db_session.execute(query) + unread_messages_info = result.scalars().all() + return unread_messages_info + + +async def get_unread_messages_per_chat( + db_session: AsyncSession, chats: list[Chat], login_user: LoginUser +) -> list[dict]: + """获取聊天未读消息数和消息预览""" + messages_info_per_chat: dict[int, dict] = {chat.id: {'message_count': 0} for chat in chats} + + # 未读消息数和预览 + unread_messages_count = await get_chat_unread_messages_count( + db_session, login_user, messages_info_per_chat + ) + for messages_info in unread_messages_count: + messages_info_per_chat[messages_info[0]] = { + 'message_count': messages_info[1], } + # 已读消息预览 + no_unread_message_chats = [ + chat_id + for chat_id, message_info in messages_info_per_chat.items() if message_info.get('message') is None + ] + messages_info = await get_chat_messages_preview(db_session, no_unread_message_chats) + for message_info in messages_info: + messages_info_per_chat[message_info.chat_id].update({ + 'message': message_info + }) + results = [] for chat in chats: users = [user for user in await chat.awaitable_attrs.users if user.user_id != login_user.user_id] - messages_info = unread_messages_info_per_chat[chat.id] + messages_info = messages_info_per_chat[chat.id] results.append(dict( chat_guid=chat.guid, created_at=format_date(chat.created_at),