Hasan/eatvio_app
1
0
Fork 0
forked from eatvio/eatvio_app
Code Pull Requests Activity

feat(chat_messages): fixed data race for chat mesages state

Browse Source
This commit is contained in:
misha
2026-08-14 11:05:03 +03:00
parent 179fc0ad99
commit 52a01bee12
5 changed files with 436 additions and 147 deletions
Download Patch File Download Diff File Expand all files Collapse all files

276
lib/features/chat/presentation/bloc/chat_room_cubit/chat_room_cubit.dart
View File

@@ -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);
}

4
lib/features/chat/presentation/bloc/chats_cubit/chats_cubit.dart
View File

@@ -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];

72
lib/features/chat/presentation/pages/chat_screen.dart
View File

@@ -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,