From 52a01bee12f5449c483389e308855fac81f2a969 Mon Sep 17 00:00:00 2001 From: misha Date: Fri, 14 Aug 2026 11:05:03 +0300 Subject: [PATCH] feat(chat_messages): fixed data race for chat mesages state --- lib/core/websockets/ws_manager.dart | 16 +- .../bloc/chat_room_cubit/chat_room_cubit.dart | 276 ++++++++++-------- .../bloc/chats_cubit/chats_cubit.dart | 4 +- .../chat/presentation/pages/chat_screen.dart | 72 +++-- .../bloc/chat_room_cubit_test.dart | 215 +++++++++++++- 5 files changed, 436 insertions(+), 147 deletions(-) diff --git a/lib/core/websockets/ws_manager.dart b/lib/core/websockets/ws_manager.dart index 3631e0ac..9ab8fe19 100644 --- a/lib/core/websockets/ws_manager.dart +++ b/lib/core/websockets/ws_manager.dart @@ -38,13 +38,16 @@ class WsManager { StreamSubscription? _eventSubscription; StreamSubscription? _connectionEstablishedSubscription; + Stream? _directChatStream; + Stream? _clientTypingStream; + Stream? get newMessagesStream => _privateMessagesChannel?.bind('App\\Events\\ChatMessageUpdated'); Stream? get newNotificationsStream => _privateNotificationChannel?.bind('Illuminate\\Notifications\\Events\\BroadcastNotificationCreated'); - Stream? get directChatStream => _privateDirectChatChannel?.bind('App\\Events\\ChatMessageUpdated'); + Stream? get directChatStream => _directChatStream; - Stream? get clientTypingStream => _privateDirectChatChannel?.bind('client-typing'); + Stream? get clientTypingStream => _clientTypingStream; Future 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 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 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; } } diff --git a/lib/features/chat/presentation/bloc/chat_room_cubit/chat_room_cubit.dart b/lib/features/chat/presentation/bloc/chat_room_cubit/chat_room_cubit.dart index 452adb18..27b81f50 100644 --- a/lib/features/chat/presentation/bloc/chat_room_cubit/chat_room_cubit.dart +++ b/lib/features/chat/presentation/bloc/chat_room_cubit/chat_room_cubit.dart @@ -54,16 +54,16 @@ class ChatRoomCubit extends Cubit { 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 { } void _setCard({required List 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 { 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 { hasCard: false, ), ); + await _subscribeToPresenceChannel(chatId: chatRoom.id); if (chosenPlan?.consultantId == chatRoom.users.sender.id) { await checkCards(); @@ -154,13 +155,15 @@ class ChatRoomCubit extends Cubit { Future 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 { } } + 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 _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 messagesList = [message, ...currentState.chatMessages]; - - chatRepository.emitNewMessageToChats(message); - emit(currentState.copyWith(chatMessages: messagesList, isLoading: isLoading)); - EasyDebounce.debounce('searchChatRoom', const Duration(milliseconds: 300), () => sl().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))) { + unawaited(_playIncomingSound()); + } + } catch (e, st) { + unawaited(reportFrom(e, st, this)); + } + } + + Future _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 _subscribeToPresenceChannel({required String chatId}) async { final currentState = state; if (currentState is! ChatRoomStateLoaded) return; @@ -341,59 +398,38 @@ class ChatRoomCubit extends Cubit { try { final messageUuid = uuid.v4(); final compressedFiles = files != null ? await ImageCompressUtility.compressFilesList(List.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 { } } - Future markMessagesAsRead(String chatId, List<({int id, int index})> messagesIds) async { + Future markMessagesAsRead(String chatId, List 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 { } } + 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 getPaymentLink() async { try { final planId = (state as ChatRoomStateLoaded).chatRoom.chosenPlan?.id; @@ -477,14 +542,16 @@ class ChatRoomCubit extends Cubit { @override Future 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 { 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? filesToResend; if (message.mediaData != null && message.mediaData!.isNotEmpty) { - // Если mediaData содержит File объекты, используем их - // Если это строки (URL), то файлы уже на сервере и повторная отправка не нужна final firstItem = message.mediaData!.first; if (firstItem is File) { filesToResend = message.mediaData!.cast(); } } - 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); } diff --git a/lib/features/chat/presentation/bloc/chats_cubit/chats_cubit.dart b/lib/features/chat/presentation/bloc/chats_cubit/chats_cubit.dart index 52e1631e..10f91095 100644 --- a/lib/features/chat/presentation/bloc/chats_cubit/chats_cubit.dart +++ b/lib/features/chat/presentation/bloc/chats_cubit/chats_cubit.dart @@ -141,8 +141,8 @@ class ChatsCubit extends Cubit { Future _handleNewMessageViaSocket(dynamic data) async { try { - final json = jsonDecode(data); - final message = ChatMessageModel.fromJson(json['message']); + final json = jsonDecode(data) as Map; + final message = ChatMessageModel.fromJson(json); final currentChatRooms = [...state.chatRooms]; diff --git a/lib/features/chat/presentation/pages/chat_screen.dart b/lib/features/chat/presentation/pages/chat_screen.dart index 432bebdb..4ef7ba4e 100644 --- a/lib/features/chat/presentation/pages/chat_screen.dart +++ b/lib/features/chat/presentation/pages/chat_screen.dart @@ -56,8 +56,7 @@ class _ChatScreenState extends State { final scrollController = ScrollController(); final TextEditingController textEditingController = TextEditingController(); final ChatRoomCubit chatRoomCubit = sl(); - final List<({int id, int index})> unreadMessagesIds = []; - final List<({int id, int index})> unreadMessagesIdsCopy = []; + final List unreadMessageIds = []; final showableChatBadgeStates = ChosenPlanStateEnum.getShowableChatBadgeStates(); late AppLifecycleListener _listener; List pickedFiles = []; @@ -116,6 +115,7 @@ class _ChatScreenState extends State { @override void dispose() { + EasyDebounce.cancel('mark_as_read'); scrollController.dispose(); _listener.dispose(); chatRoomCubit.close(); @@ -337,21 +337,15 @@ class _ChatScreenState extends State { 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().markMessagesAsRead(state.chatRoom.id, unreadMessagesIdsCopy); - if (!result) return; - if (!context.mounted) return; - context.read().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 { 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().markMessagesAsRead(state.chatRoom.id, unreadMessagesIds); - if (!result) return; - if (!context.mounted) return; - context.read().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 { ); } + 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.from(unreadMessageIds); + if (ids.isEmpty) { + return; + } + unreadMessageIds.clear(); + final result = await context.read().markMessagesAsRead(chatId, ids); + if (!result || !context.mounted) { + return; + } + context.read().reduceUnreadMessagesCount(chatId: chatId, count: ids.length); + }); + } + Widget _buildMessageBubble({ required BuildContext context, required ChatRoomStateLoaded state, diff --git a/test/features/chat/presentation/bloc/chat_room_cubit_test.dart b/test/features/chat/presentation/bloc/chat_room_cubit_test.dart index d1f14bba..38793443 100644 --- a/test/features/chat/presentation/bloc/chat_room_cubit_test.dart +++ b/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(); + 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(); + + 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); + }); }); }