From 9f39525ad4e8530fb9fe63aff9bda7d729291939 Mon Sep 17 00:00:00 2001 From: VelikovPetar Date: Wed, 1 Jul 2026 19:29:06 +0200 Subject: [PATCH] fix(llc): Don't upsert out-of-window messages on message.updated/deleted events Co-Authored-By: Claude --- packages/stream_chat/CHANGELOG.md | 6 + .../stream_chat/lib/src/client/channel.dart | 34 +- .../test/src/client/channel_test.dart | 343 ++++++++++++++++++ 3 files changed, 378 insertions(+), 5 deletions(-) diff --git a/packages/stream_chat/CHANGELOG.md b/packages/stream_chat/CHANGELOG.md index 75445d8f2..fe1ede251 100644 --- a/packages/stream_chat/CHANGELOG.md +++ b/packages/stream_chat/CHANGELOG.md @@ -1,3 +1,9 @@ +## Upcoming + +🐞 Fixed + +- 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 ✅ Added diff --git a/packages/stream_chat/lib/src/client/channel.dart b/packages/stream_chat/lib/src/client/channel.dart index 1432600a8..1fe785b5d 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 _updateMessages([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 _updateMessages([message], upsert: false); }), ); } @@ -3930,11 +3932,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 +3945,7 @@ class ChannelClientState { void _updateThreadMessages( Iterable messages, { Message Function(Message original, Message updated) update = _mergeUpdate, + bool upsert = true, }) { if (messages.isEmpty) return; @@ -3963,6 +3967,7 @@ class ChannelClientState { existing: threadMessages, toMerge: value, update: update, + upsert: upsert, ); // Update the thread with the modified message list. @@ -3976,6 +3981,7 @@ class ChannelClientState { void _updateChannelMessages( Iterable messages, { Message Function(Message original, Message updated) update = _mergeUpdate, + bool upsert = true, }) { if (messages.isEmpty) return; @@ -3996,6 +4002,7 @@ class ChannelClientState { existing: channelMessages, toMerge: affectedMessages, update: update, + upsert: upsert, ); // Calculate the new last message at time. @@ -4092,14 +4099,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 +4121,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 +4144,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..4957cf02b 100644 --- a/packages/stream_chat/test/src/client/channel_test.dart +++ b/packages/stream_chat/test/src/client/channel_test.dart @@ -5465,6 +5465,167 @@ 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 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 +5806,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';