diff --git a/packages/stream_chat/lib/src/client/channel.dart b/packages/stream_chat/lib/src/client/channel.dart index 5cfae938aa..d543da1f64 100644 --- a/packages/stream_chat/lib/src/client/channel.dart +++ b/packages/stream_chat/lib/src/client/channel.dart @@ -6,6 +6,8 @@ import 'dart:math' as math; import 'package:collection/collection.dart'; import 'package:rxdart/rxdart.dart'; import 'package:stream_chat/src/client/retry_queue.dart'; +import 'package:stream_chat/src/core/util/message_merging.dart'; +import 'package:stream_chat/src/core/util/message_predicates.dart'; import 'package:stream_chat/src/core/util/utils.dart'; import 'package:stream_chat/stream_chat.dart'; import 'package:synchronized/synchronized.dart'; @@ -3418,16 +3420,7 @@ class ChannelClientState { /// state — [message] is used as-is. Useful for local rollbacks of an /// optimistic update, where the caller has the full prior snapshot and /// doesn't want the merge falling back to the optimistic values. - void replaceMessage(Message message) => _updateMessages([message], update: _replaceUpdate); - - // Default `update` for [_updateMessages]: merge incoming with the - // locally-known message via `Message.updateWith`, preserving enrichment - // the server may strip on partial payloads. - static Message _mergeUpdate(Message original, Message updated) => original.updateWith(updated); - - // Replace `update` for [_updateMessages]: take the incoming as-is. Used - // by local rollback paths. - static Message _replaceUpdate(Message _, Message updated) => updated; + void replaceMessage(Message message) => _updateMessages([message], update: MessageMerging.replaceUpdate); /// Cleans up all the stale error messages which requires no action. void cleanUpStaleErrorMessages() { @@ -3790,8 +3783,8 @@ class ChannelClientState { final newMessages = messages.mergeSorted( updatedState.messages, key: (message) => message.id, - update: _mergeUpdate, - compare: _sortByCreatedAt, + update: MessageMerging.mergeUpdate, + compare: MessageMerging.sortByCreatedAt, ); final watchers = _channelState.watchers ?? const []; @@ -3863,8 +3856,6 @@ class ChannelClientState { return remoteState.copyWith(read: preservedReads.toList()); } - int _sortByCreatedAt(Message a, Message b) => a.createdAt.compareTo(b.createdAt); - /// The channel state related to this client. ChannelState get _channelState => _channelStateController.value; @@ -3925,7 +3916,7 @@ class ChannelClientState { final updatedThreads = {...threads}; final threadMessages = updatedThreads[parentId] ?? []; - final updatedThreadMessages = _mergeMessagesIntoExisting( + final updatedThreadMessages = MessageMerging.mergeMessages( existing: threadMessages, toMerge: messages.where((it) => it.id != parentId), ); @@ -4036,7 +4027,7 @@ class ChannelClientState { updateChannelState( _channelState.copyWith( - pinnedMessages: pinnedMessages.where(_pinIsValid).toList(), + pinnedMessages: pinnedMessages.where((it) => it.hasValidPin).toList(), messages: expiredMessages, ), ); @@ -4147,7 +4138,7 @@ class ChannelClientState { void _updateMessages( Iterable messages, { - Message Function(Message original, Message updated) update = _mergeUpdate, + Message Function(Message original, Message updated) update = MessageMerging.mergeUpdate, bool upsert = true, }) { if (messages.isEmpty) return; @@ -4160,43 +4151,21 @@ class ChannelClientState { void _updateThreadMessages( Iterable messages, { - Message Function(Message original, Message updated) update = _mergeUpdate, + Message Function(Message original, Message updated) update = MessageMerging.mergeUpdate, bool upsert = true, }) { if (messages.isEmpty) return; - // Group messages by parentId so each thread merge only sees its own - // replies — passing the full batch to every thread would leak replies - // across thread boundaries (the merge dedups by id, not by parentId). - final messagesByThread = >{}; - for (final m in messages) { - if (m.parentId case final parentId?) (messagesByThread[parentId] ??= []).add(m); - } - - // If there are no affected threads, return early. - if (messagesByThread.isEmpty) return; - - final updatedThreads = {...threads}; - for (final MapEntry(key: thread, :value) in messagesByThread.entries) { - final existingThreadMessages = updatedThreads[thread]; - - // Don't create a phantom entry for a thread that wasn't loaded: with - // `upsert: false` an out-of-window reply is dropped, so there's nothing - // to merge. Writing it back would make `threads.containsKey(parentId)` - // report a thread that was never paged in. - if (existingThreadMessages == null && !upsert) continue; - - final threadMessages = existingThreadMessages ?? []; - final updatedThreadMessages = _mergeMessagesIntoExisting( - existing: threadMessages, - toMerge: value, - update: update, - upsert: upsert, - ); + final currentThreads = threads; + final updatedThreads = MessageMerging.mergeThreadMessages( + existing: currentThreads, + toMerge: messages, + update: update, + upsert: upsert, + ); - // Update the thread with the modified message list. - updatedThreads[thread] = updatedThreadMessages.toList(); - } + // Nothing targeted a thread — skip the write. + if (identical(updatedThreads, currentThreads)) return; // Update the threads map. _threads = updatedThreads; @@ -4204,25 +4173,19 @@ class ChannelClientState { void _updateChannelMessages( Iterable messages, { - Message Function(Message original, Message updated) update = _mergeUpdate, + Message Function(Message original, Message updated) update = MessageMerging.mergeUpdate, bool upsert = true, }) { 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((it) => it.isShownInChannel); // If there are no affected messages, return early. if (affectedMessages.isEmpty) return; final channelMessages = [...this.messages]; - final updatedChannelMessages = _mergeMessagesIntoExisting( + final updatedChannelMessages = MessageMerging.mergeMessages( existing: channelMessages, toMerge: affectedMessages, update: update, @@ -4245,7 +4208,7 @@ class ChannelClientState { void _updatePinnedMessages( Iterable messages, { - Message Function(Message original, Message updated) update = _mergeUpdate, + Message Function(Message original, Message updated) update = MessageMerging.mergeUpdate, }) { if (messages.isEmpty) return; @@ -4254,7 +4217,7 @@ class ChannelClientState { // land right back on an empty `pinnedMessages` list. if (pinnedMessages.isEmpty && messages.every((m) => !m.pinned)) return; - final updatedPinnedMessages = _mergePinnedMessagesIntoExisting( + final updatedPinnedMessages = MessageMerging.mergePinnedMessages( existing: pinnedMessages, toMerge: messages, update: update, @@ -4269,7 +4232,7 @@ class ChannelClientState { if (messages.isEmpty) return; final activeLiveLocations = [...this.activeLiveLocations]; - final updatedActiveLiveLocations = _mergeActiveLocationsIntoExisting( + final updatedActiveLiveLocations = MessageMerging.mergeActiveLocations( existing: activeLiveLocations, toMerge: messages, ); @@ -4279,126 +4242,6 @@ class ChannelClientState { ); } - Iterable _mergeActiveLocationsIntoExisting({ - required Iterable existing, - required Iterable toMerge, - }) { - if (toMerge.isEmpty) return existing; - - final mergedLocations = existing.mergeFrom( - toMerge, - key: (it) => (it.userId, it.channelCid, it.createdByDeviceId), - value: (message) => message.sharedLocation, - update: (original, updated) => updated, - ); - - final toUpdateMap = {for (final m in toMerge) m.id: m}; - final updatedLocations = mergedLocations.where((it) { - // Remove the location if it's expired. - if (it.isExpired) return false; - - final updatedMessage = toUpdateMap[it.messageId]; - // Remove the location if the attached message is deleted. - if (updatedMessage?.isDeleted == true) return false; - - return true; - }); - - return updatedLocations; - } - - Iterable _mergePinnedMessagesIntoExisting({ - required Iterable existing, - required Iterable toMerge, - Message Function(Message original, Message updated) update = _mergeUpdate, - }) { - return _mergeMessagesIntoExisting( - existing: existing, - toMerge: toMerge, - update: update, - ).where(_pinIsValid); - } - - Iterable _mergeMessagesIntoExisting({ - required Iterable existing, - required Iterable toMerge, - Message Function(Message original, Message updated) update = _mergeUpdate, - bool upsert = true, - }) { - if (toMerge.isEmpty) return existing; - - // [update] decides whether each pair is reconciled (default — see - // `_mergeUpdate`) or replaced (`_replaceUpdate`, used by local rollback - // paths that don't want enrichment fallback to keep optimistic values). - // - // [upsert] controls whether ids not already in [existing] are inserted. - // Event-driven paths (`message.updated`, `message.deleted` soft) pass - // `upsert: false` so an out-of-window message isn't dropped into a gap - // between the loaded slice and history the client hasn't paged in yet. - final existingList = existing is List ? existing : existing.toList(); - var toMergeList = toMerge is List ? toMerge : toMerge.toList(); - - // Single-message fast path. The hot ingest path (server echoes, edits, - // reactions, read receipts) always lands here, and `lastIndexWhere` + - // `sortedUpsertAt` skips the O(N) keymap build that the two-pointer - // merge would otherwise do up front. - if (toMergeList.length == 1) { - final message = toMergeList.first; - final oldIndex = existingList.lastIndexWhere((it) => it.id == message.id); - - // upsert: false — skip update if message is not loaded - if (oldIndex == -1 && !upsert) return existingList; - - final resolved = oldIndex == -1 ? message : update(existingList[oldIndex], message); - - final mergedMessages = existingList.sortedUpsertAt( - oldIndex, - resolved, - update: update, - compare: _sortByCreatedAt, - ); - - // Non-delete updates can't change what embedded quotedMessage copies - // should display, so we can skip the rewrite entirely. - if (!resolved.isDeleted) return mergedMessages; - - return mergedMessages.updateIf( - (it) => it.quotedMessageId == resolved.id, - (it) => it.copyWith(quotedMessage: resolved), - ); - } - - // upsert: false - skip messages not loaded in the window - if (!upsert) { - final existingIds = {for (final m in existingList) m.id}; - toMergeList = toMergeList.where((m) => existingIds.contains(m.id)).toList(); - if (toMergeList.isEmpty) return existingList; - } - - // Batch path: receiver (`existingList`) is maintained sorted as a - // state invariant; `mergeSorted` sorts `toMergeList` internally and - // returns a sorted result. - final mergedMessages = existingList.mergeSorted( - toMergeList, - key: (message) => message.id, - update: update, - compare: _sortByCreatedAt, - ); - - // Refresh embedded `quotedMessage` refs only for messages quoting an - // incoming message that is now deleted. `updateIf` returns the same - // list reference when nothing matches, so steady-state allocates - // nothing for this step. - final deletedIds = toMergeList.where((m) => m.isDeleted).map((m) => m.id).toSet(); - if (deletedIds.isEmpty) return mergedMessages; - - final mergedById = {for (final m in mergedMessages) m.id: m}; - return mergedMessages.updateIf( - (it) => deletedIds.contains(it.quotedMessageId), - (it) => it.copyWith(quotedMessage: mergedById[it.quotedMessageId]), - ); - } - void _removeMessages(Iterable messages) { if (messages.isEmpty) return; @@ -4417,32 +4260,14 @@ class ChannelClientState { void _removeThreadMessages(Iterable messages) { if (messages.isEmpty) return; - final affectedThreads = {...messages.map((it) => it.parentId).nonNulls}; - // If there are no affected threads, return early. - if (affectedThreads.isEmpty) return; - - final updatedThreads = {...threads}; - for (final thread in affectedThreads) { - final threadMessages = updatedThreads[thread]; - // Continue if the thread doesn't exist. - if (threadMessages == null) continue; - - // Remove the deleted message from the thread messages and reference from - // other messages quoting it. - final updatedThreadMessages = _removeMessagesFromExisting( - existing: threadMessages, - toRemove: messages, - ); - - // If there are no more messages in the thread, remove the thread entry. - if (updatedThreadMessages.isEmpty) { - updatedThreads.remove(thread); - continue; - } + final currentThreads = threads; + final updatedThreads = MessageMerging.removeThreadMessages( + existing: currentThreads, + toRemove: messages, + ); - // Otherwise, update the thread with the modified message list. - updatedThreads[thread] = updatedThreadMessages.toList(); - } + // Nothing targeted a thread — skip the write. + if (identical(updatedThreads, currentThreads)) return; // Update the threads map. _threads = updatedThreads; @@ -4451,20 +4276,14 @@ 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((it) => it.isShownInChannel); // If there are no affected messages, return early. if (affectedMessages.isEmpty) return; final channelMessages = [...this.messages]; - final updatedChannelMessages = _removeMessagesFromExisting( + final updatedChannelMessages = MessageMerging.removeMessages( existing: channelMessages, toRemove: affectedMessages, ); @@ -4478,7 +4297,7 @@ class ChannelClientState { if (messages.isEmpty) return; final pinnedMessages = [...this.pinnedMessages]; - final updatedPinnedMessages = _removePinnedMessagesFromExisting( + final updatedPinnedMessages = MessageMerging.removePinnedMessages( existing: pinnedMessages, toRemove: messages, ); @@ -4492,7 +4311,7 @@ class ChannelClientState { if (messages.isEmpty) return; final activeLiveLocations = [...this.activeLiveLocations]; - final updatedActiveLiveLocations = _removeActiveLocationsFromExisting( + final updatedActiveLiveLocations = MessageMerging.removeActiveLocations( existing: activeLiveLocations, toRemove: messages, ); @@ -4502,54 +4321,6 @@ class ChannelClientState { ); } - Iterable _removeActiveLocationsFromExisting({ - required Iterable existing, - required Iterable toRemove, - }) { - if (toRemove.isEmpty) return existing; - - final toRemoveIds = toRemove.map((m) => m.id).toSet(); - final updatedLocations = existing.where( - // Remove the location if its attached message is in the toRemove list. - (it) => !toRemoveIds.contains(it.messageId), - ); - - return updatedLocations; - } - - Iterable _removePinnedMessagesFromExisting({ - required Iterable existing, - required Iterable toRemove, - }) { - return _removeMessagesFromExisting( - existing: existing, - toRemove: toRemove, - ).where(_pinIsValid); - } - - Iterable _removeMessagesFromExisting({ - required Iterable existing, - required Iterable toRemove, - }) { - if (toRemove.isEmpty) return existing; - - final toRemoveIds = toRemove.map((m) => m.id).toSet(); - final updatedMessages = existing - .where((it) { - // Remove the message if it's in the toRemove list. - return !toRemoveIds.contains(it.id); - }) - .map((it) { - // Continue if the message doesn't quote any of the deleted messages. - if (!toRemoveIds.contains(it.quotedMessageId)) return it; - - // Setting it to null will remove the quoted message from the message. - return it.copyWith(quotedMessageId: null, quotedMessage: null); - }); - - 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() { @@ -4583,21 +4354,6 @@ class ChannelClientState { } } -bool _pinIsValid(Message message) { - // If the message is deleted, the pin is not valid. - if (message.isDeleted) return false; - - // If the message is not pinned, it's not valid. - if (message.pinned != true) return false; - - // If there's no expiration, the pin is valid. - final pinExpires = message.pinExpires; - if (pinExpires == null) return true; - - // If there's an expiration, check if it's still valid. - return pinExpires.isAfter(DateTime.now()); -} - /// Extension methods for reading related operations on a ChannelClientState. extension ChannelReadHelper on ChannelClientState { /// Get the [Read] object for a specific user identified by [userId]. diff --git a/packages/stream_chat/lib/src/core/util/message_merging.dart b/packages/stream_chat/lib/src/core/util/message_merging.dart new file mode 100644 index 0000000000..9a06c5c3df --- /dev/null +++ b/packages/stream_chat/lib/src/core/util/message_merging.dart @@ -0,0 +1,302 @@ +import 'package:stream_chat/src/core/models/location.dart'; +import 'package:stream_chat/src/core/models/message.dart'; +import 'package:stream_chat/src/core/util/list_extensions.dart'; +import 'package:stream_chat/src/core/util/message_predicates.dart'; + +/// Provides the merge and removal operations reconciling incoming messages +/// with the locally-held channel state collections. +/// +/// Every operation is pure: it computes a new collection from the given +/// inputs without reading or writing any state. +class MessageMerging { + const MessageMerging._(); + + /// The default `update` strategy for the merge operations: merges the + /// incoming [updated] into the locally-known [original] via + /// [Message.updateWith], preserving enrichment the server may strip on + /// partial payloads. + static Message mergeUpdate(Message original, Message updated) => original.updateWith(updated); + + /// The replacing `update` strategy: takes the incoming [updated] as-is. + /// Used by local rollback paths. + static Message replaceUpdate(Message _, Message updated) => updated; + + /// Compares [a] and [b] by their [Message.createdAt]. + static int sortByCreatedAt(Message a, Message b) => a.createdAt.compareTo(b.createdAt); + + /// Merges the live locations carried by [toMerge] into [existing]. + /// + /// Locations are keyed by (userId, channelCid, createdByDeviceId); an + /// incoming message's [Message.sharedLocation] replaces the existing entry + /// with the same key. Locations that are expired or whose attached message + /// is deleted are dropped from the result. + static Iterable mergeActiveLocations({ + required Iterable existing, + required Iterable toMerge, + }) { + if (toMerge.isEmpty) return existing; + + final mergedLocations = existing.mergeFrom( + toMerge, + key: (it) => (it.userId, it.channelCid, it.createdByDeviceId), + value: (message) => message.sharedLocation, + update: (original, updated) => updated, + ); + + final toUpdateMap = {for (final m in toMerge) m.id: m}; + final updatedLocations = mergedLocations.where((it) { + // Remove the location if it's expired. + if (it.isExpired) return false; + + final updatedMessage = toUpdateMap[it.messageId]; + // Remove the location if the attached message is deleted. + if (updatedMessage?.isDeleted == true) return false; + + return true; + }); + + return updatedLocations; + } + + /// Merges [toMerge] into the [existing] pinned messages, keeping only + /// messages that are still valid pins (see [MessagePredicates.hasValidPin]). + static Iterable mergePinnedMessages({ + required Iterable existing, + required Iterable toMerge, + Message Function(Message original, Message updated) update = mergeUpdate, + }) { + return mergeMessages( + existing: existing, + toMerge: toMerge, + update: update, + ).where((it) => it.hasValidPin); + } + + /// Merges [toMerge] into [existing], returning a list sorted by + /// [Message.createdAt]. + /// + /// [update] decides whether each pair is reconciled (default — see + /// [mergeUpdate]) or replaced ([replaceUpdate], used by local rollback + /// paths that don't want enrichment fallback to keep optimistic values). + /// + /// [upsert] controls whether ids not already in [existing] are inserted. + /// Event-driven paths (`message.updated`, `message.deleted` soft) pass + /// `upsert: false` so an out-of-window message isn't dropped into a gap + /// between the loaded slice and history the client hasn't paged in yet. + static Iterable mergeMessages({ + required Iterable existing, + required Iterable toMerge, + Message Function(Message original, Message updated) update = mergeUpdate, + bool upsert = true, + }) { + if (toMerge.isEmpty) return existing; + + final existingList = existing is List ? existing : existing.toList(); + var toMergeList = toMerge is List ? toMerge : toMerge.toList(); + + // Single-message fast path. The hot ingest path (server echoes, edits, + // reactions, read receipts) always lands here, and `lastIndexWhere` + + // `sortedUpsertAt` skips the O(N) keymap build that the two-pointer + // merge would otherwise do up front. + if (toMergeList.length == 1) { + final message = toMergeList.first; + final oldIndex = existingList.lastIndexWhere((it) => it.id == message.id); + + // upsert: false — skip update if message is not loaded + if (oldIndex == -1 && !upsert) return existingList; + + final resolved = oldIndex == -1 ? message : update(existingList[oldIndex], message); + + final mergedMessages = existingList.sortedUpsertAt( + oldIndex, + resolved, + update: update, + compare: sortByCreatedAt, + ); + + // Non-delete updates can't change what embedded quotedMessage copies + // should display, so we can skip the rewrite entirely. + if (!resolved.isDeleted) return mergedMessages; + + return mergedMessages.updateIf( + (it) => it.quotedMessageId == resolved.id, + (it) => it.copyWith(quotedMessage: resolved), + ); + } + + // upsert: false - skip messages not loaded in the window + if (!upsert) { + final existingIds = {for (final m in existingList) m.id}; + toMergeList = toMergeList.where((m) => existingIds.contains(m.id)).toList(); + if (toMergeList.isEmpty) return existingList; + } + + // Batch path: receiver (`existingList`) is maintained sorted as a + // state invariant; `mergeSorted` sorts `toMergeList` internally and + // returns a sorted result. + final mergedMessages = existingList.mergeSorted( + toMergeList, + key: (message) => message.id, + update: update, + compare: sortByCreatedAt, + ); + + // Refresh embedded `quotedMessage` refs only for messages quoting an + // incoming message that is now deleted. `updateIf` returns the same + // list reference when nothing matches, so steady-state allocates + // nothing for this step. + final deletedIds = toMergeList.where((m) => m.isDeleted).map((m) => m.id).toSet(); + if (deletedIds.isEmpty) return mergedMessages; + + final mergedById = {for (final m in mergedMessages) m.id: m}; + return mergedMessages.updateIf( + (it) => deletedIds.contains(it.quotedMessageId), + (it) => it.copyWith(quotedMessage: mergedById[it.quotedMessageId]), + ); + } + + /// Merges [toMerge] into the [existing] threads map, returning the updated + /// map, or [existing] as-is when [toMerge] carries no thread replies. + /// + /// Replies are grouped by their parent id so each thread merge only sees + /// its own messages. With [upsert] `false`, replies to threads not present + /// in [existing] are dropped instead of creating the thread entry. + static Map> mergeThreadMessages({ + required Map> existing, + required Iterable toMerge, + Message Function(Message original, Message updated) update = mergeUpdate, + bool upsert = true, + }) { + if (toMerge.isEmpty) return existing; + + // Group messages by parentId so each thread merge only sees its own + // replies — passing the full batch to every thread would leak replies + // across thread boundaries (the merge dedups by id, not by parentId). + final messagesByThread = >{}; + for (final m in toMerge) { + if (m.parentId case final parentId?) (messagesByThread[parentId] ??= []).add(m); + } + + // If there are no affected threads, return early. + if (messagesByThread.isEmpty) return existing; + + final updatedThreads = {...existing}; + for (final MapEntry(key: thread, :value) in messagesByThread.entries) { + final existingThreadMessages = updatedThreads[thread]; + + // Don't create a phantom entry for a thread that wasn't loaded: with + // `upsert: false` an out-of-window reply is dropped, so there's nothing + // to merge. Writing it back would make `threads.containsKey(parentId)` + // report a thread that was never paged in. + if (existingThreadMessages == null && !upsert) continue; + + final threadMessages = existingThreadMessages ?? []; + final updatedThreadMessages = mergeMessages( + existing: threadMessages, + toMerge: value, + update: update, + upsert: upsert, + ); + + // Update the thread with the modified message list. + updatedThreads[thread] = updatedThreadMessages.toList(); + } + + return updatedThreads; + } + + /// Removes from [existing] the locations attached to any message in + /// [toRemove]. + static Iterable removeActiveLocations({ + required Iterable existing, + required Iterable toRemove, + }) { + if (toRemove.isEmpty) return existing; + + final toRemoveIds = toRemove.map((m) => m.id).toSet(); + final updatedLocations = existing.where( + // Remove the location if its attached message is in the toRemove list. + (it) => !toRemoveIds.contains(it.messageId), + ); + + return updatedLocations; + } + + /// Removes [toRemove] from the [existing] pinned messages, keeping only + /// messages that are still valid pins (see [MessagePredicates.hasValidPin]). + static Iterable removePinnedMessages({ + required Iterable existing, + required Iterable toRemove, + }) { + return removeMessages( + existing: existing, + toRemove: toRemove, + ).where((it) => it.hasValidPin); + } + + /// Removes [toRemove] from [existing], clearing the quoted-message + /// reference of any remaining message that quotes a removed one. + static Iterable removeMessages({ + required Iterable existing, + required Iterable toRemove, + }) { + if (toRemove.isEmpty) return existing; + + final toRemoveIds = toRemove.map((m) => m.id).toSet(); + final updatedMessages = existing + .where((it) { + // Remove the message if it's in the toRemove list. + return !toRemoveIds.contains(it.id); + }) + .map((it) { + // Continue if the message doesn't quote any of the deleted messages. + if (!toRemoveIds.contains(it.quotedMessageId)) return it; + + // Setting it to null will remove the quoted message from the message. + return it.copyWith(quotedMessageId: null, quotedMessage: null); + }); + + return updatedMessages; + } + + /// Removes [toRemove] from the [existing] threads map, returning the + /// updated map, or [existing] as-is when [toRemove] carries no thread + /// replies. + /// + /// Thread entries left with no messages are dropped from the map. + static Map> removeThreadMessages({ + required Map> existing, + required Iterable toRemove, + }) { + if (toRemove.isEmpty) return existing; + + final affectedThreads = {...toRemove.map((it) => it.parentId).nonNulls}; + // If there are no affected threads, return early. + if (affectedThreads.isEmpty) return existing; + + final updatedThreads = {...existing}; + for (final thread in affectedThreads) { + final threadMessages = updatedThreads[thread]; + // Continue if the thread doesn't exist. + if (threadMessages == null) continue; + + // Remove the deleted message from the thread messages and reference from + // other messages quoting it. + final updatedThreadMessages = removeMessages( + existing: threadMessages, + toRemove: toRemove, + ); + + // If there are no more messages in the thread, remove the thread entry. + if (updatedThreadMessages.isEmpty) { + updatedThreads.remove(thread); + continue; + } + + // Otherwise, update the thread with the modified message list. + updatedThreads[thread] = updatedThreadMessages.toList(); + } + + return updatedThreads; + } +} diff --git a/packages/stream_chat/lib/src/core/util/message_predicates.dart b/packages/stream_chat/lib/src/core/util/message_predicates.dart new file mode 100644 index 0000000000..5bb7d3c1f2 --- /dev/null +++ b/packages/stream_chat/lib/src/core/util/message_predicates.dart @@ -0,0 +1,35 @@ +import 'package:stream_chat/src/core/models/message.dart'; + +/// Predicates over a [Message] used by the channel state management. +extension MessagePredicates on Message { + /// Whether the message is shown in the channel message list. + /// + /// Non-thread messages always are; thread replies only when explicitly + /// marked to also show in the channel. + bool get isShownInChannel { + // Non-thread messages are always shown in the channel. + if (parentId == null) return true; + + // Thread messages are only shown if explicitly marked. + return showInChannel == true; + } + + /// Whether the message represents a currently valid pin. + /// + /// Returns `false` if the message is deleted, not pinned, or its + /// [Message.pinExpires] has passed. + bool get hasValidPin { + // If the message is deleted, the pin is not valid. + if (isDeleted) return false; + + // If the message is not pinned, it's not valid. + if (pinned != true) return false; + + // If there's no expiration, the pin is valid. + final expiresAt = pinExpires; + if (expiresAt == null) return true; + + // If there's an expiration, check if it's still valid. + return expiresAt.isAfter(DateTime.now()); + } +} diff --git a/packages/stream_chat/test/src/core/util/message_merging_test.dart b/packages/stream_chat/test/src/core/util/message_merging_test.dart new file mode 100644 index 0000000000..24839c5467 --- /dev/null +++ b/packages/stream_chat/test/src/core/util/message_merging_test.dart @@ -0,0 +1,489 @@ +import 'package:stream_chat/src/core/util/message_merging.dart'; +import 'package:stream_chat/stream_chat.dart'; +import 'package:test/test.dart'; + +Message _message( + String id, { + DateTime? createdAt, + String? text, + String? parentId, + bool? showInChannel, + String? quotedMessageId, + Message? quotedMessage, + bool pinned = false, + DateTime? pinExpires, + String type = 'regular', + Location? sharedLocation, +}) { + return Message( + id: id, + createdAt: createdAt ?? DateTime(2024), + text: text, + parentId: parentId, + showInChannel: showInChannel, + quotedMessageId: quotedMessageId, + quotedMessage: quotedMessage, + pinned: pinned, + pinExpires: pinExpires, + type: type, + sharedLocation: sharedLocation, + ); +} + +Location _sharedLocation({ + String? messageId, + String? userId = 'user-id', + String? channelCid = 'messaging:channel-id', + String? createdByDeviceId = 'device-id', + DateTime? endAt, + double latitude = 0, + double longitude = 0, +}) { + return Location( + messageId: messageId, + userId: userId, + channelCid: channelCid, + createdByDeviceId: createdByDeviceId, + endAt: endAt, + latitude: latitude, + longitude: longitude, + ); +} + +Iterable _ids(Iterable messages) => messages.map((it) => it.id); + +void main() { + group('MessageMerging.mergeUpdate', () { + test('merges the incoming message into the original, preserving enrichment', () { + final location = _sharedLocation(messageId: 'm1', endAt: DateTime.now().add(const Duration(hours: 1))); + final original = _message('m1', text: 'old', sharedLocation: location); + final updated = _message('m1', text: 'new'); + + final result = MessageMerging.mergeUpdate(original, updated); + + expect(result.text, 'new'); + expect(result.sharedLocation, location); + }); + }); + + group('MessageMerging.replaceUpdate', () { + test('takes the incoming message as-is, dropping enrichment', () { + final location = _sharedLocation(messageId: 'm1', endAt: DateTime.now().add(const Duration(hours: 1))); + final original = _message('m1', text: 'old', sharedLocation: location); + final updated = _message('m1', text: 'new'); + + final result = MessageMerging.replaceUpdate(original, updated); + + expect(result, same(updated)); + expect(result.sharedLocation, isNull); + }); + }); + + group('MessageMerging.sortByCreatedAt', () { + test('orders messages by their creation time', () { + final earlier = _message('m1', createdAt: DateTime(2024)); + final later = _message('m2', createdAt: DateTime(2024, 2)); + + expect(MessageMerging.sortByCreatedAt(earlier, later), isNegative); + expect(MessageMerging.sortByCreatedAt(later, earlier), isPositive); + expect(MessageMerging.sortByCreatedAt(earlier, earlier), isZero); + }); + }); + + group('MessageMerging.mergeMessages', () { + test('returns the existing messages untouched when there is nothing to merge', () { + final existing = [_message('m1')]; + + final result = MessageMerging.mergeMessages(existing: existing, toMerge: const []); + + expect(result, same(existing)); + }); + + test('inserts a single unknown message in sorted position', () { + final existing = [ + _message('m1', createdAt: DateTime(2024)), + _message('m3', createdAt: DateTime(2024, 3)), + ]; + final incoming = _message('m2', createdAt: DateTime(2024, 2)); + + final result = MessageMerging.mergeMessages(existing: existing, toMerge: [incoming]); + + expect(_ids(result), ['m1', 'm2', 'm3']); + }); + + test('updates a single existing message via the default merge strategy', () { + final location = _sharedLocation(messageId: 'm1', endAt: DateTime.now().add(const Duration(hours: 1))); + final existing = [_message('m1', text: 'old', sharedLocation: location)]; + final incoming = _message('m1', text: 'new'); + + final result = MessageMerging.mergeMessages(existing: existing, toMerge: [incoming]); + + expect(result, hasLength(1)); + expect(result.first.text, 'new'); + expect(result.first.sharedLocation, location, reason: 'enrichment should survive a stripped payload'); + }); + + test('upsert: false skips a single message that is not loaded', () { + final existing = [_message('m1')]; + final incoming = _message('m2'); + + final result = MessageMerging.mergeMessages(existing: existing, toMerge: [incoming], upsert: false); + + expect(result, same(existing)); + }); + + test('rewrites the embedded quote on quoters when the incoming message is deleted', () { + final quoted = _message('m1', createdAt: DateTime(2024), text: 'quoted'); + final quoter = _message( + 'm2', + createdAt: DateTime(2024, 2), + quotedMessageId: 'm1', + quotedMessage: quoted, + ); + final deleted = _message('m1', createdAt: DateTime(2024), type: MessageType.deleted); + + final result = MessageMerging.mergeMessages(existing: [quoted, quoter], toMerge: [deleted]); + + final updatedQuoter = result.singleWhere((it) => it.id == 'm2'); + expect(updatedQuoter.quotedMessage?.isDeleted, isTrue); + }); + + test('does not rewrite the embedded quote on quoters for a non-delete update', () { + final quoted = _message('m1', createdAt: DateTime(2024), text: 'quoted'); + final quoter = _message( + 'm2', + createdAt: DateTime(2024, 2), + quotedMessageId: 'm1', + quotedMessage: quoted, + ); + final edited = _message('m1', createdAt: DateTime(2024), text: 'edited'); + + final result = MessageMerging.mergeMessages(existing: [quoted, quoter], toMerge: [edited]); + + final updatedQuoter = result.singleWhere((it) => it.id == 'm2'); + expect(updatedQuoter.quotedMessage?.text, 'quoted'); + }); + + test('interleaves a batch of unknown messages in sorted order', () { + final existing = [ + _message('m1', createdAt: DateTime(2024)), + _message('m3', createdAt: DateTime(2024, 3)), + ]; + final incoming = [ + _message('m4', createdAt: DateTime(2024, 4)), + _message('m2', createdAt: DateTime(2024, 2)), + ]; + + final result = MessageMerging.mergeMessages(existing: existing, toMerge: incoming); + + expect(_ids(result), ['m1', 'm2', 'm3', 'm4']); + }); + + test('upsert: false only applies the batch entries that are already loaded', () { + final existing = [_message('m1', text: 'old')]; + final incoming = [ + _message('m1', text: 'new'), + _message('m2'), + _message('m3'), + ]; + + final result = MessageMerging.mergeMessages(existing: existing, toMerge: incoming, upsert: false); + + expect(_ids(result), ['m1']); + expect(result.first.text, 'new'); + }); + + test('honors a replacing update strategy for a batch', () { + final location = _sharedLocation(messageId: 'm1', endAt: DateTime.now().add(const Duration(hours: 1))); + final existing = [ + _message('m1', text: 'old', sharedLocation: location), + _message('m2', createdAt: DateTime(2024, 2)), + ]; + final incoming = [ + _message('m1', text: 'new'), + _message('m2', createdAt: DateTime(2024, 2)), + ]; + + final result = MessageMerging.mergeMessages( + existing: existing, + toMerge: incoming, + update: MessageMerging.replaceUpdate, + ); + + final replaced = result.singleWhere((it) => it.id == 'm1'); + expect(replaced.text, 'new'); + expect(replaced.sharedLocation, isNull); + }); + + test('rewrites the embedded quote on quoters when a batch entry is deleted', () { + final quoted = _message('m1', createdAt: DateTime(2024), text: 'quoted'); + final quoter = _message( + 'm3', + createdAt: DateTime(2024, 3), + quotedMessageId: 'm1', + quotedMessage: quoted, + ); + final incoming = [ + _message('m1', createdAt: DateTime(2024), type: MessageType.deleted), + _message('m2', createdAt: DateTime(2024, 2)), + ]; + + final result = MessageMerging.mergeMessages(existing: [quoted, quoter], toMerge: incoming); + + final updatedQuoter = result.singleWhere((it) => it.id == 'm3'); + expect(updatedQuoter.quotedMessage?.isDeleted, isTrue); + }); + }); + + group('MessageMerging.mergeThreadMessages', () { + test('returns the existing threads untouched when nothing targets a thread', () { + final existing = { + 'p1': [_message('m1', parentId: 'p1')], + }; + + final result = MessageMerging.mergeThreadMessages(existing: existing, toMerge: [_message('m2')]); + + expect(result, same(existing)); + }); + + test('groups replies into their own threads', () { + final existing = { + 'p1': [_message('m1', parentId: 'p1', createdAt: DateTime(2024))], + 'p2': [_message('m2', parentId: 'p2', createdAt: DateTime(2024))], + }; + final incoming = [ + _message('m3', parentId: 'p1', createdAt: DateTime(2024, 2)), + _message('m4', parentId: 'p2', createdAt: DateTime(2024, 2)), + ]; + + final result = MessageMerging.mergeThreadMessages(existing: existing, toMerge: incoming); + + expect(_ids(result['p1']!), ['m1', 'm3']); + expect(_ids(result['p2']!), ['m2', 'm4']); + }); + + test('creates the thread entry for a reply to a new thread', () { + final result = MessageMerging.mergeThreadMessages( + existing: const {}, + toMerge: [_message('m1', parentId: 'p1')], + ); + + expect(_ids(result['p1']!), ['m1']); + }); + + test('upsert: false does not create an entry for a thread that was never loaded', () { + final result = MessageMerging.mergeThreadMessages( + existing: const {}, + toMerge: [_message('m1', parentId: 'p1')], + upsert: false, + ); + + expect(result, isEmpty); + }); + + test('upsert: false still updates a reply in a loaded thread', () { + final existing = { + 'p1': [_message('m1', parentId: 'p1', text: 'old')], + }; + + final result = MessageMerging.mergeThreadMessages( + existing: existing, + toMerge: [_message('m1', parentId: 'p1', text: 'new')], + upsert: false, + ); + + expect(result['p1']!.single.text, 'new'); + }); + }); + + group('MessageMerging.mergePinnedMessages', () { + test('keeps only messages that are still valid pins', () { + final existing = [_message('m1', createdAt: DateTime(2024), pinned: true)]; + final incoming = [ + _message('m2', createdAt: DateTime(2024, 2), pinned: true), + _message('m3', createdAt: DateTime(2024, 3)), + _message( + 'm4', + createdAt: DateTime(2024, 4), + pinned: true, + pinExpires: DateTime.now().subtract(const Duration(hours: 1)), + ), + _message('m5', createdAt: DateTime(2024, 5), pinned: true, type: MessageType.deleted), + ]; + + final result = MessageMerging.mergePinnedMessages(existing: existing, toMerge: incoming); + + expect(_ids(result), ['m1', 'm2']); + }); + + test('drops an existing pin that the incoming message unpins', () { + final existing = [_message('m1', pinned: true)]; + final incoming = [_message('m1')]; + + final result = MessageMerging.mergePinnedMessages(existing: existing, toMerge: incoming); + + expect(result, isEmpty); + }); + }); + + group('MessageMerging.mergeActiveLocations', () { + test('replaces the existing location sharing the same key', () { + final endAt = DateTime.now().add(const Duration(hours: 1)); + final existing = [_sharedLocation(messageId: 'm1', endAt: endAt, latitude: 1)]; + final incoming = _message( + 'm1', + sharedLocation: _sharedLocation(messageId: 'm1', endAt: endAt, latitude: 2), + ); + + final result = MessageMerging.mergeActiveLocations(existing: existing, toMerge: [incoming]); + + expect(result, hasLength(1)); + expect(result.first.latitude, 2); + }); + + test('drops expired locations and ignores messages without a live location', () { + final expired = _sharedLocation( + messageId: 'm1', + endAt: DateTime.now().subtract(const Duration(hours: 1)), + ); + final incoming = _message('m2'); + + final result = MessageMerging.mergeActiveLocations(existing: [expired], toMerge: [incoming]); + + expect(result, isEmpty); + }); + + test('drops the location when its attached message is deleted', () { + final active = _sharedLocation(messageId: 'm1', endAt: DateTime.now().add(const Duration(hours: 1))); + final incoming = _message('m1', type: MessageType.deleted); + + final result = MessageMerging.mergeActiveLocations(existing: [active], toMerge: [incoming]); + + expect(result, isEmpty); + }); + }); + + group('MessageMerging.removeMessages', () { + test('returns the existing messages untouched when there is nothing to remove', () { + final existing = [_message('m1')]; + + final result = MessageMerging.removeMessages(existing: existing, toRemove: const []); + + expect(result, same(existing)); + }); + + test('removes the given messages by id', () { + final existing = [_message('m1'), _message('m2')]; + + final result = MessageMerging.removeMessages(existing: existing, toRemove: [_message('m1')]); + + expect(_ids(result), ['m2']); + }); + + test('clears the quoted-message reference of quoters of a removed message', () { + final quoted = _message('m1'); + final quoter = _message('m2', quotedMessageId: 'm1', quotedMessage: quoted); + + final result = MessageMerging.removeMessages(existing: [quoted, quoter], toRemove: [quoted]); + + final updatedQuoter = result.single; + expect(updatedQuoter.id, 'm2'); + expect(updatedQuoter.quotedMessageId, isNull); + expect(updatedQuoter.quotedMessage, isNull); + }); + }); + + group('MessageMerging.removeThreadMessages', () { + test('returns the existing threads untouched when nothing targets a thread', () { + final existing = { + 'p1': [_message('m1', parentId: 'p1')], + }; + + final result = MessageMerging.removeThreadMessages(existing: existing, toRemove: [_message('m2')]); + + expect(result, same(existing)); + }); + + test('removes the reply from its thread', () { + final existing = { + 'p1': [ + _message('m1', parentId: 'p1'), + _message('m2', parentId: 'p1'), + ], + }; + + final result = MessageMerging.removeThreadMessages( + existing: existing, + toRemove: [_message('m1', parentId: 'p1')], + ); + + expect(_ids(result['p1']!), ['m2']); + }); + + test('drops the thread entry when its last reply is removed', () { + final existing = { + 'p1': [_message('m1', parentId: 'p1')], + }; + + final result = MessageMerging.removeThreadMessages( + existing: existing, + toRemove: [_message('m1', parentId: 'p1')], + ); + + expect(result, isEmpty); + }); + + test('ignores replies to threads that are not loaded', () { + final existing = { + 'p1': [_message('m1', parentId: 'p1')], + }; + + final result = MessageMerging.removeThreadMessages( + existing: existing, + toRemove: [_message('m2', parentId: 'p2')], + ); + + expect(_ids(result['p1']!), ['m1']); + }); + }); + + group('MessageMerging.removePinnedMessages', () { + test('removes the given messages and filters out pins that are no longer valid', () { + final existing = [ + _message('m1', pinned: true), + _message( + 'm2', + pinned: true, + pinExpires: DateTime.now().subtract(const Duration(hours: 1)), + ), + _message('m3', pinned: true), + ]; + + final result = MessageMerging.removePinnedMessages(existing: existing, toRemove: [_message('m1')]); + + expect(_ids(result), ['m3']); + }); + }); + + group('MessageMerging.removeActiveLocations', () { + test('returns the existing locations untouched when there is nothing to remove', () { + final existing = [_sharedLocation(messageId: 'm1')]; + + final result = MessageMerging.removeActiveLocations(existing: existing, toRemove: const []); + + expect(result, same(existing)); + }); + + test('removes the locations attached to the removed messages', () { + final existing = [ + _sharedLocation(messageId: 'm1'), + _sharedLocation(messageId: 'm2'), + ]; + + final result = MessageMerging.removeActiveLocations(existing: existing, toRemove: [_message('m1')]); + + expect(result, hasLength(1)); + expect(result.first.messageId, 'm2'); + }); + }); +} diff --git a/packages/stream_chat/test/src/core/util/message_predicates_test.dart b/packages/stream_chat/test/src/core/util/message_predicates_test.dart new file mode 100644 index 0000000000..da3340a4eb --- /dev/null +++ b/packages/stream_chat/test/src/core/util/message_predicates_test.dart @@ -0,0 +1,66 @@ +import 'package:stream_chat/src/core/util/message_predicates.dart'; +import 'package:stream_chat/stream_chat.dart'; +import 'package:test/test.dart'; + +Message _message( + String id, { + String? parentId, + bool? showInChannel, + bool pinned = false, + DateTime? pinExpires, + String type = 'regular', +}) { + return Message( + id: id, + parentId: parentId, + showInChannel: showInChannel, + pinned: pinned, + pinExpires: pinExpires, + type: type, + ); +} + +void main() { + group('MessagePredicates.isShownInChannel', () { + test('is true for a non-thread message', () { + expect(_message('m1').isShownInChannel, isTrue); + }); + + test('is true for a thread reply marked to show in the channel', () { + final reply = _message('m1', parentId: 'p1', showInChannel: true); + expect(reply.isShownInChannel, isTrue); + }); + + test('is false for a thread-only reply', () { + final reply = _message('m1', parentId: 'p1'); + expect(reply.isShownInChannel, isFalse); + }); + }); + + group('MessagePredicates.hasValidPin', () { + test('is false for a deleted message', () { + final message = _message('m1', pinned: true, type: MessageType.deleted); + expect(message.hasValidPin, isFalse); + }); + + test('is false for an unpinned message', () { + final message = _message('m1'); + expect(message.hasValidPin, isFalse); + }); + + test('is true for a pinned message without expiration', () { + final message = _message('m1', pinned: true); + expect(message.hasValidPin, isTrue); + }); + + test('is true while the pin expiration is in the future', () { + final message = _message('m1', pinned: true, pinExpires: DateTime.now().add(const Duration(hours: 1))); + expect(message.hasValidPin, isTrue); + }); + + test('is false once the pin expiration has passed', () { + final message = _message('m1', pinned: true, pinExpires: DateTime.now().subtract(const Duration(hours: 1))); + expect(message.hasValidPin, isFalse); + }); + }); +}