diff --git a/packages/stream_chat/CHANGELOG.md b/packages/stream_chat/CHANGELOG.md index 02246f10b..1e2def207 100644 --- a/packages/stream_chat/CHANGELOG.md +++ b/packages/stream_chat/CHANGELOG.md @@ -3,6 +3,7 @@ 🔄 Changed - Raised minimum Dart SDK to `^3.12.0`. +- Restructured the internal channel event handling. ## 10.3.0 diff --git a/packages/stream_chat/lib/src/client/channel.dart b/packages/stream_chat/lib/src/client/channel.dart index 5cfae938a..b21a831b9 100644 --- a/packages/stream_chat/lib/src/client/channel.dart +++ b/packages/stream_chat/lib/src/client/channel.dart @@ -5,6 +5,8 @@ import 'dart:math' as math; import 'package:collection/collection.dart'; import 'package:rxdart/rxdart.dart'; +import 'package:stream_chat/src/client/channel_event_handler.dart'; +import 'package:stream_chat/src/client/channel_state_mutations.dart'; import 'package:stream_chat/src/client/retry_queue.dart'; import 'package:stream_chat/src/core/util/utils.dart'; import 'package:stream_chat/stream_chat.dart'; @@ -2481,73 +2483,17 @@ class ChannelClientState { // Update the persistence storage with the seeded channel state. _debouncedUpdatePersistenceChannelState.call([channelState]); - // region TYPING EVENTS - _listenTypingEvents(); - // endregion - - // region MESSAGE EVENTS - _listenMessageNew(); - _listenMessageDeleted(); - _listenMessageUpdated(); - // endregion - - // region DRAFT EVENTS - _listenDraftUpdated(); - _listenDraftDeleted(); - // endregion - - // region REACTION EVENTS - _listenReactionNew(); - _listenReactionUpdated(); - _listenReactionDeleted(); - // endregion - - // region POLL EVENTS - _listenPollCreated(); - _listenPollUpdated(); - _listenPollClosed(); - _listenPollAnswerCasted(); - _listenPollVoteCasted(); - _listenPollVoteChanged(); - _listenPollAnswerRemoved(); - _listenPollVoteRemoved(); - // endregion - - // region READ EVENTS - _listenReadEvents(); - // endregion - - // region CHANNEL EVENTS - _listenChannelTruncated(); - _listenChannelUpdated(); - _listenChannelMessageCount(); - // endregion - - // region MEMBER EVENTS - _listenMemberAdded(); - _listenMemberRemoved(); - _listenMemberUpdated(); - _listenMemberBanned(); - _listenMemberUnbanned(); - _listenUserMessagesDeleted(); - // endregion - - // region USER WATCHING EVENTS - _listenUserStartWatching(); - _listenUserStopWatching(); - // endregion - - // region REMINDER EVENTS - _listenReminderCreated(); - _listenReminderUpdated(); - _listenReminderDeleted(); - // endregion - - // region LOCATION EVENTS - _listenLocationShared(); - _listenLocationUpdated(); - _listenLocationExpired(); - // endregion + final mutations = ChannelStateMutations( + channel: _channel, + state: this, + upsertTypingEvent: _upsertTypingEvent, + removeTypingEvent: _removeTypingEvent, + removeWatcher: _removeWatcher, + updateMember: _updateMember, + deleteMessagesFromUser: _deleteMessagesFromUser, + ); + final handler = ChannelEventHandler(channel: _channel, mutations: mutations); + _subscriptions.add(_channel.on().listen(handler.handleEvent)); _startCleaningStaleTypingEvents(); @@ -2555,8 +2501,6 @@ class ChannelClientState { _startCleaningExpiredLocations(); - _listenChannelPushPreferenceUpdated(); - final persistenceClient = _client.chatPersistenceClient; persistenceClient ?.getChannelThreads(_channel.cid!) @@ -2571,212 +2515,21 @@ class ChannelClientState { StreamChatClient get _client => _channel._client; final _subscriptions = CompositeSubscription(); - void _listenMemberAdded() { - _subscriptions.add( - _channel.on(EventType.memberAdded).listen((Event e) { - final member = e.member!; - final existingMembers = channelState.members ?? []; - - updateChannelState( - channelState.copyWith( - members: [...existingMembers, member], - ), - ); - }), - ); - } - - void _listenMemberRemoved() { - _subscriptions.add( - _channel.on(EventType.memberRemoved).listen((Event e) { - final user = e.user!; - final existingRead = channelState.read ?? []; - final existingMembers = channelState.members ?? []; - - updateChannelState( - channelState.copyWith( - read: [...existingRead.where((r) => r.user.id != user.id)], - members: [...existingMembers.where((m) => m.userId != user.id)], - ), - ); - }), - ); - } - - void _listenMemberUpdated() { - _subscriptions - // Listen to events containing member users - ..add( - _channel.on().listen( - (event) { - final user = event.user; - if (user == null) return; - - final existingMembers = [...?channelState.members]; - final existingMembership = channelState.membership; - - // Return if the user is not a existing member of the channel. - if (!existingMembers.any((m) => m.userId == user.id)) return; - - Member? maybeUpdateMemberUser(Member? existingMember) { - if (existingMember == null) return null; - if (existingMember.userId == user.id) { - return existingMember.copyWith(user: user); - } - return existingMember; - } - - updateChannelState( - channelState.copyWith( - membership: maybeUpdateMemberUser(existingMembership), - members: [...existingMembers.map(maybeUpdateMemberUser).nonNulls], - ), - ); - }, - ), - ) - // Listen to member updated events. - ..add( - _channel.on(EventType.memberUpdated).listen( - (Event e) { - final member = e.member!; - final existingMembers = channelState.members ?? []; - final existingMembership = channelState.membership; - - Member? maybeUpdateMember(Member? existingMember) { - if (existingMember == null) return null; - if (existingMember.userId == member.userId) return member; - return existingMember; - } - - updateChannelState( - channelState.copyWith( - membership: maybeUpdateMember(existingMembership), - members: [...existingMembers.map(maybeUpdateMember).nonNulls], - ), - ); - }, - ), - ); - } - - void _listenChannelUpdated() { - _subscriptions.add( - _channel.on(EventType.channelUpdated).listen((Event e) { - final channel = e.channel!; - updateChannelState( - channelState.copyWith( - channel: channelState.channel?.merge(channel), - members: channel.members, - ), - ); - }), - ); - } - - void _listenChannelMessageCount() { - _subscriptions.add( - _channel.on().listen( - (Event e) { - final messageCount = e.channelMessageCount; - if (messageCount == null) return; - - updateChannelState( - channelState.copyWith( - channel: channelState.channel?.copyWith( - messageCount: messageCount, - ), - ), - ); - }, - ), - ); - } - - void _listenChannelTruncated() { - _subscriptions.add( - _channel.on(EventType.channelTruncated, EventType.notificationChannelTruncated).listen((event) async { - final channel = event.channel!; - await _client.chatPersistenceClient?.deleteMessageByCid(channel.cid); - truncate(); - if (event.message != null) { - updateMessage(event.message!); - } - }), - ); - } - - void _listenMemberBanned() { - _subscriptions.add( - _channel - .on(EventType.userBanned) - .where((it) => it.cid != null) // filters channel ban from app ban - .listen( - (event) async { - final user = event.user!; - final member = await _channel - .queryMembers(filter: Filter.equal('id', user.id)) - .then((it) => it.members.first); - - _updateMember(member); - }, - ), - ); - } - - void _listenUserStartWatching() { - _subscriptions.add( - _channel.on(EventType.userWatchingStart).listen((event) { - final watcher = event.user; - if (watcher != null) { - final existingWatchers = channelState.watchers; - updateChannelState( - channelState.copyWith( - watchers: [ - watcher, - ...?existingWatchers?.where((user) => user.id != watcher.id), - ], - watcherCount: event.watcherCount, - ), - ); - } - }), - ); - } - - void _listenUserStopWatching() { - _subscriptions.add( - _channel.on(EventType.userWatchingStop).listen((event) { - final watcher = event.user; - if (watcher != null) { - final existingWatchers = channelState.watchers ?? const []; - _channelState = channelState.copyWith( - watchers: existingWatchers.where((user) => user.id != watcher.id).toList(), - watcherCount: event.watcherCount, - ); - } - }), - ); - } - - void _listenMemberUnbanned() { - _subscriptions.add( - _channel - .on(EventType.userUnbanned) - .where((it) => it.cid != null) // filters channel ban from app ban - .listen( - (event) async { - final user = event.user!; - final member = await _channel - .queryMembers(filter: Filter.equal('id', user.id)) - .then((it) => it.members.first); - - _updateMember(member); - }, - ), + /// Removes the [watcher] from the channel state, optionally updating the + /// [watcherCount] when provided. + void _removeWatcher(User watcher, {int? watcherCount}) { + // Writes the state directly: the watcher list merge in + // [updateChannelState] would undo the removal. + final existingWatchers = channelState.watchers ?? const []; + _channelState = channelState.copyWith( + watchers: existingWatchers.where((user) => user.id != watcher.id).toList(), + watcherCount: watcherCount, ); } + /// Replaces the member matching [member]'s user id in the channel state. + /// + /// Does nothing if no member with the same user id exists. void _updateMember(Member member) { final currentMembers = [...members]; final memberIndex = currentMembers.indexWhere( @@ -2819,274 +2572,6 @@ class ChannelClientState { _retryQueue.add(failedMessages); } - Message? _findPollMessage(String pollId) { - final message = messages.firstWhereOrNull((it) => it.pollId == pollId); - if (message != null) return message; - - final threadMessage = threads.values.flattened.firstWhereOrNull((it) { - return it.pollId == pollId; - }); - - return threadMessage; - } - - void _listenPollCreated() { - _subscriptions.add( - _channel.on(EventType.pollCreated).listen((event) { - final message = event.message; - if (message == null || message.poll == null) return; - - return addNewMessage(message); - }), - ); - } - - void _listenPollUpdated() { - _subscriptions.add( - _channel.on(EventType.pollUpdated).listen((event) { - final eventPoll = event.poll; - if (eventPoll == null) return; - - final pollMessage = _findPollMessage(eventPoll.id); - if (pollMessage == null) return; - - final oldPoll = pollMessage.poll; - - final latestAnswers = oldPoll?.latestAnswers ?? eventPoll.latestAnswers; - final ownVotesAndAnswers = oldPoll?.ownVotesAndAnswers ?? eventPoll.ownVotesAndAnswers; - - final poll = eventPoll.copyWith( - latestAnswers: latestAnswers, - ownVotesAndAnswers: ownVotesAndAnswers, - ); - - final message = pollMessage.copyWith(poll: poll); - updateMessage(message); - }), - ); - } - - void _listenPollClosed() { - _subscriptions.add( - _channel.on(EventType.pollClosed).listen((event) { - final eventPoll = event.poll; - if (eventPoll == null) return; - - final pollMessage = _findPollMessage(eventPoll.id); - if (pollMessage == null) return; - - final oldPoll = pollMessage.poll; - final poll = oldPoll?.copyWith(isClosed: true) ?? eventPoll; - - final message = pollMessage.copyWith(poll: poll); - updateMessage(message); - }), - ); - } - - void _listenPollAnswerCasted() { - _subscriptions.add( - _channel.on(EventType.pollAnswerCasted).listen((event) { - final (eventPoll, eventPollVote) = (event.poll, event.pollVote); - if (eventPoll == null || eventPollVote == null) return; - - final pollMessage = _findPollMessage(eventPoll.id); - if (pollMessage == null) return; - - final oldPoll = pollMessage.poll; - - final latestAnswers = { - for (final ans in oldPoll?.latestAnswers ?? []) ans.id: ans, - eventPollVote.id!: eventPollVote, - }; - - final currentUserId = _client.state.currentUser?.id; - final ownVotesAndAnswers = { - for (final vote in oldPoll?.ownVotesAndAnswers ?? []) vote.id: vote, - if (eventPollVote.userId == currentUserId) eventPollVote.id!: eventPollVote, - }; - - final poll = eventPoll.copyWith( - latestAnswers: [...latestAnswers.values], - ownVotesAndAnswers: [...ownVotesAndAnswers.values], - ); - - final message = pollMessage.copyWith(poll: poll); - updateMessage(message); - }), - ); - } - - void _listenPollVoteCasted() { - _subscriptions.add( - _channel.on(EventType.pollVoteCasted).listen((event) { - final (eventPoll, eventPollVote) = (event.poll, event.pollVote); - if (eventPoll == null || eventPollVote == null) return; - - final pollMessage = _findPollMessage(eventPoll.id); - if (pollMessage == null) return; - - final oldPoll = pollMessage.poll; - - final latestAnswers = oldPoll?.latestAnswers ?? eventPoll.latestAnswers; - final currentUserId = _client.state.currentUser?.id; - final ownVotesAndAnswers = { - for (final vote in oldPoll?.ownVotesAndAnswers ?? []) vote.id: vote, - if (eventPollVote.userId == currentUserId) eventPollVote.id!: eventPollVote, - }; - - final poll = eventPoll.copyWith( - latestAnswers: latestAnswers, - ownVotesAndAnswers: [...ownVotesAndAnswers.values], - ); - - final message = pollMessage.copyWith(poll: poll); - updateMessage(message); - }), - ); - } - - void _listenPollAnswerRemoved() { - _subscriptions.add( - _channel.on(EventType.pollAnswerRemoved).listen((event) { - final (eventPoll, eventPollVote) = (event.poll, event.pollVote); - if (eventPoll == null || eventPollVote == null) return; - - final pollMessage = _findPollMessage(eventPoll.id); - if (pollMessage == null) return; - - final oldPoll = pollMessage.poll; - - final latestAnswers = { - for (final ans in oldPoll?.latestAnswers ?? []) ans.id: ans, - }..remove(eventPollVote.id); - - final ownVotesAndAnswers = { - for (final vote in oldPoll?.ownVotesAndAnswers ?? []) vote.id: vote, - }..remove(eventPollVote.id); - - final poll = eventPoll.copyWith( - latestAnswers: [...latestAnswers.values], - ownVotesAndAnswers: [...ownVotesAndAnswers.values], - ); - - final message = pollMessage.copyWith(poll: poll); - updateMessage(message); - }), - ); - } - - void _listenPollVoteRemoved() { - _subscriptions.add( - _channel.on(EventType.pollVoteRemoved).listen((event) { - final (eventPoll, eventPollVote) = (event.poll, event.pollVote); - if (eventPoll == null || eventPollVote == null) return; - - final pollMessage = _findPollMessage(eventPoll.id); - if (pollMessage == null) return; - - final oldPoll = pollMessage.poll; - - final latestAnswers = oldPoll?.latestAnswers ?? eventPoll.latestAnswers; - final ownVotesAndAnswers = { - for (final vote in oldPoll?.ownVotesAndAnswers ?? []) vote.id: vote, - }..remove(eventPollVote.id); - - final poll = eventPoll.copyWith( - latestAnswers: latestAnswers, - ownVotesAndAnswers: [...ownVotesAndAnswers.values], - ); - - final message = pollMessage.copyWith(poll: poll); - updateMessage(message); - }), - ); - } - - void _listenPollVoteChanged() { - _subscriptions.add( - _channel.on(EventType.pollVoteChanged).listen((event) { - final (eventPoll, eventPollVote) = (event.poll, event.pollVote); - if (eventPoll == null || eventPollVote == null) return; - - final pollMessage = _findPollMessage(eventPoll.id); - if (pollMessage == null) return; - - final oldPoll = pollMessage.poll; - - final latestAnswers = oldPoll?.latestAnswers ?? eventPoll.latestAnswers; - final currentUserId = _client.state.currentUser?.id; - final ownVotesAndAnswers = { - for (final vote in oldPoll?.ownVotesAndAnswers ?? []) vote.id: vote, - if (eventPollVote.userId == currentUserId) eventPollVote.id!: eventPollVote, - }; - - final poll = eventPoll.copyWith( - latestAnswers: latestAnswers, - ownVotesAndAnswers: [...ownVotesAndAnswers.values], - ); - - final message = pollMessage.copyWith(poll: poll); - updateMessage(message); - }), - ); - } - - void _listenDraftUpdated() { - _subscriptions.add( - _channel.on(EventType.draftUpdated).listen((event) { - final draft = event.draft; - if (draft == null) return; - - return updateDraft(draft); - }), - ); - } - - void _listenDraftDeleted() { - _subscriptions.add( - _channel.on(EventType.draftDeleted).listen((event) { - final draft = event.draft; - if (draft == null) return; - - return deleteDraft(draft); - }), - ); - } - - void _listenReminderCreated() { - _subscriptions.add( - _channel.on(EventType.reminderCreated).listen((event) { - final reminder = event.reminder; - if (reminder == null) return; - - updateReminder(reminder); - }), - ); - } - - void _listenReminderUpdated() { - _subscriptions.add( - _channel.on(EventType.reminderUpdated).listen((event) { - final reminder = event.reminder; - if (reminder == null) return; - - updateReminder(reminder); - }), - ); - } - - void _listenReminderDeleted() { - _subscriptions.add( - _channel.on(EventType.reminderDeleted).listen((event) { - final reminder = event.reminder; - if (reminder == null) return; - - deleteReminder(reminder); - }), - ); - } - /// Updates the [reminder] of the message if it exists. void updateReminder(MessageReminder reminder) { final messageId = reminder.messageId; @@ -3113,222 +2598,10 @@ class ChannelClientState { } } - Message? _findLocationMessage(String id) { - final message = messages.firstWhereOrNull((it) { - return it.sharedLocation?.messageId == id; - }); - - if (message != null) return message; - - final threadMessage = threads.values.flattened.firstWhereOrNull((it) { - return it.sharedLocation?.messageId == id; - }); - - return threadMessage; - } - - void _listenLocationShared() { - _subscriptions.add( - _channel.on(EventType.locationShared).listen((event) { - final message = event.message; - if (message == null || message.sharedLocation == null) return; - - return addNewMessage(message); - }), - ); - } - - void _listenLocationUpdated() { - _subscriptions.add( - _channel.on(EventType.locationUpdated).listen((event) { - final location = event.message?.sharedLocation; - if (location == null) return; - - final messageId = location.messageId; - if (messageId == null) return; - - final oldMessage = _findLocationMessage(messageId); - if (oldMessage == null) return; - - final updatedMessage = oldMessage.copyWith(sharedLocation: location); - return updateMessage(updatedMessage); - }), - ); - } - - void _listenLocationExpired() { - _subscriptions.add( - _channel.on(EventType.locationExpired).listen((event) { - final location = event.message?.sharedLocation; - if (location == null) return; - - final messageId = location.messageId; - if (messageId == null) return; - - final oldMessage = _findLocationMessage(messageId); - if (oldMessage == null) return; - - final updatedMessage = oldMessage.copyWith(sharedLocation: location); - return updateMessage(updatedMessage); - }), - ); - } - - void _listenReactionDeleted() { - _subscriptions.add( - _channel.on(EventType.reactionDeleted).listen((event) { - final (eventReaction, eventMessage) = (event.reaction, event.message); - if (eventReaction == null || eventMessage == null) return; - - final messageId = eventMessage.id; - final parentId = eventMessage.parentId; - - for (final message in [...messages, ...?threads[parentId]]) { - if (message.id == messageId) { - final currentUserId = _channel.client.state.currentUser?.id; - - final currentMessage = switch (currentUserId) { - final userId? when userId == eventReaction.userId => message.deleteMyReaction( - reactionType: eventReaction.type, - ), - _ => message, - }; - - return updateMessage( - eventMessage.copyWith( - ownReactions: currentMessage.ownReactions, - ), - ); - } - } - }), - ); - } - - void _listenReactionNew() { - _subscriptions.add( - _channel.on(EventType.reactionNew).listen((event) { - final (eventReaction, eventMessage) = (event.reaction, event.message); - if (eventReaction == null || eventMessage == null) return; - - final messageId = eventMessage.id; - final parentId = eventMessage.parentId; - - for (final message in [...messages, ...?threads[parentId]]) { - if (message.id == messageId) { - final currentUserId = _channel.client.state.currentUser?.id; - - final currentMessage = switch (currentUserId) { - final userId? when userId == eventReaction.userId => message.addMyReaction(eventReaction), - _ => message, - }; - - return updateMessage( - eventMessage.copyWith( - ownReactions: currentMessage.ownReactions, - ), - ); - } - } - }), - ); - } - - void _listenReactionUpdated() { - _subscriptions.add( - _channel.on(EventType.reactionUpdated).listen((event) { - final (eventReaction, eventMessage) = (event.reaction, event.message); - if (eventReaction == null || eventMessage == null) return; - - final messageId = eventMessage.id; - final parentId = eventMessage.parentId; - - for (final message in [...messages, ...?threads[parentId]]) { - if (message.id == messageId) { - final currentUserId = _channel.client.state.currentUser?.id; - - final currentMessage = switch (currentUserId) { - final userId? when userId == eventReaction.userId => - // reaction.updated is only called if enforce_unique is true - message.addMyReaction(eventReaction, enforceUnique: true), - _ => message, - }; - - return updateMessage( - eventMessage.copyWith( - ownReactions: currentMessage.ownReactions, - ), - ); - } - } - }), - ); - } - - void _listenMessageUpdated() { - _subscriptions.add( - _channel.on(EventType.messageUpdated).listen((event) { - final message = event.message; - if (message == null) return; - - return updateMessage(message, upsert: false); - }), - ); - } - - void _listenMessageDeleted() { - _subscriptions.add( - _channel.on(EventType.messageDeleted).listen((event) { - final hardDelete = event.hardDelete ?? false; - - final message = event.message!.copyWith( - // TODO: Remove once deletedForMe is properly enriched on the backend. - deletedForMe: event.deletedForMe, - ); - - // Decrement the locally-tracked unread count for hard-deleted - // messages that would have counted as unread. Soft-deleted messages - // keep their slot. Only applies to channels that track unread counts - // locally (see [Channel.usesLocalUnreadCount]) — server-driven - // channels get corrected counts from server read events instead. - if (hardDelete && _channel.usesLocalUnreadCount && MessageRules.canCountAsUnread(message, _channel)) { - unreadCount = math.max(0, unreadCount - 1); - } - - return deleteMessage(message, hardDelete: hardDelete); - }), - ); - } - - void _listenMessageNew() { - _subscriptions.add( - _channel - .on( - EventType.messageNew, - EventType.notificationMessageNew, - ) - .listen((event) { - final message = event.message; - if (message == null) return; - - addNewMessage(message); - - // Only message.new carries a reliable watcher count; - // notification.message_new targets non-watchers and reports 0. - if (event.watcherCount case final watcherCount? when event.type == EventType.messageNew) { - updateChannelState( - channelState.copyWith(watcherCount: watcherCount), - ); - } - }), - ); - } - /// Adds a new message to the channel state and updates the unread count. void addNewMessage(Message message) { - final isThreadMessage = message.parentId != null; - final isNotShownInChannel = message.showInChannel != true; - final isThreadOnlyMessage = isThreadMessage && isNotShownInChannel; + // A message not shown in the channel is necessarily a thread-only reply. + final isThreadOnlyMessage = !_isShownInChannel(message); // Only add the message if the channel is upToDate or if the message is // a thread-only message. @@ -3447,94 +2720,6 @@ class ChannelClientState { return _deleteMessages([message], hardDelete: hardDelete); } - void _listenReadEvents() { - _subscriptions - ..add( - _channel.on(EventType.messageRead, EventType.notificationMarkRead).listen( - (event) { - // Skip handling the event if delivered for a thread - if (event.thread != null) return; - - final user = event.user; - if (user == null) return; - - final currentRead = userReadOf(userId: user.id); - - final updatedRead = Read( - user: user, - lastRead: event.createdAt, - unreadMessages: 0, // Reset unread count - lastReadMessageId: event.lastReadMessageId, - // Preserve delivery info as it's not part of the read event. - lastDeliveredAt: currentRead?.lastDeliveredAt, - lastDeliveredMessageId: currentRead?.lastDeliveredMessageId, - ); - - updateRead([updatedRead]); - - // If the read event is from the current user, reconcile the - // channel delivery status with the updated read state. - final currentUser = _client.state.currentUser; - if (event.isFromUser(userId: currentUser?.id)) { - _client.channelDeliveryReporter.reconcileDelivery([_channel]); - } - }, - ), - ) - ..add( - _channel.on(EventType.notificationMarkUnread).listen( - (event) { - final user = event.user; - if (user == null) return; - - final currentRead = userReadOf(userId: user.id); - - final updatedRead = Read( - user: user, - lastRead: event.lastReadAt!, - unreadMessages: event.unreadMessages, - lastReadMessageId: event.lastReadMessageId, - // Preserve delivery info as it's not part of the read event. - lastDeliveredAt: currentRead?.lastDeliveredAt, - lastDeliveredMessageId: currentRead?.lastDeliveredMessageId, - ); - - return updateRead([updatedRead]); - }, - ), - ) - ..add( - _channel.on(EventType.messageDelivered).listen( - (event) { - final user = event.user; - if (user == null) return; - - final currentRead = userReadOf(userId: user.id); - final never = DateTime.fromMillisecondsSinceEpoch(0, isUtc: true); - - final updatedRead = Read( - user: user, - lastDeliveredAt: event.lastDeliveredAt, - lastDeliveredMessageId: event.lastDeliveredMessageId, - // Preserve read info as it's not part of the delivery event. - lastRead: currentRead?.lastRead ?? never, - unreadMessages: currentRead?.unreadMessages, - lastReadMessageId: currentRead?.lastReadMessageId, - ); - - updateRead([updatedRead]); - - // If the delivered event is from the current user, reconcile - // the channel delivery with the updated read state. - final currentUser = _client.state.currentUser; - if (event.isFromUser(userId: currentUser?.id)) { - _client.channelDeliveryReporter.reconcileDelivery([_channel]); - } - }, - ), - ); - } - /// Channel message list. List get messages => _channelState.messages ?? []; @@ -3957,36 +3142,14 @@ class ChannelClientState { Map get typingEvents => _typingEventsController.value; final _typingEventsController = BehaviorSubject.seeded({}); - void _listenTypingEvents() { - _subscriptions - ..add( - _channel.on(EventType.typingStart).listen( - (event) { - final user = event.user; - if (user == null) return; - - final currentUser = _client.state.currentUser; - if (event.isFromUser(userId: currentUser?.id)) return; + /// Adds or replaces the typing [event] for the given [user]. + void _upsertTypingEvent(User user, Event event) { + _typingEventsController.safeAdd({...typingEvents, user: event}); + } - final events = {...typingEvents, user: event}; - _typingEventsController.safeAdd(events); - }, - ), - ) - ..add( - _channel.on(EventType.typingStop).listen( - (event) { - final user = event.user; - if (user == null) return; - - final currentUser = _client.state.currentUser; - if (event.isFromUser(userId: currentUser?.id)) return; - - final events = {...typingEvents}..remove(user); - _typingEventsController.safeAdd(events); - }, - ), - ); + /// Removes the typing event for the given [user], if any. + void _removeTypingEvent(User user) { + _typingEventsController.safeAdd({...typingEvents}..remove(user)); } Timer? _staleTypingEventsCleanerTimer; @@ -4081,24 +3244,8 @@ class ChannelClientState { ); } - // Listens to channel push preference update events and updates the state - void _listenChannelPushPreferenceUpdated() { - _subscriptions.add( - _channel.on(EventType.channelPushPreferenceUpdated).listen( - (event) { - final pushPreferences = event.channelPushPreference; - if (pushPreferences == null) return; - - updateChannelState( - channelState.copyWith( - pushPreferences: pushPreferences, - ), - ); - }, - ), - ); - } - + /// Deletes all messages from the user identified by [userId], both from + /// the persistence layer and the channel state. Future _deleteMessagesFromUser({ required String userId, bool hardDelete = false, @@ -4209,14 +3356,8 @@ class ChannelClientState { }) { if (messages.isEmpty) return; - final affectedMessages = messages.map((it) { - // If it's not a thread message, consider it affected. - if (it.parentId == null) return it; - // If it's a thread message shown in channel, consider it affected. - if (it.showInChannel == true) return it; - - return null; // Thread message not shown in channel, ignore it. - }).nonNulls; + // Only messages shown in the channel are affected. + final affectedMessages = messages.where(_isShownInChannel); // If there are no affected messages, return early. if (affectedMessages.isEmpty) return; @@ -4451,14 +3592,8 @@ class ChannelClientState { void _removeChannelMessages(Iterable messages) { if (messages.isEmpty) return; - final affectedMessages = messages.map((it) { - // If it's not a thread message, consider it affected. - if (it.parentId == null) return it; - // If it's a thread message shown in channel, consider it affected. - if (it.showInChannel == true) return it; - - return null; // Thread message not shown in channel, ignore it. - }).nonNulls; + // Only messages shown in the channel are affected. + final affectedMessages = messages.where(_isShownInChannel); // If there are no affected messages, return early. if (affectedMessages.isEmpty) return; @@ -4550,23 +3685,6 @@ class ChannelClientState { return updatedMessages; } - // Listens to user message deleted events and marks messages from that user - // as either soft or hard deleted based on the event data. - void _listenUserMessagesDeleted() { - _subscriptions.add( - _channel.on(EventType.userMessagesDeleted).listen((event) async { - final user = event.user; - if (user == null) return; - - return _deleteMessagesFromUser( - userId: user.id, - hardDelete: event.hardDelete ?? false, - deletedAt: event.createdAt, - ); - }), - ); - } - /// Call this method to dispose this object. void dispose() { _debouncedUpdatePersistenceChannelThreads.cancel(); @@ -4583,6 +3701,14 @@ class ChannelClientState { } } +bool _isShownInChannel(Message message) { + // Non-thread messages are always shown in the channel. + if (message.parentId == null) return true; + + // Thread messages are only shown if explicitly marked. + return message.showInChannel == true; +} + bool _pinIsValid(Message message) { // If the message is deleted, the pin is not valid. if (message.isDeleted) return false; diff --git a/packages/stream_chat/lib/src/client/channel_event_handler.dart b/packages/stream_chat/lib/src/client/channel_event_handler.dart new file mode 100644 index 000000000..6fd46e39d --- /dev/null +++ b/packages/stream_chat/lib/src/client/channel_event_handler.dart @@ -0,0 +1,477 @@ +import 'package:stream_chat/src/client/channel_state_mutations.dart'; +import 'package:stream_chat/stream_chat.dart'; + +/// Routes channel events to the matching [ChannelStateMutations] methods. +/// +/// Drops events with a missing payload, as well as events that do not apply +/// to the channel or the current user. +/// +/// Also performs the side effects an event triggers outside the channel +/// state: member refresh, persisted-message cleanup, and delivery +/// reconciliation. +class ChannelEventHandler { + /// Creates a handler routing events of the given [_channel] to the given + /// [_mutations]. + const ChannelEventHandler({ + required this._channel, + required this._mutations, + }); + + final Channel _channel; + final ChannelStateMutations _mutations; + + StreamChatClient get _client => _channel.client; + + /// Handles the given channel [event]. + /// + /// Routes the event by its type through three dispatch blocks, running + /// [_onChannelMessageCount] between the first and second block and + /// [_onMemberUserUpdated] between the second and third; those two observe + /// every event regardless of its type. + void handleEvent(Event event) { + // Block 1: typing, message, draft, reaction, poll, read, and channel + // events. + switch (event.type) { + // typing events + case EventType.typingStart: + _onTypingStart(event); + case EventType.typingStop: + _onTypingStop(event); + // message events + case EventType.messageNew: + case EventType.notificationMessageNew: + _onMessageNew(event); + case EventType.messageDeleted: + _onMessageDeleted(event); + case EventType.messageUpdated: + _onMessageUpdated(event); + // draft events + case EventType.draftUpdated: + _onDraftUpdated(event); + case EventType.draftDeleted: + _onDraftDeleted(event); + // reaction events + case EventType.reactionNew: + _onReactionNew(event); + case EventType.reactionUpdated: + _onReactionUpdated(event); + case EventType.reactionDeleted: + _onReactionDeleted(event); + // poll events + case EventType.pollCreated: + _onPollCreated(event); + case EventType.pollUpdated: + _onPollUpdated(event); + case EventType.pollClosed: + _onPollClosed(event); + case EventType.pollAnswerCasted: + _onPollAnswerCasted(event); + case EventType.pollVoteCasted: + _onPollVoteCasted(event); + case EventType.pollVoteChanged: + _onPollVoteChanged(event); + case EventType.pollAnswerRemoved: + _onPollAnswerRemoved(event); + case EventType.pollVoteRemoved: + _onPollVoteRemoved(event); + // read events + case EventType.messageRead: + case EventType.notificationMarkRead: + _onMessageRead(event); + case EventType.notificationMarkUnread: + _onNotificationMarkUnread(event); + case EventType.messageDelivered: + _onMessageDelivered(event); + // channel events + case EventType.channelTruncated: + case EventType.notificationChannelTruncated: + _onChannelTruncated(event); + case EventType.channelUpdated: + _onChannelUpdated(event); + } + + // Updates the channel message count on any event carrying one. + _onChannelMessageCount(event); + + // Block 2: member added and removed events. + switch (event.type) { + // member events + case EventType.memberAdded: + _onMemberAdded(event); + case EventType.memberRemoved: + _onMemberRemoved(event); + } + + // Merges the event user into the member list on any event carrying one. + _onMemberUserUpdated(event); + + // Block 3: remaining member, watching, reminder, location, and push + // preference events. + switch (event.type) { + // member events + case EventType.memberUpdated: + _onMemberUpdated(event); + case EventType.userBanned: + _onMemberBanned(event); + case EventType.userUnbanned: + _onMemberUnbanned(event); + case EventType.userMessagesDeleted: + _onUserMessagesDeleted(event); + // user watching events + case EventType.userWatchingStart: + _onUserStartWatching(event); + case EventType.userWatchingStop: + _onUserStopWatching(event); + // reminder events + case EventType.reminderCreated: + _onReminderCreated(event); + case EventType.reminderUpdated: + _onReminderUpdated(event); + case EventType.reminderDeleted: + _onReminderDeleted(event); + // location events + case EventType.locationShared: + _onLocationShared(event); + case EventType.locationUpdated: + _onLocationUpdated(event); + case EventType.locationExpired: + _onLocationExpired(event); + // channel push preference events + case EventType.channelPushPreferenceUpdated: + _onChannelPushPreferenceUpdated(event); + } + } + + void _onTypingStart(Event event) { + final user = event.user; + if (user == null) return; + + final currentUser = _client.state.currentUser; + if (event.isFromUser(userId: currentUser?.id)) return; + + _mutations.onTypingStart(user, event); + } + + void _onTypingStop(Event event) { + final user = event.user; + if (user == null) return; + + final currentUser = _client.state.currentUser; + if (event.isFromUser(userId: currentUser?.id)) return; + + _mutations.onTypingStop(user); + } + + void _onMessageNew(Event event) { + final message = event.message; + if (message == null) return; + + // Only message.new carries a reliable watcher count; + // notification.message_new targets non-watchers and reports 0. + final watcherCount = switch (event.type) { + EventType.messageNew => event.watcherCount, + _ => null, + }; + + _mutations.onMessageNew(message, watcherCount: watcherCount); + } + + void _onMessageDeleted(Event event) { + final hardDelete = event.hardDelete ?? false; + + final message = event.message!.copyWith( + // TODO: Remove once deletedForMe is properly enriched on the backend. + deletedForMe: event.deletedForMe, + ); + + return _mutations.onMessageDeleted(message, hardDelete: hardDelete); + } + + void _onMessageUpdated(Event event) { + final message = event.message; + if (message == null) return; + + return _mutations.onMessageUpdated(message); + } + + void _onDraftUpdated(Event event) { + final draft = event.draft; + if (draft == null) return; + + return _mutations.onDraftUpdated(draft); + } + + void _onDraftDeleted(Event event) { + final draft = event.draft; + if (draft == null) return; + + return _mutations.onDraftDeleted(draft); + } + + void _onReactionNew(Event event) { + final (eventReaction, eventMessage) = (event.reaction, event.message); + if (eventReaction == null || eventMessage == null) return; + + return _mutations.onReactionNew(eventMessage, eventReaction); + } + + void _onReactionUpdated(Event event) { + final (eventReaction, eventMessage) = (event.reaction, event.message); + if (eventReaction == null || eventMessage == null) return; + + return _mutations.onReactionUpdated(eventMessage, eventReaction); + } + + void _onReactionDeleted(Event event) { + final (eventReaction, eventMessage) = (event.reaction, event.message); + if (eventReaction == null || eventMessage == null) return; + + return _mutations.onReactionDeleted(eventMessage, eventReaction); + } + + void _onPollCreated(Event event) { + final message = event.message; + if (message == null || message.poll == null) return; + + return _mutations.onPollCreated(message); + } + + void _onPollUpdated(Event event) { + final eventPoll = event.poll; + if (eventPoll == null) return; + + return _mutations.onPollUpdated(eventPoll); + } + + void _onPollClosed(Event event) { + final eventPoll = event.poll; + if (eventPoll == null) return; + + return _mutations.onPollClosed(eventPoll); + } + + void _onPollAnswerCasted(Event event) { + final (eventPoll, eventPollVote) = (event.poll, event.pollVote); + if (eventPoll == null || eventPollVote == null) return; + + return _mutations.onPollAnswerCasted(eventPoll, eventPollVote); + } + + void _onPollVoteCasted(Event event) { + final (eventPoll, eventPollVote) = (event.poll, event.pollVote); + if (eventPoll == null || eventPollVote == null) return; + + return _mutations.onPollVoteCasted(eventPoll, eventPollVote); + } + + void _onPollVoteChanged(Event event) { + final (eventPoll, eventPollVote) = (event.poll, event.pollVote); + if (eventPoll == null || eventPollVote == null) return; + + return _mutations.onPollVoteChanged(eventPoll, eventPollVote); + } + + void _onPollAnswerRemoved(Event event) { + final (eventPoll, eventPollVote) = (event.poll, event.pollVote); + if (eventPoll == null || eventPollVote == null) return; + + return _mutations.onPollAnswerRemoved(eventPoll, eventPollVote); + } + + void _onPollVoteRemoved(Event event) { + final (eventPoll, eventPollVote) = (event.poll, event.pollVote); + if (eventPoll == null || eventPollVote == null) return; + + return _mutations.onPollVoteRemoved(eventPoll, eventPollVote); + } + + void _onMessageRead(Event event) { + // Skip handling the event if delivered for a thread + if (event.thread != null) return; + + final user = event.user; + if (user == null) return; + + _mutations.onMessageRead( + user, + lastRead: event.createdAt, + lastReadMessageId: event.lastReadMessageId, + ); + + // If the read event is from the current user, reconcile the + // channel delivery status with the updated read state. + final currentUser = _client.state.currentUser; + if (event.isFromUser(userId: currentUser?.id)) { + _client.channelDeliveryReporter.reconcileDelivery([_channel]); + } + } + + void _onNotificationMarkUnread(Event event) { + final user = event.user; + if (user == null) return; + + return _mutations.onNotificationMarkUnread( + user, + lastRead: event.lastReadAt!, + unreadMessages: event.unreadMessages, + lastReadMessageId: event.lastReadMessageId, + ); + } + + void _onMessageDelivered(Event event) { + final user = event.user; + if (user == null) return; + + _mutations.onMessageDelivered( + user, + lastDeliveredAt: event.lastDeliveredAt, + lastDeliveredMessageId: event.lastDeliveredMessageId, + ); + + // If the delivered event is from the current user, reconcile + // the channel delivery with the updated read state. + final currentUser = _client.state.currentUser; + if (event.isFromUser(userId: currentUser?.id)) { + _client.channelDeliveryReporter.reconcileDelivery([_channel]); + } + } + + Future _onChannelTruncated(Event event) async { + final channel = event.channel!; + await _client.chatPersistenceClient?.deleteMessageByCid(channel.cid); + + _mutations.onChannelTruncated(message: event.message); + } + + void _onChannelUpdated(Event event) { + final channel = event.channel!; + + return _mutations.onChannelUpdated(channel); + } + + void _onChannelMessageCount(Event event) { + final messageCount = event.channelMessageCount; + if (messageCount == null) return; + + return _mutations.onChannelMessageCount(messageCount); + } + + void _onMemberAdded(Event event) { + final member = event.member!; + + return _mutations.onMemberAdded(member); + } + + void _onMemberRemoved(Event event) { + final user = event.user!; + + return _mutations.onMemberRemoved(user); + } + + void _onMemberUserUpdated(Event event) { + final user = event.user; + if (user == null) return; + + return _mutations.onMemberUserUpdated(user); + } + + void _onMemberUpdated(Event event) { + final member = event.member!; + + return _mutations.onMemberUpdated(member); + } + + Future _onMemberBanned(Event event) async { + // Filters channel ban from app ban. + if (event.cid == null) return; + + final user = event.user!; + final member = await _channel.queryMembers(filter: Filter.equal('id', user.id)).then((it) => it.members.first); + + _mutations.onMemberBanned(member); + } + + Future _onMemberUnbanned(Event event) async { + // Filters channel ban from app ban. + if (event.cid == null) return; + + final user = event.user!; + final member = await _channel.queryMembers(filter: Filter.equal('id', user.id)).then((it) => it.members.first); + + _mutations.onMemberUnbanned(member); + } + + Future _onUserMessagesDeleted(Event event) async { + final user = event.user; + if (user == null) return; + + return _mutations.onUserMessagesDeleted( + userId: user.id, + hardDelete: event.hardDelete ?? false, + deletedAt: event.createdAt, + ); + } + + void _onUserStartWatching(Event event) { + final watcher = event.user; + if (watcher == null) return; + + return _mutations.onUserStartWatching(watcher, watcherCount: event.watcherCount); + } + + void _onUserStopWatching(Event event) { + final watcher = event.user; + if (watcher == null) return; + + return _mutations.onUserStopWatching(watcher, watcherCount: event.watcherCount); + } + + void _onReminderCreated(Event event) { + final reminder = event.reminder; + if (reminder == null) return; + + return _mutations.onReminderCreated(reminder); + } + + void _onReminderUpdated(Event event) { + final reminder = event.reminder; + if (reminder == null) return; + + return _mutations.onReminderUpdated(reminder); + } + + void _onReminderDeleted(Event event) { + final reminder = event.reminder; + if (reminder == null) return; + + return _mutations.onReminderDeleted(reminder); + } + + void _onLocationShared(Event event) { + final message = event.message; + if (message == null || message.sharedLocation == null) return; + + return _mutations.onLocationShared(message); + } + + void _onLocationUpdated(Event event) { + final location = event.message?.sharedLocation; + if (location == null) return; + + return _mutations.onLocationUpdated(location); + } + + void _onLocationExpired(Event event) { + final location = event.message?.sharedLocation; + if (location == null) return; + + return _mutations.onLocationExpired(location); + } + + void _onChannelPushPreferenceUpdated(Event event) { + final pushPreferences = event.channelPushPreference; + if (pushPreferences == null) return; + + return _mutations.onChannelPushPreferenceUpdated(pushPreferences); + } +} diff --git a/packages/stream_chat/lib/src/client/channel_state_mutations.dart b/packages/stream_chat/lib/src/client/channel_state_mutations.dart new file mode 100644 index 000000000..c132765c6 --- /dev/null +++ b/packages/stream_chat/lib/src/client/channel_state_mutations.dart @@ -0,0 +1,581 @@ +import 'dart:math' as math; + +import 'package:collection/collection.dart'; +import 'package:stream_chat/stream_chat.dart'; + +/// Applies channel event payloads as [ChannelClientState] mutations. +/// +/// Each method computes the new state from the current one and the given +/// payload, and writes it to the state. +class ChannelStateMutations { + /// Creates mutations writing to the given [_state] of the given [_channel]. + const ChannelStateMutations({ + required this._channel, + required this._state, + required this._upsertTypingEvent, + required this._removeTypingEvent, + required this._removeWatcher, + required this._updateMember, + required this._deleteMessagesFromUser, + }); + + final Channel _channel; + final ChannelClientState _state; + + final void Function(User user, Event event) _upsertTypingEvent; + final void Function(User user) _removeTypingEvent; + final void Function(User watcher, {int? watcherCount}) _removeWatcher; + final void Function(Member member) _updateMember; + final Future Function({ + required String userId, + bool hardDelete, + DateTime? deletedAt, + }) + _deleteMessagesFromUser; + + StreamChatClient get _client => _channel.client; + + /// Records the typing [event] for the given [user]. + void onTypingStart(User user, Event event) => _upsertTypingEvent(user, event); + + /// Clears the typing state for the given [user]. + void onTypingStop(User user) => _removeTypingEvent(user); + + /// Adds the new [message], updating the watcher count when provided. + void onMessageNew(Message message, {int? watcherCount}) { + _state.addNewMessage(message); + + if (watcherCount != null) { + _state.updateChannelState( + _state.channelState.copyWith(watcherCount: watcherCount), + ); + } + } + + /// Marks the [message] as deleted. + void onMessageDeleted(Message message, {bool hardDelete = false}) { + // Decrement the locally-tracked unread count for hard-deleted + // messages that would have counted as unread. Soft-deleted messages + // keep their slot. Only applies to channels that track unread counts + // locally (see [Channel.usesLocalUnreadCount]) — server-driven + // channels get corrected counts from server read events instead. + if (hardDelete && _channel.usesLocalUnreadCount && MessageRules.canCountAsUnread(message, _channel)) { + _state.unreadCount = math.max(0, _state.unreadCount - 1); + } + + return _state.deleteMessage(message, hardDelete: hardDelete); + } + + /// Applies the updated [message] without inserting it when absent. + void onMessageUpdated(Message message) { + return _state.updateMessage(message, upsert: false); + } + + /// Applies the updated [draft]. + void onDraftUpdated(Draft draft) => _state.updateDraft(draft); + + /// Removes the deleted [draft]. + void onDraftDeleted(Draft draft) => _state.deleteDraft(draft); + + /// Applies the [eventMessage] carrying the new [eventReaction], preserving + /// the current user's own reactions. + void onReactionNew(Message eventMessage, Reaction eventReaction) { + final messageId = eventMessage.id; + final parentId = eventMessage.parentId; + + for (final message in [..._state.messages, ...?_state.threads[parentId]]) { + if (message.id == messageId) { + final currentUserId = _client.state.currentUser?.id; + + final currentMessage = switch (currentUserId) { + final userId? when userId == eventReaction.userId => message.addMyReaction(eventReaction), + _ => message, + }; + + return _state.updateMessage( + eventMessage.copyWith( + ownReactions: currentMessage.ownReactions, + ), + ); + } + } + } + + /// Applies the [eventMessage] carrying the updated [eventReaction], + /// preserving the current user's own reactions. + void onReactionUpdated(Message eventMessage, Reaction eventReaction) { + final messageId = eventMessage.id; + final parentId = eventMessage.parentId; + + for (final message in [..._state.messages, ...?_state.threads[parentId]]) { + if (message.id == messageId) { + final currentUserId = _client.state.currentUser?.id; + + final currentMessage = switch (currentUserId) { + final userId? when userId == eventReaction.userId => + // reaction.updated is only called if enforce_unique is true + message.addMyReaction(eventReaction, enforceUnique: true), + _ => message, + }; + + return _state.updateMessage( + eventMessage.copyWith( + ownReactions: currentMessage.ownReactions, + ), + ); + } + } + } + + /// Applies the [eventMessage] carrying the deleted [eventReaction], + /// preserving the current user's own reactions. + void onReactionDeleted(Message eventMessage, Reaction eventReaction) { + final messageId = eventMessage.id; + final parentId = eventMessage.parentId; + + for (final message in [..._state.messages, ...?_state.threads[parentId]]) { + if (message.id == messageId) { + final currentUserId = _client.state.currentUser?.id; + + final currentMessage = switch (currentUserId) { + final userId? when userId == eventReaction.userId => message.deleteMyReaction( + reactionType: eventReaction.type, + ), + _ => message, + }; + + return _state.updateMessage( + eventMessage.copyWith( + ownReactions: currentMessage.ownReactions, + ), + ); + } + } + } + + Message? _findPollMessage(String pollId) { + final message = _state.messages.firstWhereOrNull((it) => it.pollId == pollId); + if (message != null) return message; + + final threadMessage = _state.threads.values.flattened.firstWhereOrNull((it) { + return it.pollId == pollId; + }); + + return threadMessage; + } + + /// Adds the [message] carrying a newly created poll. + void onPollCreated(Message message) => _state.addNewMessage(message); + + /// Applies the updated [eventPoll] to the message carrying it, preserving + /// the known answers and own votes. + void onPollUpdated(Poll eventPoll) { + final pollMessage = _findPollMessage(eventPoll.id); + if (pollMessage == null) return; + + final oldPoll = pollMessage.poll; + + final latestAnswers = oldPoll?.latestAnswers ?? eventPoll.latestAnswers; + final ownVotesAndAnswers = oldPoll?.ownVotesAndAnswers ?? eventPoll.ownVotesAndAnswers; + + final poll = eventPoll.copyWith( + latestAnswers: latestAnswers, + ownVotesAndAnswers: ownVotesAndAnswers, + ); + + final message = pollMessage.copyWith(poll: poll); + _state.updateMessage(message); + } + + /// Marks the poll matching [eventPoll] as closed. + void onPollClosed(Poll eventPoll) { + final pollMessage = _findPollMessage(eventPoll.id); + if (pollMessage == null) return; + + final oldPoll = pollMessage.poll; + final poll = oldPoll?.copyWith(isClosed: true) ?? eventPoll; + + final message = pollMessage.copyWith(poll: poll); + _state.updateMessage(message); + } + + /// Applies the casted answer [eventPollVote] to the poll matching + /// [eventPoll]. + void onPollAnswerCasted(Poll eventPoll, PollVote eventPollVote) { + final pollMessage = _findPollMessage(eventPoll.id); + if (pollMessage == null) return; + + final oldPoll = pollMessage.poll; + + final latestAnswers = { + for (final ans in oldPoll?.latestAnswers ?? []) ans.id: ans, + eventPollVote.id!: eventPollVote, + }; + + final currentUserId = _client.state.currentUser?.id; + final ownVotesAndAnswers = { + for (final vote in oldPoll?.ownVotesAndAnswers ?? []) vote.id: vote, + if (eventPollVote.userId == currentUserId) eventPollVote.id!: eventPollVote, + }; + + final poll = eventPoll.copyWith( + latestAnswers: [...latestAnswers.values], + ownVotesAndAnswers: [...ownVotesAndAnswers.values], + ); + + final message = pollMessage.copyWith(poll: poll); + _state.updateMessage(message); + } + + /// Applies the casted [eventPollVote] to the poll matching [eventPoll]. + void onPollVoteCasted(Poll eventPoll, PollVote eventPollVote) { + final pollMessage = _findPollMessage(eventPoll.id); + if (pollMessage == null) return; + + final oldPoll = pollMessage.poll; + + final latestAnswers = oldPoll?.latestAnswers ?? eventPoll.latestAnswers; + final currentUserId = _client.state.currentUser?.id; + final ownVotesAndAnswers = { + for (final vote in oldPoll?.ownVotesAndAnswers ?? []) vote.id: vote, + if (eventPollVote.userId == currentUserId) eventPollVote.id!: eventPollVote, + }; + + final poll = eventPoll.copyWith( + latestAnswers: latestAnswers, + ownVotesAndAnswers: [...ownVotesAndAnswers.values], + ); + + final message = pollMessage.copyWith(poll: poll); + _state.updateMessage(message); + } + + /// Applies the changed [eventPollVote] to the poll matching [eventPoll]. + void onPollVoteChanged(Poll eventPoll, PollVote eventPollVote) { + final pollMessage = _findPollMessage(eventPoll.id); + if (pollMessage == null) return; + + final oldPoll = pollMessage.poll; + + final latestAnswers = oldPoll?.latestAnswers ?? eventPoll.latestAnswers; + final currentUserId = _client.state.currentUser?.id; + final ownVotesAndAnswers = { + for (final vote in oldPoll?.ownVotesAndAnswers ?? []) vote.id: vote, + if (eventPollVote.userId == currentUserId) eventPollVote.id!: eventPollVote, + }; + + final poll = eventPoll.copyWith( + latestAnswers: latestAnswers, + ownVotesAndAnswers: [...ownVotesAndAnswers.values], + ); + + final message = pollMessage.copyWith(poll: poll); + _state.updateMessage(message); + } + + /// Removes the answer [eventPollVote] from the poll matching [eventPoll]. + void onPollAnswerRemoved(Poll eventPoll, PollVote eventPollVote) { + final pollMessage = _findPollMessage(eventPoll.id); + if (pollMessage == null) return; + + final oldPoll = pollMessage.poll; + + final latestAnswers = { + for (final ans in oldPoll?.latestAnswers ?? []) ans.id: ans, + }..remove(eventPollVote.id); + + final ownVotesAndAnswers = { + for (final vote in oldPoll?.ownVotesAndAnswers ?? []) vote.id: vote, + }..remove(eventPollVote.id); + + final poll = eventPoll.copyWith( + latestAnswers: [...latestAnswers.values], + ownVotesAndAnswers: [...ownVotesAndAnswers.values], + ); + + final message = pollMessage.copyWith(poll: poll); + _state.updateMessage(message); + } + + /// Removes the [eventPollVote] from the poll matching [eventPoll]. + void onPollVoteRemoved(Poll eventPoll, PollVote eventPollVote) { + final pollMessage = _findPollMessage(eventPoll.id); + if (pollMessage == null) return; + + final oldPoll = pollMessage.poll; + + final latestAnswers = oldPoll?.latestAnswers ?? eventPoll.latestAnswers; + final ownVotesAndAnswers = { + for (final vote in oldPoll?.ownVotesAndAnswers ?? []) vote.id: vote, + }..remove(eventPollVote.id); + + final poll = eventPoll.copyWith( + latestAnswers: latestAnswers, + ownVotesAndAnswers: [...ownVotesAndAnswers.values], + ); + + final message = pollMessage.copyWith(poll: poll); + _state.updateMessage(message); + } + + /// Applies a read for the [user], resetting its unread count and + /// preserving its delivery info. + void onMessageRead( + User user, { + required DateTime lastRead, + String? lastReadMessageId, + }) { + final currentRead = _state.userReadOf(userId: user.id); + + final updatedRead = Read( + user: user, + lastRead: lastRead, + unreadMessages: 0, // Reset unread count + lastReadMessageId: lastReadMessageId, + // Preserve delivery info as it's not part of the read event. + lastDeliveredAt: currentRead?.lastDeliveredAt, + lastDeliveredMessageId: currentRead?.lastDeliveredMessageId, + ); + + _state.updateRead([updatedRead]); + } + + /// Applies an unread mark for the [user], preserving its delivery info. + void onNotificationMarkUnread( + User user, { + required DateTime lastRead, + int? unreadMessages, + String? lastReadMessageId, + }) { + final currentRead = _state.userReadOf(userId: user.id); + + final updatedRead = Read( + user: user, + lastRead: lastRead, + unreadMessages: unreadMessages, + lastReadMessageId: lastReadMessageId, + // Preserve delivery info as it's not part of the read event. + lastDeliveredAt: currentRead?.lastDeliveredAt, + lastDeliveredMessageId: currentRead?.lastDeliveredMessageId, + ); + + return _state.updateRead([updatedRead]); + } + + /// Applies a delivery for the [user], preserving its read info. + void onMessageDelivered( + User user, { + DateTime? lastDeliveredAt, + String? lastDeliveredMessageId, + }) { + final currentRead = _state.userReadOf(userId: user.id); + final never = DateTime.fromMillisecondsSinceEpoch(0, isUtc: true); + + final updatedRead = Read( + user: user, + lastDeliveredAt: lastDeliveredAt, + lastDeliveredMessageId: lastDeliveredMessageId, + // Preserve read info as it's not part of the delivery event. + lastRead: currentRead?.lastRead ?? never, + unreadMessages: currentRead?.unreadMessages, + lastReadMessageId: currentRead?.lastReadMessageId, + ); + + _state.updateRead([updatedRead]); + } + + /// Clears the channel messages, applying the truncation system [message] + /// when provided. + void onChannelTruncated({Message? message}) { + _state.truncate(); + if (message != null) { + _state.updateMessage(message); + } + } + + /// Merges the updated [channel] model and replaces the member list. + void onChannelUpdated(ChannelModel channel) { + _state.updateChannelState( + _state.channelState.copyWith( + channel: _state.channelState.channel?.merge(channel), + members: channel.members, + ), + ); + } + + /// Updates the channel [messageCount]. + void onChannelMessageCount(int messageCount) { + _state.updateChannelState( + _state.channelState.copyWith( + channel: _state.channelState.channel?.copyWith( + messageCount: messageCount, + ), + ), + ); + } + + /// Appends the added [member] to the member list. + void onMemberAdded(Member member) { + final existingMembers = _state.channelState.members ?? []; + + _state.updateChannelState( + _state.channelState.copyWith( + members: [...existingMembers, member], + ), + ); + } + + /// Removes the [user]'s membership and read state. + void onMemberRemoved(User user) { + final existingRead = _state.channelState.read ?? []; + final existingMembers = _state.channelState.members ?? []; + + _state.updateChannelState( + _state.channelState.copyWith( + read: [...existingRead.where((r) => r.user.id != user.id)], + members: [...existingMembers.where((m) => m.userId != user.id)], + ), + ); + } + + /// Merges the updated [user] into the matching member and membership. + /// + /// Does nothing if the user is not an existing member of the channel. + void onMemberUserUpdated(User user) { + final existingMembers = [...?_state.channelState.members]; + final existingMembership = _state.channelState.membership; + + // Return if the user is not a existing member of the channel. + if (!existingMembers.any((m) => m.userId == user.id)) return; + + Member? maybeUpdateMemberUser(Member? existingMember) { + if (existingMember == null) return null; + if (existingMember.userId == user.id) { + return existingMember.copyWith(user: user); + } + return existingMember; + } + + _state.updateChannelState( + _state.channelState.copyWith( + membership: maybeUpdateMemberUser(existingMembership), + members: [...existingMembers.map(maybeUpdateMemberUser).nonNulls], + ), + ); + } + + /// Replaces the matching [member] and membership. + void onMemberUpdated(Member member) { + final existingMembers = _state.channelState.members ?? []; + final existingMembership = _state.channelState.membership; + + Member? maybeUpdateMember(Member? existingMember) { + if (existingMember == null) return null; + if (existingMember.userId == member.userId) return member; + return existingMember; + } + + _state.updateChannelState( + _state.channelState.copyWith( + membership: maybeUpdateMember(existingMembership), + members: [...existingMembers.map(maybeUpdateMember).nonNulls], + ), + ); + } + + /// Replaces the member matching the banned [member]'s user id, if any. + void onMemberBanned(Member member) => _updateMember(member); + + /// Replaces the member matching the unbanned [member]'s user id, if any. + void onMemberUnbanned(Member member) => _updateMember(member); + + /// Marks all messages from the user identified by [userId] as deleted. + Future onUserMessagesDeleted({ + required String userId, + bool hardDelete = false, + DateTime? deletedAt, + }) { + return _deleteMessagesFromUser( + userId: userId, + hardDelete: hardDelete, + deletedAt: deletedAt, + ); + } + + /// Adds the [watcher], updating the watcher count when provided. + void onUserStartWatching(User watcher, {int? watcherCount}) { + final existingWatchers = _state.channelState.watchers; + + _state.updateChannelState( + _state.channelState.copyWith( + watchers: [ + watcher, + ...?existingWatchers?.where((user) => user.id != watcher.id), + ], + watcherCount: watcherCount, + ), + ); + } + + /// Removes the [watcher], updating the watcher count when provided. + void onUserStopWatching(User watcher, {int? watcherCount}) { + return _removeWatcher(watcher, watcherCount: watcherCount); + } + + /// Applies the created [reminder]. + void onReminderCreated(MessageReminder reminder) => _state.updateReminder(reminder); + + /// Applies the updated [reminder]. + void onReminderUpdated(MessageReminder reminder) => _state.updateReminder(reminder); + + /// Removes the deleted [reminder]. + void onReminderDeleted(MessageReminder reminder) => _state.deleteReminder(reminder); + + Message? _findLocationMessage(String id) { + final message = _state.messages.firstWhereOrNull((it) { + return it.sharedLocation?.messageId == id; + }); + + if (message != null) return message; + + return _state.threads.values.flattened.firstWhereOrNull((it) { + return it.sharedLocation?.messageId == id; + }); + } + + /// Adds the [message] sharing a live location. + void onLocationShared(Message message) => _state.addNewMessage(message); + + /// Applies the updated [location] to the message sharing it. + void onLocationUpdated(Location location) { + final messageId = location.messageId; + if (messageId == null) return; + + final oldMessage = _findLocationMessage(messageId); + if (oldMessage == null) return; + + final updatedMessage = oldMessage.copyWith(sharedLocation: location); + return _state.updateMessage(updatedMessage); + } + + /// Applies the expired [location] to the message sharing it. + void onLocationExpired(Location location) { + final messageId = location.messageId; + if (messageId == null) return; + + final oldMessage = _findLocationMessage(messageId); + if (oldMessage == null) return; + + final updatedMessage = oldMessage.copyWith(sharedLocation: location); + return _state.updateMessage(updatedMessage); + } + + /// Applies the updated channel [pushPreferences]. + void onChannelPushPreferenceUpdated(ChannelPushPreference pushPreferences) { + _state.updateChannelState( + _state.channelState.copyWith( + pushPreferences: pushPreferences, + ), + ); + } +} diff --git a/packages/stream_chat/test/src/client/channel_event_handler_test.dart b/packages/stream_chat/test/src/client/channel_event_handler_test.dart new file mode 100644 index 000000000..ae3cc9063 --- /dev/null +++ b/packages/stream_chat/test/src/client/channel_event_handler_test.dart @@ -0,0 +1,794 @@ +// ignore_for_file: cascade_invocations + +import 'package:mocktail/mocktail.dart'; +import 'package:stream_chat/src/client/channel_event_handler.dart'; +import 'package:stream_chat/src/client/channel_state_mutations.dart'; +import 'package:stream_chat/stream_chat.dart'; +import 'package:test/test.dart'; + +import '../fakes.dart'; +import '../mocks.dart'; + +class MockChannel extends Mock implements Channel {} + +class MockChannelStateMutations extends Mock implements ChannelStateMutations {} + +class FakeDraft extends Fake implements Draft {} + +class FakeReaction extends Fake implements Reaction {} + +class FakePoll extends Fake implements Poll {} + +class FakeMember extends Fake implements Member {} + +class FakeMessageReminder extends Fake implements MessageReminder {} + +class FakeLocation extends Fake implements Location {} + +class FakeChannelPushPreference extends Fake implements ChannelPushPreference {} + +class FakeQueryMembersResponse extends Fake implements QueryMembersResponse { + FakeQueryMembersResponse(this.members); + + @override + final List members; +} + +void main() { + late MockChannel channel; + late MockStreamChatClient client; + late MockChannelStateMutations mutations; + late ChannelEventHandler handler; + + // Matches the default current user of [FakeClientState]. + const currentUserId = 'test-user-id'; + final otherUser = User(id: 'other-user'); + + setUpAll(() { + registerFallbackValue(FakeMessage()); + registerFallbackValue(FakeUser()); + registerFallbackValue(FakeEvent()); + registerFallbackValue(FakeDraft()); + registerFallbackValue(FakeReaction()); + registerFallbackValue(FakePoll()); + registerFallbackValue(FakePollVote()); + registerFallbackValue(FakeMember()); + registerFallbackValue(FakeMessageReminder()); + registerFallbackValue(FakeLocation()); + registerFallbackValue(FakeChannelPushPreference()); + registerFallbackValue(Filter.equal('id', '')); + registerFallbackValue(''); + registerFallbackValue(0); + registerFallbackValue(false); + registerFallbackValue(DateTime(0)); + registerFallbackValue([]); + }); + + setUp(() { + channel = MockChannel(); + client = MockStreamChatClient(); + mutations = MockChannelStateMutations(); + + when(() => channel.client).thenReturn(client); + when(() => client.state).thenReturn(FakeClientState()); + when(() => client.channelDeliveryReporter.reconcileDelivery(any())).thenAnswer((_) async {}); + + when( + () => mutations.onUserMessagesDeleted( + userId: any(named: 'userId'), + hardDelete: any(named: 'hardDelete'), + deletedAt: any(named: 'deletedAt'), + ), + ).thenAnswer((_) async {}); + + handler = ChannelEventHandler(channel: channel, mutations: mutations); + }); + + group('dispatch', () { + test('does nothing for an unknown event without payloads', () { + handler.handleEvent(Event(type: 'unknown.event')); + + verifyZeroInteractions(mutations); + }); + + test('runs the unfiltered handlers around the typed blocks in order', () { + final member = Member(userId: otherUser.id); + final event = Event( + type: EventType.memberAdded, + member: member, + user: otherUser, + channelMessageCount: 5, + ); + + handler.handleEvent(event); + + verifyInOrder([ + () => mutations.onChannelMessageCount(5), + () => mutations.onMemberAdded(member), + () => mutations.onMemberUserUpdated(otherUser), + ]); + }); + }); + + group('typing events', () { + test('typing.start delegates the typing user and event', () { + final event = Event(type: EventType.typingStart, user: otherUser); + + handler.handleEvent(event); + + verify(() => mutations.onTypingStart(otherUser, event)).called(1); + }); + + test('typing.start ignores events without a user', () { + handler.handleEvent(Event(type: EventType.typingStart)); + + verifyNever(() => mutations.onTypingStart(any(), any())); + }); + + test('typing.start ignores events from the current user', () { + final event = Event( + type: EventType.typingStart, + user: User(id: currentUserId), + ); + + handler.handleEvent(event); + + verifyNever(() => mutations.onTypingStart(any(), any())); + }); + + test('typing.stop delegates the typing user', () { + final event = Event(type: EventType.typingStop, user: otherUser); + + handler.handleEvent(event); + + verify(() => mutations.onTypingStop(otherUser)).called(1); + }); + + test('typing.stop ignores events from the current user', () { + final event = Event( + type: EventType.typingStop, + user: User(id: currentUserId), + ); + + handler.handleEvent(event); + + verifyNever(() => mutations.onTypingStop(any())); + }); + }); + + group('message events', () { + final message = Message(id: 'message-id', user: otherUser); + + test('message.new delegates the message with its watcher count', () { + final event = Event( + type: EventType.messageNew, + message: message, + watcherCount: 7, + ); + + handler.handleEvent(event); + + verify(() => mutations.onMessageNew(message, watcherCount: 7)).called(1); + }); + + test('notification.message_new never forwards the watcher count', () { + final event = Event( + type: EventType.notificationMessageNew, + message: message, + watcherCount: 7, + ); + + handler.handleEvent(event); + + verify(() => mutations.onMessageNew(message, watcherCount: null)).called(1); + }); + + test('message.new ignores events without a message', () { + handler.handleEvent(Event(type: EventType.messageNew)); + + verifyNever(() => mutations.onMessageNew(any(), watcherCount: any(named: 'watcherCount'))); + }); + + test('message.deleted delegates the message enriched with deletedForMe', () { + final event = Event( + type: EventType.messageDeleted, + message: message, + hardDelete: true, + deletedForMe: true, + ); + + handler.handleEvent(event); + + final expected = message.copyWith(deletedForMe: true); + verify(() => mutations.onMessageDeleted(expected, hardDelete: true)).called(1); + }); + + test('message.deleted defaults to a soft delete', () { + final event = Event(type: EventType.messageDeleted, message: message); + + handler.handleEvent(event); + + verify(() => mutations.onMessageDeleted(any(), hardDelete: false)).called(1); + }); + + test('message.updated delegates the message', () { + final event = Event(type: EventType.messageUpdated, message: message); + + handler.handleEvent(event); + + verify(() => mutations.onMessageUpdated(message)).called(1); + }); + + test('message.updated ignores events without a message', () { + handler.handleEvent(Event(type: EventType.messageUpdated)); + + verifyNever(() => mutations.onMessageUpdated(any())); + }); + }); + + group('draft events', () { + final draft = Draft( + channelCid: 'messaging:test', + createdAt: DateTime.now(), + message: DraftMessage(text: 'draft'), + ); + + test('draft.updated delegates the draft', () { + handler.handleEvent(Event(type: EventType.draftUpdated, draft: draft)); + + verify(() => mutations.onDraftUpdated(draft)).called(1); + }); + + test('draft.deleted delegates the draft', () { + handler.handleEvent(Event(type: EventType.draftDeleted, draft: draft)); + + verify(() => mutations.onDraftDeleted(draft)).called(1); + }); + + test('draft events ignore events without a draft', () { + handler.handleEvent(Event(type: EventType.draftUpdated)); + handler.handleEvent(Event(type: EventType.draftDeleted)); + + verifyNever(() => mutations.onDraftUpdated(any())); + verifyNever(() => mutations.onDraftDeleted(any())); + }); + }); + + group('reaction events', () { + final message = Message(id: 'message-id', user: otherUser); + final reaction = Reaction( + type: 'like', + messageId: 'message-id', + userId: otherUser.id, + createdAt: DateTime.now(), + ); + + test('reaction.new delegates the message and reaction', () { + final event = Event( + type: EventType.reactionNew, + message: message, + reaction: reaction, + ); + + handler.handleEvent(event); + + verify(() => mutations.onReactionNew(message, reaction)).called(1); + }); + + test('reaction.updated delegates the message and reaction', () { + final event = Event( + type: EventType.reactionUpdated, + message: message, + reaction: reaction, + ); + + handler.handleEvent(event); + + verify(() => mutations.onReactionUpdated(message, reaction)).called(1); + }); + + test('reaction.deleted delegates the message and reaction', () { + final event = Event( + type: EventType.reactionDeleted, + message: message, + reaction: reaction, + ); + + handler.handleEvent(event); + + verify(() => mutations.onReactionDeleted(message, reaction)).called(1); + }); + + test('reaction events ignore events missing the reaction or message', () { + handler.handleEvent(Event(type: EventType.reactionNew, message: message)); + handler.handleEvent(Event(type: EventType.reactionNew, reaction: reaction)); + + verifyNever(() => mutations.onReactionNew(any(), any())); + }); + }); + + group('poll events', () { + final poll = Poll( + id: 'poll-id', + name: 'Favorite color?', + options: const [PollOption(id: 'option-a', text: 'A')], + ); + final pollVote = PollVote( + id: 'vote-id', + pollId: 'poll-id', + userId: otherUser.id, + optionId: 'option-a', + ); + + test('poll.created delegates the poll message', () { + final message = Message(id: 'message-id', poll: poll); + final event = Event(type: EventType.pollCreated, message: message); + + handler.handleEvent(event); + + verify(() => mutations.onPollCreated(message)).called(1); + }); + + test('poll.created ignores messages without a poll', () { + final message = Message(id: 'message-id'); + handler.handleEvent(Event(type: EventType.pollCreated, message: message)); + handler.handleEvent(Event(type: EventType.pollCreated)); + + verifyNever(() => mutations.onPollCreated(any())); + }); + + test('poll.updated and poll.closed delegate the poll', () { + handler.handleEvent(Event(type: EventType.pollUpdated, poll: poll)); + handler.handleEvent(Event(type: EventType.pollClosed, poll: poll)); + + verify(() => mutations.onPollUpdated(poll)).called(1); + verify(() => mutations.onPollClosed(poll)).called(1); + }); + + test('poll.updated ignores events without a poll', () { + handler.handleEvent(Event(type: EventType.pollUpdated)); + + verifyNever(() => mutations.onPollUpdated(any())); + }); + + test('poll vote events delegate the poll and vote', () { + handler.handleEvent(Event(type: EventType.pollAnswerCasted, poll: poll, pollVote: pollVote)); + handler.handleEvent(Event(type: EventType.pollVoteCasted, poll: poll, pollVote: pollVote)); + handler.handleEvent(Event(type: EventType.pollVoteChanged, poll: poll, pollVote: pollVote)); + handler.handleEvent(Event(type: EventType.pollAnswerRemoved, poll: poll, pollVote: pollVote)); + handler.handleEvent(Event(type: EventType.pollVoteRemoved, poll: poll, pollVote: pollVote)); + + verify(() => mutations.onPollAnswerCasted(poll, pollVote)).called(1); + verify(() => mutations.onPollVoteCasted(poll, pollVote)).called(1); + verify(() => mutations.onPollVoteChanged(poll, pollVote)).called(1); + verify(() => mutations.onPollAnswerRemoved(poll, pollVote)).called(1); + verify(() => mutations.onPollVoteRemoved(poll, pollVote)).called(1); + }); + + test('poll vote events ignore events missing the poll or vote', () { + handler.handleEvent(Event(type: EventType.pollVoteCasted, poll: poll)); + handler.handleEvent(Event(type: EventType.pollVoteCasted, pollVote: pollVote)); + + verifyNever(() => mutations.onPollVoteCasted(any(), any())); + }); + }); + + group('read events', () { + test('message.read delegates the read with the event creation time', () { + final event = Event( + type: EventType.messageRead, + user: otherUser, + createdAt: DateTime.now(), + lastReadMessageId: 'last-read-id', + ); + + handler.handleEvent(event); + + verify( + () => mutations.onMessageRead( + otherUser, + lastRead: event.createdAt, + lastReadMessageId: 'last-read-id', + ), + ).called(1); + verifyNever(() => client.channelDeliveryReporter.reconcileDelivery(any())); + }); + + test('notification.mark_read routes to the same read handling', () { + final event = Event(type: EventType.notificationMarkRead, user: otherUser); + + handler.handleEvent(event); + + verify( + () => mutations.onMessageRead( + otherUser, + lastRead: event.createdAt, + lastReadMessageId: null, + ), + ).called(1); + }); + + test('message.read from the current user reconciles channel delivery', () { + final event = Event( + type: EventType.messageRead, + user: User(id: currentUserId), + ); + + handler.handleEvent(event); + + verify(() => client.channelDeliveryReporter.reconcileDelivery([channel])).called(1); + }); + + test('message.read ignores thread reads', () { + final event = Event( + type: EventType.messageRead, + user: otherUser, + thread: Thread( + parentMessageId: 'parent-id', + channelCid: 'messaging:test', + createdByUserId: otherUser.id, + participantCount: 1, + replyCount: 1, + createdAt: DateTime.now(), + updatedAt: DateTime.now(), + ), + ); + + handler.handleEvent(event); + + verifyNever( + () => mutations.onMessageRead( + any(), + lastRead: any(named: 'lastRead'), + lastReadMessageId: any(named: 'lastReadMessageId'), + ), + ); + }); + + test('notification.mark_unread delegates the unread mark', () { + final lastReadAt = DateTime.now(); + final event = Event( + type: EventType.notificationMarkUnread, + user: otherUser, + lastReadAt: lastReadAt, + unreadMessages: 3, + lastReadMessageId: 'last-read-id', + ); + + handler.handleEvent(event); + + verify( + () => mutations.onNotificationMarkUnread( + otherUser, + lastRead: lastReadAt, + unreadMessages: 3, + lastReadMessageId: 'last-read-id', + ), + ).called(1); + }); + + test('message.delivered delegates the delivery info', () { + final deliveredAt = DateTime.now(); + final event = Event( + type: EventType.messageDelivered, + user: otherUser, + lastDeliveredAt: deliveredAt, + lastDeliveredMessageId: 'delivered-id', + ); + + handler.handleEvent(event); + + verify( + () => mutations.onMessageDelivered( + otherUser, + lastDeliveredAt: deliveredAt, + lastDeliveredMessageId: 'delivered-id', + ), + ).called(1); + verifyNever(() => client.channelDeliveryReporter.reconcileDelivery(any())); + }); + + test('message.delivered from the current user reconciles channel delivery', () { + final event = Event( + type: EventType.messageDelivered, + user: User(id: currentUserId), + ); + + handler.handleEvent(event); + + verify(() => client.channelDeliveryReporter.reconcileDelivery([channel])).called(1); + }); + }); + + group('channel events', () { + test('channel.truncated wipes persisted messages before mutating', () async { + final persistentClient = MockStreamChatClientWithPersistence(); + when(() => channel.client).thenReturn(persistentClient); + when(() => persistentClient.state).thenReturn(FakeClientState()); + + final persistence = persistentClient.chatPersistenceClient; + when(() => persistence.deleteMessageByCid(any())).thenAnswer((_) async {}); + + final systemMessage = Message(id: 'system-id'); + final event = Event( + type: EventType.channelTruncated, + channel: ChannelModel(cid: 'messaging:test'), + message: systemMessage, + ); + + handler.handleEvent(event); + await Future.value(); + + verifyInOrder([ + () => persistence.deleteMessageByCid('messaging:test'), + () => mutations.onChannelTruncated(message: systemMessage), + ]); + }); + + test('channel.updated delegates the channel model', () { + final channelModel = ChannelModel(cid: 'messaging:test'); + final event = Event(type: EventType.channelUpdated, channel: channelModel); + + handler.handleEvent(event); + + verify(() => mutations.onChannelUpdated(channelModel)).called(1); + }); + + test('any event carrying a channel message count delegates it', () { + handler.handleEvent(Event(type: 'unknown.event', channelMessageCount: 42)); + + verify(() => mutations.onChannelMessageCount(42)).called(1); + }); + + test('events without a channel message count skip the count update', () { + handler.handleEvent(Event(type: 'unknown.event')); + + verifyNever(() => mutations.onChannelMessageCount(any())); + }); + + test('channel.push_preference.updated delegates the preferences', () { + const pushPreference = ChannelPushPreference(chatLevel: ChatLevel.mentions); + final event = Event( + type: EventType.channelPushPreferenceUpdated, + channelPushPreference: pushPreference, + ); + + handler.handleEvent(event); + + verify(() => mutations.onChannelPushPreferenceUpdated(pushPreference)).called(1); + }); + + test('channel.push_preference.updated ignores events without preferences', () { + handler.handleEvent(Event(type: EventType.channelPushPreferenceUpdated)); + + verifyNever(() => mutations.onChannelPushPreferenceUpdated(any())); + }); + }); + + group('member events', () { + final member = Member(userId: otherUser.id); + + test('member.added delegates the member', () { + final event = Event(type: EventType.memberAdded, member: member); + + handler.handleEvent(event); + + verify(() => mutations.onMemberAdded(member)).called(1); + }); + + test('member.removed delegates the removed user', () { + final event = Event(type: EventType.memberRemoved, user: otherUser); + + handler.handleEvent(event); + + verify(() => mutations.onMemberRemoved(otherUser)).called(1); + }); + + test('member.updated delegates the member', () { + final event = Event(type: EventType.memberUpdated, member: member, user: otherUser); + + handler.handleEvent(event); + + verify(() => mutations.onMemberUpdated(member)).called(1); + }); + + test('any event carrying a user delegates the member-user merge', () { + handler.handleEvent(Event(type: 'unknown.event', user: otherUser)); + + verify(() => mutations.onMemberUserUpdated(otherUser)).called(1); + }); + + test('user.banned refreshes the member before delegating', () async { + when( + () => channel.queryMembers(filter: any(named: 'filter')), + ).thenAnswer((_) async => FakeQueryMembersResponse([member])); + + final event = Event( + type: EventType.userBanned, + cid: 'messaging:test', + user: otherUser, + ); + + handler.handleEvent(event); + await Future.value(); + + verify(() => channel.queryMembers(filter: Filter.equal('id', otherUser.id))).called(1); + verify(() => mutations.onMemberBanned(member)).called(1); + }); + + test('user.banned ignores app-level bans without a cid', () async { + handler.handleEvent(Event(type: EventType.userBanned, user: otherUser)); + await Future.value(); + + verifyNever(() => channel.queryMembers(filter: any(named: 'filter'))); + verifyNever(() => mutations.onMemberBanned(any())); + }); + + test('user.unbanned refreshes the member before delegating', () async { + when( + () => channel.queryMembers(filter: any(named: 'filter')), + ).thenAnswer((_) async => FakeQueryMembersResponse([member])); + + final event = Event( + type: EventType.userUnbanned, + cid: 'messaging:test', + user: otherUser, + ); + + handler.handleEvent(event); + await Future.value(); + + verify(() => mutations.onMemberUnbanned(member)).called(1); + }); + + test('user.messages.deleted delegates the deletion parameters', () async { + final event = Event( + type: EventType.userMessagesDeleted, + user: otherUser, + hardDelete: true, + ); + + handler.handleEvent(event); + await Future.value(); + + verify( + () => mutations.onUserMessagesDeleted( + userId: otherUser.id, + hardDelete: true, + deletedAt: event.createdAt, + ), + ).called(1); + }); + + test('user.messages.deleted ignores events without a user', () async { + handler.handleEvent(Event(type: EventType.userMessagesDeleted)); + await Future.value(); + + verifyNever( + () => mutations.onUserMessagesDeleted( + userId: any(named: 'userId'), + hardDelete: any(named: 'hardDelete'), + deletedAt: any(named: 'deletedAt'), + ), + ); + }); + }); + + group('watching events', () { + test('user.watching.start delegates the watcher and count', () { + final event = Event( + type: EventType.userWatchingStart, + user: otherUser, + watcherCount: 3, + ); + + handler.handleEvent(event); + + verify(() => mutations.onUserStartWatching(otherUser, watcherCount: 3)).called(1); + }); + + test('user.watching.stop delegates the watcher and count', () { + final event = Event( + type: EventType.userWatchingStop, + user: otherUser, + watcherCount: 2, + ); + + handler.handleEvent(event); + + verify(() => mutations.onUserStopWatching(otherUser, watcherCount: 2)).called(1); + }); + + test('watching events ignore events without a user', () { + handler.handleEvent(Event(type: EventType.userWatchingStart)); + handler.handleEvent(Event(type: EventType.userWatchingStop)); + + verifyNever(() => mutations.onUserStartWatching(any(), watcherCount: any(named: 'watcherCount'))); + verifyNever(() => mutations.onUserStopWatching(any(), watcherCount: any(named: 'watcherCount'))); + }); + }); + + group('reminder events', () { + final reminder = MessageReminder( + messageId: 'message-id', + channelCid: 'messaging:test', + userId: currentUserId, + remindAt: DateTime.now(), + ); + + test('reminder.created and reminder.updated delegate the reminder', () { + handler.handleEvent(Event(type: EventType.reminderCreated, reminder: reminder)); + handler.handleEvent(Event(type: EventType.reminderUpdated, reminder: reminder)); + + verify(() => mutations.onReminderCreated(reminder)).called(1); + verify(() => mutations.onReminderUpdated(reminder)).called(1); + }); + + test('reminder.deleted delegates the reminder', () { + handler.handleEvent(Event(type: EventType.reminderDeleted, reminder: reminder)); + + verify(() => mutations.onReminderDeleted(reminder)).called(1); + }); + + test('reminder events ignore events without a reminder', () { + handler.handleEvent(Event(type: EventType.reminderCreated)); + handler.handleEvent(Event(type: EventType.reminderDeleted)); + + verifyNever(() => mutations.onReminderCreated(any())); + verifyNever(() => mutations.onReminderDeleted(any())); + }); + }); + + group('location events', () { + final location = Location( + channelCid: 'messaging:test', + messageId: 'message-id', + userId: currentUserId, + latitude: 1, + longitude: 2, + createdByDeviceId: 'device-id', + ); + + test('location.shared delegates the location message', () { + final message = Message(id: 'message-id', sharedLocation: location); + final event = Event(type: EventType.locationShared, message: message); + + handler.handleEvent(event); + + verify(() => mutations.onLocationShared(message)).called(1); + }); + + test('location.shared ignores messages without a location', () { + final message = Message(id: 'message-id'); + handler.handleEvent(Event(type: EventType.locationShared, message: message)); + + verifyNever(() => mutations.onLocationShared(any())); + }); + + test('location.updated delegates the location', () { + final message = Message(id: 'message-id', sharedLocation: location); + final event = Event(type: EventType.locationUpdated, message: message); + + handler.handleEvent(event); + + verify(() => mutations.onLocationUpdated(location)).called(1); + }); + + test('location.expired delegates the location', () { + final message = Message(id: 'message-id', sharedLocation: location); + final event = Event(type: EventType.locationExpired, message: message); + + handler.handleEvent(event); + + verify(() => mutations.onLocationExpired(location)).called(1); + }); + + test('location events ignore events without a location', () { + handler.handleEvent(Event(type: EventType.locationUpdated)); + handler.handleEvent(Event(type: EventType.locationExpired)); + + verifyNever(() => mutations.onLocationUpdated(any())); + verifyNever(() => mutations.onLocationExpired(any())); + }); + }); +} diff --git a/packages/stream_chat/test/src/client/channel_state_mutations_test.dart b/packages/stream_chat/test/src/client/channel_state_mutations_test.dart new file mode 100644 index 000000000..c3ccec5fe --- /dev/null +++ b/packages/stream_chat/test/src/client/channel_state_mutations_test.dart @@ -0,0 +1,750 @@ +// ignore_for_file: cascade_invocations + +import 'package:mocktail/mocktail.dart'; +import 'package:stream_chat/src/client/channel_state_mutations.dart'; +import 'package:stream_chat/stream_chat.dart'; +import 'package:test/test.dart'; + +import '../fakes.dart'; +import '../mocks.dart'; + +class MockChannelClientState extends Mock implements ChannelClientState {} + +void main() { + late Channel channel; + late MockStreamChatClient client; + late MockChannelClientState state; + late ChannelStateMutations mutations; + + late List<(User, Event)> upsertTypingEventCalls; + late List removeTypingEventCalls; + late List<(User, int?)> removeWatcherCalls; + late List updateMemberCalls; + late List<(String, bool, DateTime?)> deleteMessagesFromUserCalls; + + // Matches the default current user of [FakeClientState]. + const currentUserId = 'test-user-id'; + final otherUser = User(id: 'other-user'); + + setUpAll(() { + registerFallbackValue(FakeMessage()); + registerFallbackValue(FakeChannelState()); + registerFallbackValue(0); + }); + + setUp(() { + client = MockStreamChatClient(); + state = MockChannelClientState(); + + when(() => client.logger).thenReturn(MockLogger()); + when(() => client.state).thenReturn(FakeClientState()); + + // A real channel (without an attached state) so that the capability and + // unread-count lookups the mutations rely on resolve to their defaults. + channel = Channel(client, 'messaging', 'test'); + + upsertTypingEventCalls = []; + removeTypingEventCalls = []; + removeWatcherCalls = []; + updateMemberCalls = []; + deleteMessagesFromUserCalls = []; + + mutations = ChannelStateMutations( + channel: channel, + state: state, + upsertTypingEvent: (user, event) => upsertTypingEventCalls.add((user, event)), + removeTypingEvent: removeTypingEventCalls.add, + removeWatcher: (watcher, {watcherCount}) => removeWatcherCalls.add((watcher, watcherCount)), + updateMember: updateMemberCalls.add, + deleteMessagesFromUser: ({required userId, hardDelete = false, deletedAt}) async { + deleteMessagesFromUserCalls.add((userId, hardDelete, deletedAt)); + }, + ); + }); + + ChannelState stubChannelState(ChannelState channelState) { + when(() => state.channelState).thenReturn(channelState); + return channelState; + } + + ChannelState capturedChannelState() { + return verify(() => state.updateChannelState(captureAny())).captured.single as ChannelState; + } + + group('typing', () { + test('onTypingStart records the typing event for the user', () { + final event = Event(type: EventType.typingStart, user: otherUser); + + mutations.onTypingStart(otherUser, event); + + expect(upsertTypingEventCalls, [(otherUser, event)]); + }); + + test('onTypingStop clears the typing event for the user', () { + mutations.onTypingStop(otherUser); + + expect(removeTypingEventCalls, [otherUser]); + }); + }); + + group('messages', () { + final message = Message(id: 'message-id', user: otherUser); + + test('onMessageNew adds the message', () { + mutations.onMessageNew(message); + + verify(() => state.addNewMessage(message)).called(1); + verifyNever(() => state.updateChannelState(any())); + }); + + test('onMessageNew also applies the watcher count when provided', () { + stubChannelState(ChannelState(channel: ChannelModel(cid: 'messaging:test'))); + + mutations.onMessageNew(message, watcherCount: 7); + + verify(() => state.addNewMessage(message)).called(1); + expect(capturedChannelState().watcherCount, 7); + }); + + test('onMessageDeleted soft-deletes without touching the unread count', () { + mutations.onMessageDeleted(message); + + verify(() => state.deleteMessage(message, hardDelete: false)).called(1); + verifyNever(() => state.unreadCount = any()); + }); + + test('onMessageDeleted decrements the local unread count on hard delete', () { + client.isLocalUnreadCountEnabled = true; + when(() => state.unreadCount).thenReturn(5); + + mutations.onMessageDeleted(message, hardDelete: true); + + verify(() => state.unreadCount = 4).called(1); + verify(() => state.deleteMessage(message, hardDelete: true)).called(1); + }); + + test('onMessageDeleted never decrements the unread count below zero', () { + client.isLocalUnreadCountEnabled = true; + when(() => state.unreadCount).thenReturn(0); + + mutations.onMessageDeleted(message, hardDelete: true); + + verify(() => state.unreadCount = 0).called(1); + }); + + test('onMessageDeleted skips the unread count on server-driven channels', () { + client.isLocalUnreadCountEnabled = false; + + mutations.onMessageDeleted(message, hardDelete: true); + + verifyNever(() => state.unreadCount = any()); + verify(() => state.deleteMessage(message, hardDelete: true)).called(1); + }); + + test('onMessageUpdated applies the message without upserting', () { + mutations.onMessageUpdated(message); + + verify(() => state.updateMessage(message, upsert: false)).called(1); + }); + }); + + group('drafts', () { + final draft = Draft( + channelCid: 'messaging:test', + createdAt: DateTime.now(), + message: DraftMessage(text: 'draft'), + ); + + test('onDraftUpdated applies the draft', () { + mutations.onDraftUpdated(draft); + + verify(() => state.updateDraft(draft)).called(1); + }); + + test('onDraftDeleted removes the draft', () { + mutations.onDraftDeleted(draft); + + verify(() => state.deleteDraft(draft)).called(1); + }); + }); + + group('reactions', () { + final ownReaction = Reaction( + type: 'like', + messageId: 'message-id', + userId: currentUserId, + createdAt: DateTime.now(), + ); + + Message capturedMessage() { + return verify(() => state.updateMessage(captureAny())).captured.single as Message; + } + + test('onReactionNew adds an own reaction of the current user', () { + final existing = Message(id: 'message-id', user: otherUser); + when(() => state.messages).thenReturn([existing]); + when(() => state.threads).thenReturn(const {}); + + final eventMessage = Message(id: 'message-id', user: otherUser, latestReactions: [ownReaction]); + mutations.onReactionNew(eventMessage, ownReaction); + + expect(capturedMessage().ownReactions, [ownReaction]); + }); + + test('onReactionNew preserves own reactions for other users', () { + final existing = Message(id: 'message-id', user: otherUser, ownReactions: [ownReaction]); + when(() => state.messages).thenReturn([existing]); + when(() => state.threads).thenReturn(const {}); + + final theirReaction = Reaction( + type: 'love', + messageId: 'message-id', + userId: otherUser.id, + createdAt: DateTime.now(), + ); + final eventMessage = Message(id: 'message-id', user: otherUser); + mutations.onReactionNew(eventMessage, theirReaction); + + expect(capturedMessage().ownReactions, [ownReaction]); + }); + + test('onReactionNew finds the message in a thread', () { + final existing = Message(id: 'message-id', parentId: 'parent-id', user: otherUser); + when(() => state.messages).thenReturn([]); + when(() => state.threads).thenReturn({ + 'parent-id': [existing], + }); + + final eventMessage = Message(id: 'message-id', parentId: 'parent-id', user: otherUser); + mutations.onReactionNew(eventMessage, ownReaction); + + expect(capturedMessage().ownReactions, [ownReaction]); + }); + + test('onReactionNew ignores reactions to unknown messages', () { + when(() => state.messages).thenReturn([]); + when(() => state.threads).thenReturn(const {}); + + mutations.onReactionNew(Message(id: 'message-id'), ownReaction); + + verifyNever(() => state.updateMessage(any())); + }); + + test('onReactionUpdated replaces own reactions of the current user', () { + final oldReaction = Reaction( + type: 'love', + messageId: 'message-id', + userId: currentUserId, + createdAt: DateTime.now(), + ); + final existing = Message(id: 'message-id', user: otherUser, ownReactions: [oldReaction]); + when(() => state.messages).thenReturn([existing]); + when(() => state.threads).thenReturn(const {}); + + mutations.onReactionUpdated(Message(id: 'message-id', user: otherUser), ownReaction); + + expect(capturedMessage().ownReactions, [ownReaction]); + }); + + test('onReactionDeleted removes the own reaction of the current user', () { + final existing = Message(id: 'message-id', user: otherUser, ownReactions: [ownReaction]); + when(() => state.messages).thenReturn([existing]); + when(() => state.threads).thenReturn(const {}); + + mutations.onReactionDeleted(Message(id: 'message-id', user: otherUser), ownReaction); + + expect(capturedMessage().ownReactions, isEmpty); + }); + }); + + group('polls', () { + const pollId = 'poll-id'; + + Poll createPoll({ + String name = 'Favorite color?', + List latestAnswers = const [], + List ownVotesAndAnswers = const [], + bool isClosed = false, + }) { + return Poll( + id: pollId, + name: name, + options: const [PollOption(id: 'option-a', text: 'A')], + latestAnswers: latestAnswers, + ownVotesAndAnswers: ownVotesAndAnswers, + isClosed: isClosed, + ); + } + + PollVote createVote(String id, {String? userId, String optionId = 'option-a'}) { + return PollVote( + id: id, + pollId: pollId, + userId: userId ?? currentUserId, + optionId: optionId, + ); + } + + Message capturedMessage() { + return verify(() => state.updateMessage(captureAny())).captured.single as Message; + } + + void seedPollMessage(Poll poll) { + final message = Message(id: 'poll-message-id', poll: poll); + when(() => state.messages).thenReturn([message]); + when(() => state.threads).thenReturn(const {}); + } + + test('onPollCreated adds the poll message', () { + final message = Message(id: 'poll-message-id', poll: createPoll()); + + mutations.onPollCreated(message); + + verify(() => state.addNewMessage(message)).called(1); + }); + + test('onPollUpdated applies the poll preserving known answers and votes', () { + final answer = createVote('answer-1'); + final ownVote = createVote('vote-1'); + seedPollMessage(createPoll(latestAnswers: [answer], ownVotesAndAnswers: [ownVote])); + + mutations.onPollUpdated(createPoll(name: 'Updated?')); + + final poll = capturedMessage().poll!; + expect(poll.name, 'Updated?'); + expect(poll.latestAnswers, [answer]); + expect(poll.ownVotesAndAnswers, [ownVote]); + }); + + test('onPollUpdated ignores polls without a matching message', () { + when(() => state.messages).thenReturn([]); + when(() => state.threads).thenReturn(const {}); + + mutations.onPollUpdated(createPoll()); + + verifyNever(() => state.updateMessage(any())); + }); + + test('onPollClosed closes the known poll', () { + seedPollMessage(createPoll()); + + mutations.onPollClosed(createPoll(isClosed: true)); + + final poll = capturedMessage().poll!; + expect(poll.isClosed, isTrue); + expect(poll.name, 'Favorite color?'); + }); + + test('onPollAnswerCasted appends the answer and tracks own answers', () { + final existingAnswer = createVote('answer-1', userId: otherUser.id); + seedPollMessage(createPoll(latestAnswers: [existingAnswer])); + + final castedAnswer = createVote('answer-2'); + mutations.onPollAnswerCasted(createPoll(), castedAnswer); + + final poll = capturedMessage().poll!; + expect(poll.latestAnswers, [existingAnswer, castedAnswer]); + expect(poll.ownVotesAndAnswers, [castedAnswer]); + }); + + test('onPollAnswerCasted skips own tracking for other users', () { + seedPollMessage(createPoll()); + + final castedAnswer = createVote('answer-2', userId: otherUser.id); + mutations.onPollAnswerCasted(createPoll(), castedAnswer); + + final poll = capturedMessage().poll!; + expect(poll.latestAnswers, [castedAnswer]); + expect(poll.ownVotesAndAnswers, isEmpty); + }); + + test('onPollVoteCasted tracks own votes preserving known answers', () { + final answer = createVote('answer-1', userId: otherUser.id); + seedPollMessage(createPoll(latestAnswers: [answer])); + + final vote = createVote('vote-1'); + mutations.onPollVoteCasted(createPoll(), vote); + + final poll = capturedMessage().poll!; + expect(poll.latestAnswers, [answer]); + expect(poll.ownVotesAndAnswers, [vote]); + }); + + test('onPollVoteChanged replaces the own vote with the same id', () { + final oldVote = createVote('vote-1'); + seedPollMessage(createPoll(ownVotesAndAnswers: [oldVote])); + + final changedVote = createVote('vote-1', optionId: 'option-b'); + mutations.onPollVoteChanged(createPoll(), changedVote); + + final poll = capturedMessage().poll!; + expect(poll.ownVotesAndAnswers, [changedVote]); + }); + + test('onPollAnswerRemoved removes the answer from both lists', () { + final answer = createVote('answer-1'); + seedPollMessage(createPoll(latestAnswers: [answer], ownVotesAndAnswers: [answer])); + + mutations.onPollAnswerRemoved(createPoll(), answer); + + final poll = capturedMessage().poll!; + expect(poll.latestAnswers, isEmpty); + expect(poll.ownVotesAndAnswers, isEmpty); + }); + + test('onPollVoteRemoved removes the own vote preserving known answers', () { + final answer = createVote('answer-1', userId: otherUser.id); + final vote = createVote('vote-1'); + seedPollMessage(createPoll(latestAnswers: [answer], ownVotesAndAnswers: [vote])); + + mutations.onPollVoteRemoved(createPoll(), vote); + + final poll = capturedMessage().poll!; + expect(poll.latestAnswers, [answer]); + expect(poll.ownVotesAndAnswers, isEmpty); + }); + }); + + group('reads', () { + test('onMessageRead resets the unread count preserving delivery info', () { + final deliveredAt = DateTime.now(); + final existingRead = Read( + user: otherUser, + lastRead: DateTime.now().subtract(const Duration(days: 1)), + unreadMessages: 5, + lastDeliveredAt: deliveredAt, + lastDeliveredMessageId: 'delivered-id', + ); + when(() => state.read).thenReturn([existingRead]); + + final lastRead = DateTime.now(); + mutations.onMessageRead(otherUser, lastRead: lastRead, lastReadMessageId: 'read-id'); + + final expectedRead = Read( + user: otherUser, + lastRead: lastRead, + unreadMessages: 0, + lastReadMessageId: 'read-id', + lastDeliveredAt: deliveredAt, + lastDeliveredMessageId: 'delivered-id', + ); + verify(() => state.updateRead([expectedRead])).called(1); + }); + + test('onNotificationMarkUnread applies the unread mark preserving delivery info', () { + final deliveredAt = DateTime.now(); + final existingRead = Read( + user: otherUser, + lastRead: DateTime.now().subtract(const Duration(days: 1)), + lastDeliveredAt: deliveredAt, + ); + when(() => state.read).thenReturn([existingRead]); + + final lastRead = DateTime.now(); + mutations.onNotificationMarkUnread( + otherUser, + lastRead: lastRead, + unreadMessages: 3, + lastReadMessageId: 'read-id', + ); + + final expectedRead = Read( + user: otherUser, + lastRead: lastRead, + unreadMessages: 3, + lastReadMessageId: 'read-id', + lastDeliveredAt: deliveredAt, + ); + verify(() => state.updateRead([expectedRead])).called(1); + }); + + test('onMessageDelivered applies the delivery preserving read info', () { + final lastRead = DateTime.now().subtract(const Duration(days: 1)); + final existingRead = Read( + user: otherUser, + lastRead: lastRead, + unreadMessages: 2, + lastReadMessageId: 'read-id', + ); + when(() => state.read).thenReturn([existingRead]); + + final deliveredAt = DateTime.now(); + mutations.onMessageDelivered( + otherUser, + lastDeliveredAt: deliveredAt, + lastDeliveredMessageId: 'delivered-id', + ); + + final expectedRead = Read( + user: otherUser, + lastRead: lastRead, + unreadMessages: 2, + lastReadMessageId: 'read-id', + lastDeliveredAt: deliveredAt, + lastDeliveredMessageId: 'delivered-id', + ); + verify(() => state.updateRead([expectedRead])).called(1); + }); + + test('onMessageDelivered falls back to the epoch without a current read', () { + when(() => state.read).thenReturn([]); + + mutations.onMessageDelivered(otherUser); + + final expectedRead = Read( + user: otherUser, + lastRead: DateTime.fromMillisecondsSinceEpoch(0, isUtc: true), + ); + verify(() => state.updateRead([expectedRead])).called(1); + }); + }); + + group('channel', () { + test('onChannelTruncated truncates before applying the system message', () { + final systemMessage = Message(id: 'system-id'); + + mutations.onChannelTruncated(message: systemMessage); + + verifyInOrder([ + () => state.truncate(), + () => state.updateMessage(systemMessage), + ]); + }); + + test('onChannelTruncated only truncates without a system message', () { + mutations.onChannelTruncated(); + + verify(() => state.truncate()).called(1); + verifyNever(() => state.updateMessage(any())); + }); + + test('onChannelUpdated merges the channel and replaces the members', () { + final member = Member(user: otherUser); + stubChannelState(ChannelState(channel: ChannelModel(cid: 'messaging:test'))); + + mutations.onChannelUpdated( + ChannelModel(cid: 'messaging:test', frozen: true, members: [member]), + ); + + final updated = capturedChannelState(); + expect(updated.channel?.frozen, isTrue); + expect(updated.members, [member]); + }); + + test('onChannelMessageCount updates the message count', () { + stubChannelState(ChannelState(channel: ChannelModel(cid: 'messaging:test'))); + + mutations.onChannelMessageCount(42); + + expect(capturedChannelState().channel?.messageCount, 42); + }); + + test('onChannelPushPreferenceUpdated applies the preferences', () { + const pushPreference = ChannelPushPreference(chatLevel: ChatLevel.mentions); + stubChannelState(ChannelState(channel: ChannelModel(cid: 'messaging:test'))); + + mutations.onChannelPushPreferenceUpdated(pushPreference); + + expect(capturedChannelState().pushPreferences, pushPreference); + }); + }); + + group('members', () { + final member = Member(user: otherUser); + + test('onMemberAdded appends the member', () { + final existingMember = Member(user: User(id: 'existing-user')); + stubChannelState(ChannelState(members: [existingMember])); + + mutations.onMemberAdded(member); + + expect(capturedChannelState().members, [existingMember, member]); + }); + + test('onMemberRemoved removes the member and its read state', () { + final existingMember = Member(user: User(id: 'existing-user')); + final existingRead = Read( + user: User(id: 'existing-user'), + lastRead: DateTime.now(), + ); + stubChannelState( + ChannelState( + members: [existingMember, member], + read: [ + existingRead, + Read(user: otherUser, lastRead: DateTime.now()), + ], + ), + ); + + mutations.onMemberRemoved(otherUser); + + final updated = capturedChannelState(); + expect(updated.members, [existingMember]); + expect(updated.read, [existingRead]); + }); + + test('onMemberUserUpdated merges the user into the member and membership', () { + final updatedUser = User(id: otherUser.id, name: 'Updated'); + stubChannelState(ChannelState(members: [member], membership: member)); + + mutations.onMemberUserUpdated(updatedUser); + + final updated = capturedChannelState(); + expect(updated.members?.single.user, updatedUser); + expect(updated.membership?.user, updatedUser); + }); + + test('onMemberUserUpdated ignores users that are not members', () { + stubChannelState(ChannelState(members: [member])); + + mutations.onMemberUserUpdated(User(id: 'not-a-member')); + + verifyNever(() => state.updateChannelState(any())); + }); + + test('onMemberUpdated replaces the matching member and membership', () { + final updatedMember = Member(user: otherUser, channelRole: 'admin'); + stubChannelState(ChannelState(members: [member], membership: member)); + + mutations.onMemberUpdated(updatedMember); + + final updated = capturedChannelState(); + expect(updated.members, [updatedMember]); + expect(updated.membership, updatedMember); + }); + + test('onMemberBanned and onMemberUnbanned replace the refreshed member', () { + mutations.onMemberBanned(member); + mutations.onMemberUnbanned(member); + + expect(updateMemberCalls, [member, member]); + }); + + test('onUserMessagesDeleted deletes the user messages', () async { + final deletedAt = DateTime.now(); + + await mutations.onUserMessagesDeleted( + userId: otherUser.id, + hardDelete: true, + deletedAt: deletedAt, + ); + + expect(deleteMessagesFromUserCalls, [(otherUser.id, true, deletedAt)]); + }); + }); + + group('watchers', () { + test('onUserStartWatching upserts the watcher and count', () { + final existingWatcher = User(id: 'existing-watcher'); + stubChannelState(ChannelState(watchers: [existingWatcher, otherUser])); + + final rejoined = User(id: otherUser.id, name: 'Rejoined'); + mutations.onUserStartWatching(rejoined, watcherCount: 3); + + final updated = capturedChannelState(); + expect(updated.watchers, [rejoined, existingWatcher]); + expect(updated.watcherCount, 3); + }); + + test('onUserStopWatching removes the watcher', () { + mutations.onUserStopWatching(otherUser, watcherCount: 1); + + expect(removeWatcherCalls, [(otherUser, 1)]); + }); + }); + + group('reminders', () { + final reminder = MessageReminder( + messageId: 'message-id', + channelCid: 'messaging:test', + userId: currentUserId, + remindAt: DateTime.now(), + ); + + test('onReminderCreated and onReminderUpdated apply the reminder', () { + mutations.onReminderCreated(reminder); + mutations.onReminderUpdated(reminder); + + verify(() => state.updateReminder(reminder)).called(2); + }); + + test('onReminderDeleted removes the reminder', () { + mutations.onReminderDeleted(reminder); + + verify(() => state.deleteReminder(reminder)).called(1); + }); + }); + + group('locations', () { + Location createLocation({String? messageId = 'message-id', double latitude = 1}) { + return Location( + channelCid: 'messaging:test', + messageId: messageId, + userId: currentUserId, + latitude: latitude, + longitude: 2, + createdByDeviceId: 'device-id', + ); + } + + test('onLocationShared adds the location message', () { + final message = Message(id: 'message-id', sharedLocation: createLocation()); + + mutations.onLocationShared(message); + + verify(() => state.addNewMessage(message)).called(1); + }); + + test('onLocationUpdated applies the location to the message sharing it', () { + final message = Message(id: 'message-id', sharedLocation: createLocation()); + when(() => state.messages).thenReturn([message]); + when(() => state.threads).thenReturn(const {}); + + final updatedLocation = createLocation(latitude: 42); + mutations.onLocationUpdated(updatedLocation); + + final captured = verify(() => state.updateMessage(captureAny())).captured.single as Message; + expect(captured.sharedLocation, updatedLocation); + }); + + test('onLocationUpdated finds the message in a thread', () { + final message = Message(id: 'message-id', parentId: 'parent-id', sharedLocation: createLocation()); + when(() => state.messages).thenReturn([]); + when(() => state.threads).thenReturn({ + 'parent-id': [message], + }); + + mutations.onLocationUpdated(createLocation(latitude: 42)); + + verify(() => state.updateMessage(any())).called(1); + }); + + test('onLocationUpdated ignores locations without a message id', () { + mutations.onLocationUpdated(createLocation(messageId: null)); + + verifyNever(() => state.updateMessage(any())); + }); + + test('onLocationUpdated ignores locations of unknown messages', () { + when(() => state.messages).thenReturn([]); + when(() => state.threads).thenReturn(const {}); + + mutations.onLocationUpdated(createLocation()); + + verifyNever(() => state.updateMessage(any())); + }); + + test('onLocationExpired applies the expired location', () { + final message = Message(id: 'message-id', sharedLocation: createLocation()); + when(() => state.messages).thenReturn([message]); + when(() => state.threads).thenReturn(const {}); + + final expiredLocation = createLocation(latitude: 42); + mutations.onLocationExpired(expiredLocation); + + final captured = verify(() => state.updateMessage(captureAny())).captured.single as Message; + expect(captured.sharedLocation, expiredLocation); + }); + }); +}