fix(llc): Don't upsert out-of-window messages on message.updated/message.deleted events

packages/stream_chat/CHANGELOG.md

@@ -1,13 +1,17 @@
 ## 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.
 
 🐞 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
 

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,18 +3937,20 @@ class ChannelClientState {
   void _updateMessages(
     Iterable<Message> 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);
   }
 
   void _updateThreadMessages(
     Iterable<Message> 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] ?? <Message>[];
+      final existingThreadMessages = updatedThreads[thread];
+      final threadMessages = existingThreadMessages ?? <Message>[];
       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<Message> 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<Message> existing,
     required Iterable<Message> 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<Message> ? existing : existing.toList();
-    final toMergeList = toMerge is List<Message> ? toMerge : toMerge.toList();
+    var toMergeList = toMerge is List<Message> ? 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.