diff --git a/packages/stream_chat/CHANGELOG.md b/packages/stream_chat/CHANGELOG.md index 09b5f0cf2..66a05047a 100644 --- a/packages/stream_chat/CHANGELOG.md +++ b/packages/stream_chat/CHANGELOG.md @@ -1,5 +1,9 @@ ## Upcoming +✅ Added + +- Added an `upsert` flag to `ChannelClientState.updateMessage` (defaults to `true`). Pass `false` to update a message only if it's already loaded in the state, skipping unknown messages instead of adding them. + 🔄 Changed - `StreamChatClient.updateSystemEnvironment` now sanitizes the passed `SystemEnvironment`: `sdkName`, `sdkVersion`, and `osName` are locked to internal defaults, and `sdkIdentifier` only accepts the `dart` → `flutter` promotion (other values, including a `flutter` → `dart` demotion, are ignored). `appName`, `appVersion`, `osVersion`, and `deviceModel` continue to pass through as-is. @@ -7,7 +11,7 @@ 🐞 Fixed - `StreamChatNetworkError.fromDioException` no longer throws `FormatException` when an edge/proxy returns a non-JSON body (e.g. a plain `upstream request timeout` from a 504); the original network error now surfaces with `statusCode` / `statusMessage` intact. - +- Fixed `message.updated` and soft `message.deleted` events being incorrectly upserted into `ChannelState.messages` (and thread reply lists) when they targeted a message outside the currently loaded window. ## 10.1.0 diff --git a/packages/stream_chat/lib/src/client/channel.dart b/packages/stream_chat/lib/src/client/channel.dart index 1432600a8..d921f5cf2 100644 --- a/packages/stream_chat/lib/src/client/channel.dart +++ b/packages/stream_chat/lib/src/client/channel.dart @@ -3194,7 +3194,7 @@ class ChannelClientState { final message = event.message; if (message == null) return; - return updateMessage(message); + return updateMessage(message, upsert: false); }), ); } @@ -3209,7 +3209,9 @@ class ChannelClientState { deletedForMe: event.deletedForMe, ); - return deleteMessage(message, hardDelete: hardDelete); + if (hardDelete) return deleteMessage(message, hardDelete: true); + // Soft delete, update the message only if loaded (upsert: false) + return updateMessage(message, upsert: false); }), ); } @@ -3305,13 +3307,18 @@ class ChannelClientState { ); } - /// Updates the [message] in the state if it exists. Adds it otherwise. + /// Updates the [message] in the state. /// /// Reconciles via `Message.updateWith`, so locally-known enrichment /// (poll, sharedLocation, ownReactions, nested quotedMessage) is /// preserved when [message] omits those fields. Use [replaceMessage] /// for paths that need a strict overwrite. - void updateMessage(Message message) => _updateMessages([message]); + /// + /// When [upsert] is `true` (the default) and [message] isn't already in + /// the state, it's added. When `false`, an unknown [message] is skipped + /// and the state is left unchanged; only a message already loaded in the + /// state is updated. + void updateMessage(Message message, {bool upsert = true}) => _updateMessages([message], upsert: upsert); /// Replaces the [message] in the state if it exists, no-op otherwise. /// @@ -3930,11 +3937,12 @@ class ChannelClientState { void _updateMessages( Iterable messages, { Message Function(Message original, Message updated) update = _mergeUpdate, + bool upsert = true, }) { if (messages.isEmpty) return; - _updateThreadMessages(messages, update: update); - _updateChannelMessages(messages, update: update); + _updateThreadMessages(messages, update: update, upsert: upsert); + _updateChannelMessages(messages, update: update, upsert: upsert); _updatePinnedMessages(messages, update: update); _updateActiveLiveLocations(messages); } @@ -3942,6 +3950,7 @@ class ChannelClientState { void _updateThreadMessages( Iterable messages, { Message Function(Message original, Message updated) update = _mergeUpdate, + bool upsert = true, }) { if (messages.isEmpty) return; @@ -3958,13 +3967,23 @@ class ChannelClientState { final updatedThreads = {...threads}; for (final MapEntry(key: thread, :value) in messagesByThread.entries) { - final threadMessages = updatedThreads[thread] ?? []; + final existingThreadMessages = updatedThreads[thread]; + final threadMessages = existingThreadMessages ?? []; final updatedThreadMessages = _mergeMessagesIntoExisting( existing: threadMessages, toMerge: value, update: update, + upsert: upsert, ); + // Don't create a phantom entry for a thread that wasn't loaded: with + // `upsert: false` an out-of-window reply is dropped, leaving the merged + // list empty. Writing it back would make `threads.containsKey(parentId)` + // report a thread that was never paged in. + if (existingThreadMessages == null && updatedThreadMessages.isEmpty) { + continue; + } + // Update the thread with the modified message list. updatedThreads[thread] = updatedThreadMessages.toList(); } @@ -3976,6 +3995,7 @@ class ChannelClientState { void _updateChannelMessages( Iterable messages, { Message Function(Message original, Message updated) update = _mergeUpdate, + bool upsert = true, }) { if (messages.isEmpty) return; @@ -3996,6 +4016,7 @@ class ChannelClientState { existing: channelMessages, toMerge: affectedMessages, update: update, + upsert: upsert, ); // Calculate the new last message at time. @@ -4092,14 +4113,20 @@ class ChannelClientState { 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(); - final toMergeList = toMerge is List ? toMerge : toMerge.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` + @@ -4108,6 +4135,10 @@ class ChannelClientState { 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( @@ -4127,6 +4158,13 @@ class ChannelClientState { ); } + // 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. diff --git a/packages/stream_chat/test/src/client/channel_test.dart b/packages/stream_chat/test/src/client/channel_test.dart index 4a9b1ef09..4e363b24f 100644 --- a/packages/stream_chat/test/src/client/channel_test.dart +++ b/packages/stream_chat/test/src/client/channel_test.dart @@ -1854,6 +1854,60 @@ void main() { }); }); + group('`ChannelClientState.updateMessage`', () { + test('upsert: true (default) adds an unknown message', () async { + final message = Message( + id: 'unknown-message', + user: client.state.currentUser, + text: 'hello', + createdAt: DateTime.utc(2026), + ); + + expect(channel.state!.messages, isEmpty); + + channel.state!.updateMessage(message); + + expect(channel.state!.messages.map((m) => m.id), ['unknown-message']); + }); + + test('upsert: false does NOT add an unknown message', () async { + final message = Message( + id: 'unknown-message', + user: client.state.currentUser, + text: 'hello', + createdAt: DateTime.utc(2026), + ); + + expect(channel.state!.messages, isEmpty); + + channel.state!.updateMessage(message, upsert: false); + + expect(channel.state!.messages, isEmpty); + }); + + test('upsert: false updates a message already in the window', () async { + const messageId = 'known-message'; + final seeded = Message( + id: messageId, + user: client.state.currentUser, + text: 'old', + createdAt: DateTime.utc(2026), + ); + channel.state!.updateChannelState( + channel.state!.channelState.copyWith(messages: [seeded]), + ); + + channel.state!.updateMessage( + seeded.copyWith(text: 'new'), + upsert: false, + ); + + final stored = channel.state!.messages.single; + expect(stored.id, equals(messageId)); + expect(stored.text, equals('new')); + }); + }); + test('`.partialUpdateMessage`', () async { final message = Message( id: 'test-message-id', @@ -5465,6 +5519,191 @@ void main() { expect(channel.state?.pinnedMessages, isEmpty); }, ); + + // A `message.updated` event for a message outside the loaded window + // would otherwise upsert into the sorted list — creating a phantom + // entry with a gap. The guard is "id not in the loaded list", and + // is independent of `isUpToDate` — even at the latest page we may + // have paginated past older history and receive an event for a + // message no longer in memory. + group('when message is outside the loaded window', () { + test( + 'should NOT insert unknown message into `messages` list', + () async { + // Simulate "we have the latest page but not older history": + // seed the tail messages. + final tail = List.generate( + 3, + (i) => Message( + id: 'tail-$i', + user: client.state.currentUser, + text: 'tail $i', + createdAt: DateTime.utc(2026, 6, 1).add(Duration(seconds: i)), + ), + ); + channel.state!.updateChannelState( + channel.state!.channelState.copyWith(messages: tail), + ); + expect(channel.state!.messages, hasLength(3)); + + // Event for a message on an older page we don't have loaded. + final olderPageEdit = Message( + id: 'older-page-msg', + user: client.state.currentUser, + text: 'edited on older page', + createdAt: DateTime.utc(2025, 1, 1), + ); + client.addEvent(createUpdateMessageEvent(olderPageEdit)); + await Future.delayed(Duration.zero); + + // Tail is unchanged, no phantom entry inserted at position 0. + expect(channel.state!.messages.map((m) => m.id), ['tail-0', 'tail-1', 'tail-2']); + expect(channel.state!.pinnedMessages, isEmpty); + }, + ); + + test( + 'should update message in place when it IS in the loaded window', + () async { + const messageId = 'known'; + final seeded = Message( + id: messageId, + user: client.state.currentUser, + text: 'old', + createdAt: DateTime.utc(2026), + ); + channel.state!.updateChannelState( + channel.state!.channelState.copyWith(messages: [seeded]), + ); + channel.state!.isUpToDate = false; + + final edited = seeded.copyWith(text: 'new'); + client.addEvent(createUpdateMessageEvent(edited)); + await Future.delayed(Duration.zero); + + final stored = channel.state!.messages.singleWhere((m) => m.id == messageId); + expect(stored.text, equals('new')); + }, + ); + + test( + 'should still add to pinnedMessages when pinned:true even if not in loaded window', + () async { + channel.state!.isUpToDate = false; + expect(channel.state!.messages, isEmpty); + expect(channel.state!.pinnedMessages, isEmpty); + + const messageId = 'pin-me'; + final pinned = Message( + id: messageId, + user: client.state.currentUser, + pinned: true, + ); + client.addEvent(createUpdateMessageEvent(pinned)); + await Future.delayed(Duration.zero); + + expect(channel.state!.messages, isEmpty); + expect(channel.state!.pinnedMessages.length, equals(1)); + expect(channel.state!.pinnedMessages.first.id, equals(messageId)); + }, + ); + + test( + 'should NOT insert unknown reply into threads[parentId]', + () async { + const parentId = 'parent-1'; + final knownReply = Message( + id: 'known-reply', + parentId: parentId, + user: client.state.currentUser, + createdAt: DateTime.utc(2026), + ); + // Populate threads[parentId] via addNewMessage's thread-only path. + channel.state!.addNewMessage(knownReply); + await Future.delayed(Duration.zero); + expect(channel.state!.threads[parentId], hasLength(1)); + + channel.state!.isUpToDate = false; + + final phantomReply = Message( + id: 'other-reply', + parentId: parentId, + user: client.state.currentUser, + text: 'edited', + createdAt: DateTime.utc(2026, 1, 2), + ); + client.addEvent(createUpdateMessageEvent(phantomReply)); + await Future.delayed(Duration.zero); + + expect(channel.state!.threads[parentId]!.map((m) => m.id), ['known-reply']); + }, + ); + + test( + 'should NOT create phantom threads[parentId] entry for unloaded thread', + () async { + const parentId = 'unloaded-parent'; + // The thread was never paged in, so there's no entry for it. + expect(channel.state!.threads.containsKey(parentId), isFalse); + + channel.state!.isUpToDate = false; + + final phantomReply = Message( + id: 'phantom-reply', + parentId: parentId, + user: client.state.currentUser, + text: 'edited', + createdAt: DateTime.utc(2026, 1, 2), + ); + client.addEvent(createUpdateMessageEvent(phantomReply)); + await Future.delayed(Duration.zero); + + // The dropped reply must not leave behind an empty thread entry. + expect(channel.state!.threads.containsKey(parentId), isFalse); + }, + ); + + test( + 'should still expire activeLiveLocations for out-of-window message', + () async { + final liveLocation = Location( + channelCid: channel.cid, + userId: 'user1', + messageId: 'loc-msg', + latitude: 40.7128, + longitude: -74.0060, + createdByDeviceId: 'device1', + endAt: DateTime.now().add(const Duration(hours: 1)), + ); + + // Seed only activeLiveLocations, keeping `messages` empty — + // the exact "message is outside the loaded window" scenario. + channel.state!.updateChannelState( + ChannelState( + channel: channel.state!.channelState.channel, + activeLiveLocations: [liveLocation], + ), + ); + channel.state!.isUpToDate = false; + expect(channel.state!.messages, isEmpty); + expect(channel.state!.activeLiveLocations, hasLength(1)); + + // A message.updated that expires the live location. + final expiredMessage = Message( + id: 'loc-msg', + text: 'Live location shared', + sharedLocation: liveLocation.copyWith( + endAt: DateTime.now().subtract(const Duration(minutes: 1)), + ), + ); + client.addEvent(createUpdateMessageEvent(expiredMessage)); + await Future.delayed(Duration.zero); + + expect(channel.state!.messages, isEmpty); + expect(channel.state!.activeLiveLocations, isEmpty); + }, + ); + }); }, ); @@ -5645,6 +5884,188 @@ void main() { }, ); + // A `message.deleted` event for a message outside the loaded window + // must not upsert a "deleted" record into the sorted list — that would + // create a phantom entry with a gap. Pinned + live-location + // side-effects must still fire. + group( + EventType.messageDeleted, + () { + const channelId = 'test-channel-id'; + const channelType = 'test-channel-type'; + late Channel channel; + + setUp(() { + final channelState = _generateChannelState( + channelId, + channelType, + mockChannelConfig: true, + ownCapabilities: const [ChannelCapability.readEvents], + ); + channel = Channel.fromState(client, channelState); + }); + + tearDown(() => channel.dispose()); + + Event createDeleteMessageEvent(Message message, {bool hardDelete = false}) { + return Event( + cid: channel.cid, + type: EventType.messageDeleted, + message: message.copyWith( + type: MessageType.deleted, + deletedAt: DateTime.timestamp(), + ), + hardDelete: hardDelete, + ); + } + + // Same design as the `messageUpdated` guards: the check is + // "message-in-loaded-window" and is independent of `isUpToDate` — + // an event for a message on an older, unloaded page must not be + // turned into a phantom "deleted" record inserted into the sorted + // list. + group('when message is outside the loaded window', () { + test( + 'soft delete does NOT insert phantom "deleted" record into messages', + () async { + final tail = List.generate( + 3, + (i) => Message( + id: 'tail-$i', + user: client.state.currentUser, + text: 'tail $i', + createdAt: DateTime.utc(2026, 6, 1).add(Duration(seconds: i)), + ), + ); + channel.state!.updateChannelState( + channel.state!.channelState.copyWith(messages: tail), + ); + expect(channel.state!.messages, hasLength(3)); + + final olderPage = Message( + id: 'older-page-msg', + user: client.state.currentUser, + text: 'gone', + createdAt: DateTime.utc(2025, 1, 1), + ); + client.addEvent(createDeleteMessageEvent(olderPage)); + await Future.delayed(Duration.zero); + + expect(channel.state!.messages.map((m) => m.id), ['tail-0', 'tail-1', 'tail-2']); + }, + ); + + test( + 'soft delete marks message as deleted when it IS in the loaded window', + () async { + const messageId = 'known'; + final seeded = Message( + id: messageId, + user: client.state.currentUser, + text: 'hi', + createdAt: DateTime.utc(2026), + ); + channel.state!.updateChannelState( + channel.state!.channelState.copyWith(messages: [seeded]), + ); + channel.state!.isUpToDate = false; + + client.addEvent(createDeleteMessageEvent(seeded)); + await Future.delayed(Duration.zero); + + final stored = channel.state!.messages.singleWhere((m) => m.id == messageId); + expect(stored.type, equals(MessageType.deleted)); + expect(stored.deletedAt, isNotNull); + }, + ); + + test( + 'soft delete unpins a pinned-but-not-in-window message via _pinIsValid', + () async { + const messageId = 'pinned-msg'; + final pinned = Message( + id: messageId, + user: client.state.currentUser, + pinned: true, + createdAt: DateTime.utc(2026), + ); + // Seed only the pinnedMessages list — message absent from + // the main `messages` window. + channel.state!.updateChannelState( + channel.state!.channelState.copyWith(pinnedMessages: [pinned]), + ); + channel.state!.isUpToDate = false; + expect(channel.state!.messages, isEmpty); + expect(channel.state!.pinnedMessages, hasLength(1)); + + client.addEvent(createDeleteMessageEvent(pinned)); + await Future.delayed(Duration.zero); + + expect(channel.state!.messages, isEmpty); + expect(channel.state!.pinnedMessages, isEmpty); + }, + ); + + test( + 'soft delete still clears activeLiveLocations even when message not in window', + () async { + final liveLocation = Location( + channelCid: channel.cid, + userId: 'user1', + messageId: 'loc-msg', + latitude: 40.7128, + longitude: -74.0060, + createdByDeviceId: 'device1', + endAt: DateTime.now().add(const Duration(hours: 1)), + ); + + // Seed only activeLiveLocations, keeping `messages` empty. + channel.state!.updateChannelState( + ChannelState( + channel: channel.state!.channelState.channel, + activeLiveLocations: [liveLocation], + ), + ); + channel.state!.isUpToDate = false; + expect(channel.state!.messages, isEmpty); + expect(channel.state!.activeLiveLocations, hasLength(1)); + + final locationMessage = Message( + id: 'loc-msg', + text: 'Live location shared', + sharedLocation: liveLocation, + ); + client.addEvent(createDeleteMessageEvent(locationMessage)); + await Future.delayed(Duration.zero); + + expect(channel.state!.messages, isEmpty); + expect(channel.state!.activeLiveLocations, isEmpty); + }, + ); + + test( + 'hard delete is a no-op when message is not in the loaded window', + () async { + channel.state!.isUpToDate = false; + expect(channel.state!.messages, isEmpty); + + final phantom = Message( + id: 'phantom', + user: client.state.currentUser, + text: 'gone', + createdAt: DateTime.utc(2026), + ); + client.addEvent(createDeleteMessageEvent(phantom, hardDelete: true)); + await Future.delayed(Duration.zero); + + expect(channel.state!.messages, isEmpty); + expect(channel.state!.pinnedMessages, isEmpty); + }, + ); + }); + }, + ); + group('Member Events', () { const channelId = 'test-channel-id'; const channelType = 'test-channel-type';