feat(chat): 单向聊天可二次激活 单向聊天可二次激活，未删除方聊天消息保留，双向删除后记录清空

alembic/versions/43b2d2efe0f7_merage_chat_participant_and_read_status.py

@@ -0,0 +1,52 @@
+"""merage chat_participant and read_status
+
+Revision ID: 43b2d2efe0f7
+Revises: e9466c397aa2
+Create Date: 2025-05-16 22:20:04.592819
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import mysql
+
+# revision identifiers, used by Alembic.
+revision: str = '43b2d2efe0f7'
+down_revision: Union[str, None] = 'e9466c397aa2'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_index('idx_read_status_on_chat_id', table_name='system_read_status')
+    op.drop_index('idx_read_status_on_user_id', table_name='system_read_status')
+    op.drop_table('system_read_status')
+    op.add_column('system_chat_participant', sa.Column('last_read_message_id', sa.Integer(), nullable=True))
+    op.add_column('system_chat_participant', sa.Column('last_del_message_id', sa.Integer(), nullable=True))
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_column('system_chat_participant', 'last_del_message_id')
+    op.drop_column('system_chat_participant', 'last_read_message_id')
+    op.create_table('system_read_status',
+    sa.Column('last_read_message_id', mysql.INTEGER(display_width=11), autoincrement=False, nullable=True),
+    sa.Column('user_id', mysql.INTEGER(display_width=11), autoincrement=False, nullable=False),
+    sa.Column('chat_id', mysql.INTEGER(display_width=11), autoincrement=False, nullable=False),
+    sa.Column('id', mysql.INTEGER(display_width=11), autoincrement=True, nullable=False, comment='自增id'),
+    sa.Column('created_at', mysql.DATETIME(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False, comment='创建时间'),
+    sa.Column('updated_at', mysql.DATETIME(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False, comment='更新时间'),
+    sa.ForeignKeyConstraint(['chat_id'], ['system_chats.id'], name='system_read_status_ibfk_1'),
+    sa.ForeignKeyConstraint(['user_id'], ['system_users.user_id'], name='system_read_status_ibfk_2'),
+    sa.PrimaryKeyConstraint('id'),
+    comment='聊天消息读取状态表',
+    mysql_comment='聊天消息读取状态表',
+    mysql_default_charset='utf8mb4',
+    mysql_engine='InnoDB'
+    )
+    op.create_index('idx_read_status_on_user_id', 'system_read_status', ['user_id'], unique=False)
+    op.create_index('idx_read_status_on_chat_id', 'system_read_status', ['chat_id'], unique=False)
+    # ### end Alembic commands ###

core/db/models.py

@@ -99,7 +99,7 @@ class User(BaseModel, DateModel):
     )
     chats: Mapped[List['Chat']] = relationship(secondary='system_chat_participant', back_populates='users')
     messages: Mapped[List['Message']] = relationship(back_populates='user')
-    read_statuses: Mapped[List['ReadStatus']] = relationship(back_populates='user')
+    chat_participants: Mapped[List['ChatParticipant']] = relationship(back_populates='user')
 
     def __str__(self):
         return f'{self.user_id}'
@@ -264,7 +264,7 @@ class Chat(BaseModel, DateModel):
     users: Mapped[List['User']] = relationship(secondary='system_chat_participant', back_populates='chats')
     messages: Mapped[List['Message']] = relationship(back_populates='chat', cascade='all,delete')
     is_deleted: Mapped[bool] = mapped_column(Boolean, default=False)
-    read_statuses: Mapped[List['ReadStatus']] = relationship(back_populates='chat', cascade='all,delete')
+    chat_participants: Mapped[List['ChatParticipant']] = relationship(back_populates='chat', cascade='all,delete')
 
 
 class Message(BaseModel, DateModel):
@@ -292,25 +292,6 @@ class Message(BaseModel, DateModel):
         return f'{self.content}'
 
 
-class ReadStatus(BaseModel, DateModel):
-    __tablename__ = 'system_read_status'
-    __table_args__ = (
-        Index('idx_read_status_on_chat_id', 'chat_id'),
-        Index('idx_read_status_on_user_id', 'user_id'),
-        {'comment': '聊天消息读取状态表'}
-    )
-
-    last_read_message_id: Mapped[int] = mapped_column(nullable=True)
-    user_id: Mapped[int] = mapped_column(ForeignKey('system_users.user_id'))
-    chat_id: Mapped[int] = mapped_column(ForeignKey('system_chats.id'))
-
-    chat: Mapped['Chat'] = relationship(back_populates='read_statuses')
-    user: Mapped['User'] = relationship(back_populates='read_statuses')
-
-    def __str__(self):
-        return f'User: {self.user_id}, Message: {self.last_read_message_id}'
-
-
 class PointsTransaction(BaseModel, DateModel):
     __tablename__ = 'system_points_transaction'
     __table_args__ = (
@@ -335,3 +316,8 @@ class ChatParticipant(BaseModel):
     user_id: Mapped[int] = mapped_column(Integer, ForeignKey('system_users.user_id'), primary_key=True)
     chat_id: Mapped[int] = mapped_column(Integer, ForeignKey('system_chats.id'), primary_key=True)
     is_deleted: Mapped[bool] = mapped_column(Boolean, default=False)
+    last_read_message_id: Mapped[int] = mapped_column(default=0, nullable=True)
+    last_del_message_id: Mapped[int] = mapped_column(default=0, nullable=True)
+
+    chat: Mapped['Chat'] = relationship(back_populates='chat_participants')
+    user: Mapped['User'] = relationship(back_populates='chat_participants')

core/routers/chat/handlers.py

@@ -10,7 +10,7 @@ from fastapi import WebSocket
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from core.db.crud import CRUD
-from core.db.models import ReadStatus, User, Chat, Message
+from core.db.models import ChatParticipant, User, Chat, Message
 
 from .managers import SocketManager
 from . import schema
@@ -29,7 +29,6 @@ async def send_new_message(
         incoming_message: schema.WSMessage,
 ):
     chat_guid: str = str(incoming_message.chat_guid)
-    # chat: Chat | None = await CRUD(db_session, Chat).get(filters={'guid': chat_guid, 'is_deleted': False}
     chat: Chat | None = await get_chat(chat_guid, current_user, db_session)
 
     # 判断是否为新聊天
@@ -73,16 +72,7 @@ async def send_new_message(
         content=message.content,
         friend_user=current_user.dict(),
     )
-    send_status_replay = schema.WSMessageOut(
-        type='chat.send_status_reply',
-        message_id=message.id,
-        message_guid=message.guid,
-        message_type=message.message_type,
-        chat_guid=chat.guid,
-    )
-
     await socket_manager.broadcast_to_chat(chat_guid, send_message_schema.model_dump_json())
-    await websocket.send_text(send_status_replay.model_dump_json())
 
 
 @socket_manager.handler('chat.message_read')
@@ -90,16 +80,23 @@ async def read_message(
         websocket: WebSocket,
         db_session: AsyncSession,
         current_user: User,
+        chats: dict,
         incoming_message: schema.WSMessage,
 ):
     last_read_message_id: str | None = incoming_message.content
     if not last_read_message_id:
         return
 
-    db_crud = CRUD(db_session, ReadStatus)
+    db_crud = CRUD(db_session, ChatParticipant)
+    chat_guid: str = str(incoming_message.chat_guid)
+    chat: Chat | None = await get_chat(chat_guid, current_user, db_session)
+    if chat is None:
+        await socket_manager.send_error('聊天尚未添加', websocket)
+        return
+
     condition = {
         'user_id': current_user.user_id,
-        'chat_guid': incoming_message.chat_guid,
+        'chat_id': chat.id,
     }
     result = await db_crud.update(filters=condition, data={'last_read_message_id': last_read_message_id})
     if result is None:

core/routers/chat/routes.py

@@ -17,7 +17,7 @@ from sqlalchemy.orm import selectinload
 from core.base import schema as base_schema
 from core.db.crud import CRUD
 from core.db.engine import get_async_session
-from core.db.models import Message, Chat, User, ReadStatus, ChatParticipant
+from core.db.models import Message, Chat, User, ChatParticipant
 from core.routers.auth.schema import LoginUser
 from core.routers.auth.services import get_current_user_form_http, get_current_user_form_ws
 from core.base.exceptions import WebsocketTooManyRequests
@@ -100,10 +100,14 @@ async def get_single_chat_messages(
     if not chat:
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail='您访问的聊天不存在')
 
+    read_status = await CRUD(db_session, ChatParticipant).get(
+        filters={'chat_id': chat.id, 'user_id': login_user.user_id})
+
     # 组装查询条件
     conditions = and_(
         Message.chat_id == chat.id,
-        Message.id <= last_read_message_id
+        Message.id > (read_status.last_del_message_id or 0),
+        Message.id <= last_read_message_id,
     )
 
     # 记录总数查询
@@ -152,11 +156,13 @@ async def create_new_chat(
     if payload.recipient_user_id == login_user.user_id:
         raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail='操作禁止')
 
-    recipient_user: User | None = await CRUD(db_session, User).get(filters={'user_id': payload.recipient_user_id})
+    db_crud = CRUD(db_session, User)
+
+    recipient_user: User | None = await db_crud.get(filters={'user_id': payload.recipient_user_id})
     if recipient_user is None:
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail='收件人不存在')
 
-    initiator_user: User = await CRUD(db_session, User).get(filters={'user_id': login_user.user_id})
+    initiator_user: User = await db_crud.get(filters={'user_id': login_user.user_id})
     # 查询是否已存在 对应且未关闭的聊天
     query = (
         select(Chat)
@@ -170,7 +176,7 @@ async def create_new_chat(
                     (ChatParticipant.chat_id == Chat.id) &
                     (ChatParticipant.user_id == recipient_user.user_id)
                 ),
-                Chat.is_deleted == False
+                # Chat.is_deleted == False
             )
         ).order_by(Chat.created_at.desc())
     )
@@ -184,16 +190,16 @@ async def create_new_chat(
             chat.users.append(recipient_user)
             db_session.add(chat)
             await db_session.flush()
-
-            initiator_read_status = ReadStatus(chat_id=chat.id, user_id=initiator_user.user_id, last_read_message_id=0)
-            recipient_read_status = ReadStatus(chat_id=chat.id, user_id=recipient_user.user_id, last_read_message_id=0)
-            db_session.add_all([initiator_read_status, recipient_read_status])
             await db_session.commit()
 
         except Exception as exc_info:
             await db_session.rollback()
             raise exc_info
 
+    if chat.is_deleted is True:
+        await db_crud.updates(model=Chat, data={'is_deleted': False}, filters={'id': chat.id})
+        await db_crud.updates(model=ChatParticipant, data={'is_deleted': False}, filters={'chat_id': chat.id})
+
     return {
         'data': {
             'chat_guid': chat.guid,
@@ -219,12 +225,21 @@ async def delete_single_chat(
             Chat.id.in_(subquery),
             Chat.guid == chat_guid
         ))
+        .options(selectinload(Chat.chat_participants))
     )
     result = await db_session.execute(query)
     chat = result.scalar_one_or_none()
 
     if chat is None:
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail='聊天不存在')
+
+    deleted_status = [
+        chat_participant.is_deleted
+        for chat_participant in chat.chat_participants if chat_participant.user_id != login_user.user_id
+    ]
+    if all(deleted_status):
+        await db_session.delete(chat)
+        await db_session.commit()
     else:
         db_curd = CRUD(db_session, Chat)
         await db_curd.update(
@@ -232,10 +247,21 @@ async def delete_single_chat(
             filters={'id': chat.id},
         )
 
+        message_query = (
+            select(Message).where(Message.chat_id == chat.id).order_by(Message.id.desc()).limit(1)
+        )
+        message_result = await db_session.execute(message_query)
+        message: Message | None = message_result.scalar()
+
+        condition = {'user_id': login_user.user_id, 'chat_id': chat.id}
+        data = {'is_deleted': True}
+        if message:
+            data['last_del_message_id'] = message.id
+
         await db_curd.update(
             model=ChatParticipant,
-            data={'is_deleted': True},
+            data=data,
-            filters={'user_id': login_user.user_id, 'chat_id': chat.id},
+            filters=condition,
         )
 
     return {'message': '聊天已删除'}
@@ -287,6 +313,15 @@ async def get_chat_websocket(
                     incoming_message=message,
                 )
 
+                send_status_replay = schema.WSMessageOut(
+                    type='chat.send_status_reply',
+                    message_id=-1,
+                    message_guid=message.message_guid,
+                    message_type=message.message_type,
+                    chat_guid=message.chat_guid,
+                )
+                await websocket.send_text(send_status_replay.model_dump_json(exclude_none=True))
+
             except (JSONDecodeError, AttributeError) as exc:
                 logger.exception(f'Websocket error, detail: {exc}')
                 await socket_manager.send_error('Wrong message format', websocket)

core/routers/chat/services.py

@@ -11,7 +11,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import aliased, selectinload
 
 from core.base.exceptions import WebsocketTooManyRequests
-from core.db.models import User, Chat, ReadStatus, Message, ChatParticipant
+from core.db.models import User, Chat, Message, ChatParticipant
 from core.routers.auth.schema import LoginUser
 from core.utils.helper import format_date
 
@@ -44,7 +44,7 @@ async def get_user_active_direct_chats(
 async def get_chat_unread_messages_count(db_session: AsyncSession, login_user: LoginUser, chats):
     """查询未读消息的预览"""
     # 查询消息情况
-    read_status_alias = aliased(ReadStatus)
+    read_status_alias = aliased(ChatParticipant)
     query = (
         select(Message.chat_id, func.count().label("message_count"))
         .join(