Merge pull request 'feat(chat_messages): fixed data race for chat mesages state' (#1042) from misha/eatvio_app:chat_messages into chat_messages

Reviewed-on: eatvio/eatvio_app#1042

lib/core/websockets/ws_manager.dart

@@ -38,13 +38,16 @@ class WsManager {
   StreamSubscription? _eventSubscription;
   StreamSubscription? _connectionEstablishedSubscription;
 
+  Stream<ChannelReadEvent>? _directChatStream;
+  Stream<ChannelReadEvent>? _clientTypingStream;
+
   Stream<ChannelReadEvent>? get newMessagesStream => _privateMessagesChannel?.bind('App\\Events\\ChatMessageUpdated');
 
   Stream<ChannelReadEvent>? get newNotificationsStream => _privateNotificationChannel?.bind('Illuminate\\Notifications\\Events\\BroadcastNotificationCreated');
 
-  Stream<ChannelReadEvent>? get directChatStream => _privateDirectChatChannel?.bind('App\\Events\\ChatMessageUpdated');
+  Stream<ChannelReadEvent>? get directChatStream => _directChatStream;
 
-  Stream<ChannelReadEvent>? get clientTypingStream => _privateDirectChatChannel?.bind('client-typing');
+  Stream<ChannelReadEvent>? get clientTypingStream => _clientTypingStream;
 
   Future<void> initialize() async {
     if (client == null) {
@@ -107,6 +110,8 @@ class WsManager {
           if (data == PusherChannelsClientLifeCycleState.establishedConnection) {
             _unsubscribe();
             _subscribe();
+            _privateDirectChatChannel?.subscribe();
+            _presenceDirectChatChannel?.subscribe();
           }
         });
         _eventSubscription = client?.eventStream.listen((event) {
@@ -170,6 +175,7 @@ class WsManager {
   }
 
   Future<void> initPrivateDirectChatChannel(String chatId) async {
+    _privateDirectChatChannel?.unsubscribe();
     final token = await TokenStorage.getTokenFromStorage();
     _privateDirectChatChannel = client?.privateChannel(
       '$directChatChannelName.$chatId',
@@ -180,6 +186,8 @@ class WsManager {
         },
       ),
     );
+    _directChatStream = _privateDirectChatChannel?.bind('App\\Events\\ChatMessageUpdated');
+    _clientTypingStream = _privateDirectChatChannel?.bind('client-typing');
   }
 
   Future<void> initPresenceDirectChatChannel(String chatId) async {
@@ -299,6 +307,8 @@ class WsManager {
 
   void unsubscribeDirectChatChannel() {
     _privateDirectChatChannel?.unsubscribe();
+    _directChatStream = null;
+    _clientTypingStream = null;
   }
 
   void unsubscribePresenceDirectChatChannel() {
@@ -331,6 +341,8 @@ class WsManager {
     _privateDirectChatChannel = null;
     _privateMessagesChannel = null;
     _privateNotificationChannel = null;
+    _directChatStream = null;
+    _clientTypingStream = null;
   }
 }
 

lib/features/chat/presentation/bloc/chat_room_cubit/chat_room_cubit.dart

@@ -54,16 +54,16 @@ class ChatRoomCubit extends Cubit<ChatRoomState> {
     if (currentPage < lastPage && !isLoading) {
       isLoading = true;
 
-      emit((state as ChatRoomStateLoaded).copyWith(isLoading: isLoading));
+      _emitLoaded((current) => current.copyWith(isLoading: isLoading));
 
       try {
         final chatMessages = await chatRepository.getChatMessages(uuid, currentPage + 1);
 
         isLoading = false;
-        emit(
-          (state as ChatRoomStateLoaded).copyWith(
+        _emitLoaded(
+          (current) => current.copyWith(
             isLoading: isLoading,
-            chatMessages: [...(state as ChatRoomStateLoaded).chatMessages, ...chatMessages.messagesList],
+            chatMessages: [...current.chatMessages, ...chatMessages.messagesList],
           ),
         );
 
@@ -86,11 +86,8 @@ class ChatRoomCubit extends Cubit<ChatRoomState> {
   }
 
   void _setCard({required List<CardModel> cards}) {
-    final currentState = state;
-    if (currentState is ChatRoomStateLoaded) {
-      if (cards.where((card) => card.isDefault).isNotEmpty) {
-        emit(currentState.copyWith(hasCard: true));
-      }
+    if (cards.where((card) => card.isDefault).isNotEmpty) {
+      _emitLoaded((current) => current.copyWith(hasCard: true));
     }
   }
 
@@ -117,6 +114,9 @@ class ChatRoomCubit extends Cubit<ChatRoomState> {
         throw ArgumentError('initChat requires uuid or orderId');
       }
 
+      currentPage = 1;
+      isLoading = false;
+
       final chatRoom = await chatRepository.getChatRoom(resolvedUuid);
       ChosenPlanModel? chosenPlan;
       final planId = resolvedChosenPlanId ?? chatRoom.chosenPlanId;
@@ -135,6 +135,7 @@ class ChatRoomCubit extends Cubit<ChatRoomState> {
           hasCard: false,
         ),
       );
+      await _subscribeToPresenceChannel(chatId: chatRoom.id);
 
       if (chosenPlan?.consultantId == chatRoom.users.sender.id) {
         await checkCards();
@@ -154,13 +155,15 @@ class ChatRoomCubit extends Cubit<ChatRoomState> {
   Future<void> updateChosenPlanState() async {
     try {
       final currentState = state;
-      if (currentState is ChatRoomStateLoaded) {
-        final chosenPlanId = currentState.chatRoom.chosenPlanId;
-        if (chosenPlanId != null) {
-          final chosenPlan = await chatRepository.getChosenPlan(id: chosenPlanId);
-          emit(currentState.copyWith.chatRoom(chosenPlan: chosenPlan));
-        }
+      if (currentState is! ChatRoomStateLoaded) {
+        return;
       }
+      final chosenPlanId = currentState.chatRoom.chosenPlanId;
+      if (chosenPlanId == null) {
+        return;
+      }
+      final chosenPlan = await chatRepository.getChosenPlan(id: chosenPlanId);
+      _emitLoaded((latest) => latest.copyWith.chatRoom(chosenPlan: chosenPlan));
     } catch (e, st) {
       await reportFrom(e, st, this);
     }
@@ -193,38 +196,92 @@ class ChatRoomCubit extends Cubit<ChatRoomState> {
     }
   }
 
+  bool _emitLoaded(ChatRoomStateLoaded Function(ChatRoomStateLoaded current) update) {
+    if (isClosed) {
+      return false;
+    }
+    final currentState = state;
+    if (currentState is! ChatRoomStateLoaded) {
+      return false;
+    }
+    emit(update(currentState));
+    return true;
+  }
+
   Future<void> _subscribeToChatChannel({required String uuid, required String chatId}) async {
     final client = wsManager.client;
     if (client == null) return;
     _activeChatUuid = uuid;
+    await messagesSubscription?.cancel();
+    await typingSubscription?.cancel();
+    messagesSubscription = null;
+    typingSubscription = null;
     await wsManager.initPrivateDirectChatChannel(chatId);
     wsManager.subscribeDirectChatChannel();
-    await _subscribeToPresenceChannel(chatId: chatId);
-    messagesSubscription = wsManager.directChatStream?.listen((event) {
-      AudioPlayer().play((AssetSource('audio/recive_message.wav')));
-      final message = ChatMessageModel.fromJson(jsonDecode(event.data));
-      final currentState = state as ChatRoomStateLoaded;
-
-      // Сообщения от текущего пользователя не приходят через WebSocket,
-      // они обрабатываются через ответ API в sendMessage
-      // Здесь обрабатываем только входящие сообщения от других пользователей
-      List<ChatMessageModel> messagesList = [message, ...currentState.chatMessages];
-
-      chatRepository.emitNewMessageToChats(message);
-      emit(currentState.copyWith(chatMessages: messagesList, isLoading: isLoading));
-      EasyDebounce.debounce('searchChatRoom', const Duration(milliseconds: 300), () => sl<ChatsCubit>().getChatRooms());
-    });
+    messagesSubscription = wsManager.directChatStream?.listen(
+      _onDirectChatEvent,
+      onError: (Object e, StackTrace st) {
+        unawaited(reportFrom(e, st, this));
+      },
+    );
 
     typingSubscription = wsManager.clientTypingStream?.listen((event) {
-      emit((state as ChatRoomStateLoaded).copyWith(isTyping: true, isLoading: isLoading));
+      _emitLoaded((current) => current.copyWith(isTyping: true, isLoading: isLoading));
       EasyDebounce.debounce(
         'client-typing',
         const Duration(seconds: 3),
-        () => emit((state as ChatRoomStateLoaded).copyWith(isTyping: false, isLoading: isLoading)),
+        () => _emitLoaded((current) => current.copyWith(isTyping: false, isLoading: isLoading)),
       );
     });
   }
 
+  void _onDirectChatEvent(ChannelReadEvent event) {
+    try {
+      if (addIncomingMessage(ChatMessageModel.fromJson(jsonDecode(event.data) as Map<String, dynamic>))) {
+        unawaited(_playIncomingSound());
+      }
+    } catch (e, st) {
+      unawaited(reportFrom(e, st, this));
+    }
+  }
+
+  Future<void> _playIncomingSound() async {
+    try {
+      await AudioPlayer().play(AssetSource('audio/recive_message.wav'));
+    } catch (_) {}
+  }
+
+  /// Добавляет входящее сообщение в текущий список, не затирая более новый state.
+  bool addIncomingMessage(ChatMessageModel message) {
+    final currentUserId = wsManager.userId;
+    if (currentUserId != null && message.authorId == currentUserId) {
+      return false;
+    }
+
+    final currentState = state;
+    if (currentState is! ChatRoomStateLoaded) {
+      return false;
+    }
+    if (currentState.chatMessages.any((existing) => existing.id == message.id && message.id != 0)) {
+      return false;
+    }
+
+    final added = _emitLoaded((current) {
+      if (current.chatMessages.any((existing) => existing.id == message.id && message.id != 0)) {
+        return current;
+      }
+      return current.copyWith(
+        chatMessages: [message, ...current.chatMessages],
+        isLoading: isLoading,
+      );
+    });
+    if (!added) {
+      return false;
+    }
+    chatRepository.emitNewMessageToChats(message);
+    return true;
+  }
+
   Future<void> _subscribeToPresenceChannel({required String chatId}) async {
     final currentState = state;
     if (currentState is! ChatRoomStateLoaded) return;
@@ -341,59 +398,38 @@ class ChatRoomCubit extends Cubit<ChatRoomState> {
     try {
       final messageUuid = uuid.v4();
       final compressedFiles = files != null ? await ImageCompressUtility.compressFilesList(List<File>.from(files)) : null;
-      AudioPlayer().play((AssetSource('audio/send_message.wav')));
+      try {
+        unawaited(AudioPlayer().play(AssetSource('audio/send_message.wav')));
+      } catch (_) {}
 
-      // Создаем временное сообщение с состоянием loading
-      emit(
-        (state as ChatRoomStateLoaded).copyWith(
+      _emitLoaded(
+        (current) => current.copyWith(
           chatMessages: [
             ChatMessageModel(
-              id: 1,
+              id: -messageUuid.hashCode.abs(),
               data: data ?? '',
               mediaData: compressedFiles,
-              authorId: 1,
+              authorId: wsManager.userId ?? 0,
               updatedAt: DateTime.now(),
               isRead: true,
               isCurrentUserAuthor: true,
-              roomId: '',
+              roomId: current.chatRoom.id,
               messageState: MessageState.loading,
               uuid: messageUuid,
             ),
-            ...(state as ChatRoomStateLoaded).chatMessages,
+            ...current.chatMessages,
           ],
         ),
       );
 
       try {
-        // Отправляем сообщение и получаем ответ от сервера с правильными данными
         final serverMessage = await chatRepository.sendMessage(data: data, files: compressedFiles, slug: userSlug);
-
-        // Заменяем локальное сообщение данными с сервера (с правильными URL файлов)
-        emit(
-          (state as ChatRoomStateLoaded).copyWith(
-            chatMessages: (state as ChatRoomStateLoaded).chatMessages.map((elem) {
-              if (elem.uuid == messageUuid) {
-                // Сохраняем UUID, состояние и обновляем данные из ответа сервера
-                // Используем значения из локального сообщения для полей, которые могут отсутствовать в ответе
-                return serverMessage.copyWith(
-                  uuid: messageUuid,
-                  messageState: MessageState.sent,
-                  isCurrentUserAuthor: true,
-                  isRead: elem.isRead,
-                  roomId: elem.roomId ?? serverMessage.roomId,
-                  authorId: serverMessage.authorId,
-                  updatedAt: serverMessage.updatedAt,
-                );
-              }
-              return elem;
-            }).toList(),
-          ),
-        );
+        _replaceOptimisticMessage(messageUuid: messageUuid, serverMessage: serverMessage);
       } catch (e, st) {
         await reportFrom(e, st, this);
-        emit(
-          (state as ChatRoomStateLoaded).copyWith(
-            chatMessages: (state as ChatRoomStateLoaded).chatMessages.map((elem) {
+        _emitLoaded(
+          (current) => current.copyWith(
+            chatMessages: current.chatMessages.map((elem) {
               if (elem.uuid == messageUuid) {
                 return elem.copyWith(messageState: MessageState.error);
               }
@@ -411,21 +447,21 @@ class ChatRoomCubit extends Cubit<ChatRoomState> {
     }
   }
 
-  Future<bool> markMessagesAsRead(String chatId, List<({int id, int index})> messagesIds) async {
+  Future<bool> markMessagesAsRead(String chatId, List<int> messageIds) async {
     try {
-      if (state is ChatRoomStateLoaded) {
-        final ids = messagesIds.map((e) => e.id).toList();
-        await chatRepository.markMessagesAsRead(chatId, ids);
-        final currentMessages = [...(state as ChatRoomStateLoaded).chatMessages];
-        for (var elem in messagesIds) {
-          currentMessages[elem.index] = currentMessages[elem.index].copyWith(isRead: true);
-        }
-        if (!isClosed) {
-          emit((state as ChatRoomStateLoaded).copyWith(chatMessages: currentMessages));
-        }
-        return true;
+      if (state is! ChatRoomStateLoaded) {
+        return false;
       }
-      return false;
+      await chatRepository.markMessagesAsRead(chatId, messageIds);
+      final idSet = messageIds.toSet();
+      _emitLoaded(
+        (current) => current.copyWith(
+          chatMessages: current.chatMessages
+              .map((message) => idSet.contains(message.id) ? message.copyWith(isRead: true) : message)
+              .toList(),
+        ),
+      );
+      return true;
     } catch (e, st) {
       await reportFrom(e, st, this);
       log(e.toString(), stackTrace: st);
@@ -433,6 +469,35 @@ class ChatRoomCubit extends Cubit<ChatRoomState> {
     }
   }
 
+  void _replaceOptimisticMessage({
+    required String messageUuid,
+    required ChatMessageModel serverMessage,
+  }) {
+    _emitLoaded((current) {
+      final sentMessage = serverMessage.copyWith(
+        uuid: messageUuid,
+        messageState: MessageState.sent,
+        isCurrentUserAuthor: true,
+      );
+      final withoutDuplicate = current.chatMessages.where((message) => message.id != serverMessage.id || message.uuid == messageUuid);
+      final hasOptimistic = withoutDuplicate.any((message) => message.uuid == messageUuid);
+      if (!hasOptimistic) {
+        return current.copyWith(chatMessages: [sentMessage, ...withoutDuplicate]);
+      }
+      return current.copyWith(
+        chatMessages: withoutDuplicate.map((elem) {
+          if (elem.uuid == messageUuid) {
+            return sentMessage.copyWith(
+              isRead: elem.isRead,
+              roomId: elem.roomId?.isNotEmpty == true ? elem.roomId : serverMessage.roomId,
+            );
+          }
+          return elem;
+        }).toList(),
+      );
+    });
+  }
+
   Future<String?> getPaymentLink() async {
     try {
       final planId = (state as ChatRoomStateLoaded).chatRoom.chosenPlan?.id;
@@ -477,14 +542,16 @@ class ChatRoomCubit extends Cubit<ChatRoomState> {
 
   @override
   Future<void> close() async {
+    EasyDebounce.cancel('client-typing');
+    EasyDebounce.cancel('searchChatRoom');
     if (_presenceOnlineListener != null) {
       wsManager.removePresenceDirectChatListener(_presenceOnlineListener!);
       _presenceOnlineListener = null;
     }
     wsManager.unsubscribeDirectChatChannel();
     _timer?.cancel();
-    messagesSubscription?.cancel();
-    typingSubscription?.cancel();
+    await messagesSubscription?.cancel();
+    await typingSubscription?.cancel();
     super.close();
   }
 
@@ -501,67 +568,46 @@ class ChatRoomCubit extends Cubit<ChatRoomState> {
       final message = currentState.chatMessages.firstWhereOrNull((message) => message.uuid == uuid);
       if (message != null) {
         try {
-          AudioPlayer().play((AssetSource('audio/send_message.wav')));
+          try {
+            unawaited(AudioPlayer().play(AssetSource('audio/send_message.wav')));
+          } catch (_) {}
 
-          // Определяем, есть ли файлы для повторной отправки
           List<File>? filesToResend;
           if (message.mediaData != null && message.mediaData!.isNotEmpty) {
-            // Если mediaData содержит File объекты, используем их
-            // Если это строки (URL), то файлы уже на сервере и повторная отправка не нужна
             final firstItem = message.mediaData!.first;
             if (firstItem is File) {
               filesToResend = message.mediaData!.cast<File>();
             }
           }
 
-          emit(
-            (state as ChatRoomStateLoaded).copyWith(
+          _emitLoaded(
+            (current) => current.copyWith(
               chatMessages: [
                 ChatMessageModel(
-                  id: 1,
+                  id: -uuid.hashCode.abs(),
                   data: message.data,
                   mediaData: message.mediaData,
-                  authorId: 1,
+                  authorId: wsManager.userId ?? 0,
                   updatedAt: DateTime.now(),
                   isRead: true,
                   isCurrentUserAuthor: true,
-                  roomId: '',
+                  roomId: current.chatRoom.id,
                   messageState: MessageState.loading,
                   uuid: uuid,
                 ),
-                ...(state as ChatRoomStateLoaded).chatMessages.where((message) => message.uuid != uuid),
+                ...current.chatMessages.where((item) => item.uuid != uuid),
               ],
             ),
           );
 
           try {
-            // Отправляем сообщение и получаем ответ от сервера
             final serverMessage = await chatRepository.sendMessage(data: message.data, files: filesToResend, slug: userSlug);
-
-            // Заменяем локальное сообщение данными с сервера
-            emit(
-              (state as ChatRoomStateLoaded).copyWith(
-                chatMessages: (state as ChatRoomStateLoaded).chatMessages.map((elem) {
-                  if (elem.uuid == uuid) {
-                    return serverMessage.copyWith(
-                      uuid: uuid,
-                      messageState: MessageState.sent,
-                      isCurrentUserAuthor: true,
-                      isRead: elem.isRead,
-                      roomId: elem.roomId ?? serverMessage.roomId,
-                      authorId: serverMessage.authorId,
-                      updatedAt: serverMessage.updatedAt,
-                    );
-                  }
-                  return elem;
-                }).toList(),
-              ),
-            );
+            _replaceOptimisticMessage(messageUuid: uuid, serverMessage: serverMessage);
           } catch (e, st) {
             await reportFrom(e, st, this);
-            emit(
-              (state as ChatRoomStateLoaded).copyWith(
-                chatMessages: (state as ChatRoomStateLoaded).chatMessages.map((elem) {
+            _emitLoaded(
+              (current) => current.copyWith(
+                chatMessages: current.chatMessages.map((elem) {
                   if (elem.uuid == uuid) {
                     return elem.copyWith(messageState: MessageState.error);
                   }

lib/features/chat/presentation/bloc/chats_cubit/chats_cubit.dart

@@ -141,8 +141,8 @@ class ChatsCubit extends Cubit<ChatsState> {
 
   Future<void> _handleNewMessageViaSocket(dynamic data) async {
     try {
-      final json = jsonDecode(data);
-      final message = ChatMessageModel.fromJson(json['message']);
+      final json = jsonDecode(data) as Map<String, dynamic>;
+      final message = ChatMessageModel.fromJson(json);
 
       final currentChatRooms = [...state.chatRooms];
 

lib/features/chat/presentation/pages/chat_screen.dart

@@ -56,8 +56,7 @@ class _ChatScreenState extends State<ChatScreen> {
   final scrollController = ScrollController();
   final TextEditingController textEditingController = TextEditingController();
   final ChatRoomCubit chatRoomCubit = sl<ChatRoomCubit>();
-  final List<({int id, int index})> unreadMessagesIds = [];
-  final List<({int id, int index})> unreadMessagesIdsCopy = [];
+  final List<int> unreadMessageIds = [];
   final showableChatBadgeStates = ChosenPlanStateEnum.getShowableChatBadgeStates();
   late AppLifecycleListener _listener;
   List<File> pickedFiles = [];
@@ -116,6 +115,7 @@ class _ChatScreenState extends State<ChatScreen> {
 
   @override
   void dispose() {
+    EasyDebounce.cancel('mark_as_read');
     scrollController.dispose();
     _listener.dispose();
     chatRoomCubit.close();
@@ -337,21 +337,15 @@ class _ChatScreenState extends State<ChatScreen> {
                                         isCurrentUser: isCurrentUserAuthor,
                                       ),
                                     VisibilityDetector(
-                                      key: Key(message.id.toString()),
+                                      key: ValueKey('msg-${message.uuid ?? message.id}'),
                                       onVisibilityChanged: (VisibilityInfo info) {
-                                        if (!message.isRead && !isCurrentUserAuthor && info.visibleFraction > 0) {
-                                          unreadMessagesIdsCopy.add((id: message.id, index: index));
-                                          EasyDebounce.debounce('mark_as_read', const Duration(milliseconds: 500), () async {
-                                            final result = await context.read<ChatRoomCubit>().markMessagesAsRead(state.chatRoom.id, unreadMessagesIdsCopy);
-                                            if (!result) return;
-                                            if (!context.mounted) return;
-                                            context.read<ChatsCubit>().reduceUnreadMessagesCount(
-                                              chatId: state.chatRoom.id,
-                                              count: unreadMessagesIdsCopy.length,
-                                            );
-                                            unreadMessagesIdsCopy.clear();
-                                          });
-                                        }
+                                        _queueMarkAsRead(
+                                          context: context,
+                                          chatId: state.chatRoom.id,
+                                          message: message,
+                                          isCurrentUserAuthor: isCurrentUserAuthor,
+                                          visibleFraction: info.visibleFraction,
+                                        );
                                       },
                                       child: _buildMessageBubble(
                                         context: context,
@@ -391,18 +385,15 @@ class _ChatScreenState extends State<ChatScreen> {
                                       isCurrentUser: isCurrentUserAuthor,
                                     ),
                                   VisibilityDetector(
-                                    key: Key(message.id.toString()),
+                                    key: ValueKey('msg-${message.uuid ?? message.id}'),
                                     onVisibilityChanged: (VisibilityInfo info) {
-                                      if (!message.isRead && !isCurrentUserAuthor && info.visibleFraction > 0) {
-                                        unreadMessagesIds.add((id: message.id, index: index));
-                                        EasyDebounce.debounce('mark_as_read', const Duration(milliseconds: 500), () async {
-                                          final result = await context.read<ChatRoomCubit>().markMessagesAsRead(state.chatRoom.id, unreadMessagesIds);
-                                          if (!result) return;
-                                          if (!context.mounted) return;
-                                          context.read<ChatsCubit>().reduceUnreadMessagesCount(chatId: state.chatRoom.id, count: unreadMessagesIds.length);
-                                          unreadMessagesIds.clear();
-                                        });
-                                      }
+                                      _queueMarkAsRead(
+                                        context: context,
+                                        chatId: state.chatRoom.id,
+                                        message: message,
+                                        isCurrentUserAuthor: isCurrentUserAuthor,
+                                        visibleFraction: info.visibleFraction,
+                                      );
                                     },
                                     child: _buildMessageBubble(
                                       context: context,
@@ -793,6 +784,33 @@ class _ChatScreenState extends State<ChatScreen> {
     );
   }
 
+  void _queueMarkAsRead({
+    required BuildContext context,
+    required String chatId,
+    required ChatMessageModel message,
+    required bool isCurrentUserAuthor,
+    required double visibleFraction,
+  }) {
+    if (message.isRead || isCurrentUserAuthor || visibleFraction <= 0) {
+      return;
+    }
+    if (!unreadMessageIds.contains(message.id)) {
+      unreadMessageIds.add(message.id);
+    }
+    EasyDebounce.debounce('mark_as_read', const Duration(milliseconds: 500), () async {
+      final ids = List<int>.from(unreadMessageIds);
+      if (ids.isEmpty) {
+        return;
+      }
+      unreadMessageIds.clear();
+      final result = await context.read<ChatRoomCubit>().markMessagesAsRead(chatId, ids);
+      if (!result || !context.mounted) {
+        return;
+      }
+      context.read<ChatsCubit>().reduceUnreadMessagesCount(chatId: chatId, count: ids.length);
+    });
+  }
+
   Widget _buildMessageBubble({
     required BuildContext context,
     required ChatRoomStateLoaded state,

test/features/chat/presentation/bloc/chat_room_cubit_test.dart

@@ -200,7 +200,7 @@ void main() {
 
     test('markMessagesAsRead - отмечает сообщения как прочитанные', () async {
       const chatId = 'chat-id';
-      final messagesIds = [(id: 1, index: 0)];
+      final messagesIds = [1];
 
       // Сначала инициализируем чат
       when(mockChatRepository.getChatRoom(any)).thenAnswer((_) async => testChatRoom);
@@ -1443,5 +1443,218 @@ void main() {
 
       verifyNever(mockChatRepository.getChatMessages(any, any));
     });
+
+    test('addIncomingMessage - добавляет входящее сообщение в начало списка', () async {
+      const uuid = 'room-uuid';
+      when(mockChatRepository.getChatRoom(uuid)).thenAnswer((_) async => testChatRoom);
+      when(mockChatRepository.getChatMessages(uuid, 1)).thenAnswer((_) async => testChatMessages);
+      when(mockWsManager.client).thenReturn(null);
+      when(mockWsManager.initPrivateDirectChatChannel(any)).thenAnswer((_) async => Future.value());
+      when(mockWsManager.subscribeDirectChatChannel()).thenReturn(null);
+      when(mockWsManager.directChatStream).thenReturn(null);
+      when(mockWsManager.clientTypingStream).thenReturn(null);
+      when(mockChatRepository.getChosenPlan(id: anyNamed('id'))).thenAnswer((_) async => throw Exception('No plan'));
+
+      await chatRoomCubit.initChat(uuid);
+      await Future.delayed(const Duration(milliseconds: 200));
+
+      final incoming = ChatMessageModel(
+        id: 99,
+        data: 'Incoming socket message',
+        mediaData: null,
+        authorId: 2,
+        updatedAt: DateTime.now(),
+        isRead: false,
+        isCurrentUserAuthor: false,
+        roomId: 'chat-id',
+      );
+
+      chatRoomCubit.addIncomingMessage(incoming);
+
+      final state = chatRoomCubit.state as ChatRoomStateLoaded;
+      expect(state.chatMessages.first.id, 99);
+      expect(state.chatMessages.length, 2);
+    });
+
+    test('addIncomingMessage - не добавляет дубликат по id', () async {
+      const uuid = 'room-uuid';
+      when(mockChatRepository.getChatRoom(uuid)).thenAnswer((_) async => testChatRoom);
+      when(mockChatRepository.getChatMessages(uuid, 1)).thenAnswer((_) async => testChatMessages);
+      when(mockWsManager.client).thenReturn(null);
+      when(mockWsManager.initPrivateDirectChatChannel(any)).thenAnswer((_) async => Future.value());
+      when(mockWsManager.subscribeDirectChatChannel()).thenReturn(null);
+      when(mockWsManager.directChatStream).thenReturn(null);
+      when(mockWsManager.clientTypingStream).thenReturn(null);
+      when(mockChatRepository.getChosenPlan(id: anyNamed('id'))).thenAnswer((_) async => throw Exception('No plan'));
+
+      await chatRoomCubit.initChat(uuid);
+      await Future.delayed(const Duration(milliseconds: 200));
+
+      final incoming = ChatMessageModel(
+        id: 1,
+        data: 'Duplicate of existing',
+        mediaData: null,
+        authorId: 2,
+        updatedAt: DateTime.now(),
+        isRead: false,
+        isCurrentUserAuthor: false,
+        roomId: 'chat-id',
+      );
+
+      chatRoomCubit.addIncomingMessage(incoming);
+
+      final state = chatRoomCubit.state as ChatRoomStateLoaded;
+      expect(state.chatMessages.length, 1);
+      expect(state.chatMessages.first.data, 'Test message');
+    });
+
+    test('addIncomingMessage - игнорирует собственные сообщения', () async {
+      const uuid = 'room-uuid';
+      when(mockChatRepository.getChatRoom(uuid)).thenAnswer((_) async => testChatRoom);
+      when(mockChatRepository.getChatMessages(uuid, 1)).thenAnswer((_) async => testChatMessages);
+      when(mockWsManager.client).thenReturn(null);
+      when(mockWsManager.userId).thenReturn(1);
+      when(mockWsManager.initPrivateDirectChatChannel(any)).thenAnswer((_) async => Future.value());
+      when(mockWsManager.subscribeDirectChatChannel()).thenReturn(null);
+      when(mockWsManager.directChatStream).thenReturn(null);
+      when(mockWsManager.clientTypingStream).thenReturn(null);
+      when(mockChatRepository.getChosenPlan(id: anyNamed('id'))).thenAnswer((_) async => throw Exception('No plan'));
+
+      await chatRoomCubit.initChat(uuid);
+      await Future.delayed(const Duration(milliseconds: 200));
+
+      chatRoomCubit.addIncomingMessage(
+        ChatMessageModel(
+          id: 77,
+          data: 'Own echo',
+          mediaData: null,
+          authorId: 1,
+          updatedAt: DateTime.now(),
+          isRead: true,
+          isCurrentUserAuthor: true,
+          roomId: 'chat-id',
+        ),
+      );
+
+      final state = chatRoomCubit.state as ChatRoomStateLoaded;
+      expect(state.chatMessages.length, 1);
+      expect(state.chatMessages.any((message) => message.id == 77), isFalse);
+    });
+
+    test('updateChosenPlanState - не затирает сообщения, пришедшие во время запроса', () async {
+      const uuid = 'room-uuid';
+      const planId = 123;
+      final chatRoomWithPlan = testChatRoom.copyWith(chosenPlanId: planId);
+      const testPlan = PlanModel(
+        id: 1,
+        slug: 'test-plan',
+        name: 'Test Plan',
+        image: 'https://example.com/image.jpg',
+        duration: 7,
+        state: 'active',
+        type: PlanType(fit: true, slim: false, behavior: false, wellness: false),
+        details: 'Test details',
+        packetsCount: 3,
+        author: Author(authorSlug: 'author', authorName: 'Author Name', authorProfileIcon: 'https://example.com/icon.jpg', authorIsMe: false),
+        totalLikes: 10,
+        totalViews: 100,
+        totalComments: 5,
+        isFavorite: false,
+        minPriceOfOptions: 1000,
+        maxPriceOfOptions: 2000,
+        wasLikedByMe: false,
+      );
+      final chosenPlan = ChosenPlanModel(
+        id: planId,
+        optionType: 'options1',
+        startDate: DateTime.now(),
+        chosenPlanState: ChosenPlanStateEnum.stateStarted,
+        chatId: 'chat-1',
+        consultantId: 1,
+        plan: testPlan,
+      );
+      final updatedPlan = chosenPlan.copyWith(chosenPlanState: ChosenPlanStateEnum.statePaid);
+      final planCompleter = Completer<ChosenPlanModel>();
+      var getChosenPlanCalls = 0;
+
+      when(mockChatRepository.getChatRoom(uuid)).thenAnswer((_) async => chatRoomWithPlan);
+      when(mockChatRepository.getChatMessages(uuid, 1)).thenAnswer((_) async => testChatMessages);
+      when(mockChatRepository.getChosenPlan(id: planId)).thenAnswer((_) {
+        getChosenPlanCalls++;
+        if (getChosenPlanCalls == 1) {
+          return Future.value(chosenPlan);
+        }
+        return planCompleter.future;
+      });
+      when(mockWsManager.client).thenReturn(null);
+      when(mockWsManager.initPrivateDirectChatChannel(any)).thenAnswer((_) async => Future.value());
+      when(mockWsManager.subscribeDirectChatChannel()).thenReturn(null);
+      when(mockWsManager.directChatStream).thenReturn(null);
+      when(mockWsManager.clientTypingStream).thenReturn(null);
+
+      await chatRoomCubit.initChat(uuid);
+      await Future.delayed(const Duration(milliseconds: 200));
+
+      final updateFuture = chatRoomCubit.updateChosenPlanState();
+      chatRoomCubit.addIncomingMessage(
+        ChatMessageModel(
+          id: 55,
+          data: 'Arrived during plan poll',
+          mediaData: null,
+          authorId: 2,
+          updatedAt: DateTime.now(),
+          isRead: false,
+          isCurrentUserAuthor: false,
+          roomId: 'chat-id',
+        ),
+      );
+
+      planCompleter.complete(updatedPlan);
+      await updateFuture;
+
+      final state = chatRoomCubit.state as ChatRoomStateLoaded;
+      expect(state.chatMessages.any((message) => message.id == 55), isTrue);
+      expect(state.chatRoom.chosenPlan?.chosenPlanState, ChosenPlanStateEnum.statePaid);
+    });
+
+    test('markMessagesAsRead - не затирает сообщения, пришедшие во время запроса', () async {
+      const uuid = 'room-uuid';
+      const chatId = 'chat-id';
+      final markCompleter = Completer<void>();
+
+      when(mockChatRepository.getChatRoom(uuid)).thenAnswer((_) async => testChatRoom);
+      when(mockChatRepository.getChatMessages(uuid, 1)).thenAnswer((_) async => testChatMessages);
+      when(mockWsManager.client).thenReturn(null);
+      when(mockWsManager.initPrivateDirectChatChannel(any)).thenAnswer((_) async => Future.value());
+      when(mockWsManager.subscribeDirectChatChannel()).thenReturn(null);
+      when(mockWsManager.directChatStream).thenReturn(null);
+      when(mockWsManager.clientTypingStream).thenReturn(null);
+      when(mockChatRepository.getChosenPlan(id: anyNamed('id'))).thenAnswer((_) async => throw Exception('No plan'));
+      when(mockChatRepository.markMessagesAsRead(chatId, [1])).thenAnswer((_) => markCompleter.future);
+
+      await chatRoomCubit.initChat(uuid);
+      await Future.delayed(const Duration(milliseconds: 200));
+
+      final markFuture = chatRoomCubit.markMessagesAsRead(chatId, [1]);
+      chatRoomCubit.addIncomingMessage(
+        ChatMessageModel(
+          id: 88,
+          data: 'Arrived during mark as read',
+          mediaData: null,
+          authorId: 2,
+          updatedAt: DateTime.now(),
+          isRead: false,
+          isCurrentUserAuthor: false,
+          roomId: chatId,
+        ),
+      );
+
+      markCompleter.complete();
+      await markFuture;
+
+      final state = chatRoomCubit.state as ChatRoomStateLoaded;
+      expect(state.chatMessages.any((message) => message.id == 88), isTrue);
+      expect(state.chatMessages.firstWhere((message) => message.id == 1).isRead, isTrue);
+    });
   });
 }