message: Consider unsubscribed/unknown channels in reconcileMessages

chrisbobbe · chrisbobbe · commit 6ed224456b50 · 2025-10-16T13:56:43.000-07:00
This fixes the "fourth buggy behavior" in zulip#1798: zulip#1798 (comment) Fixes-partly: zulip#1798
diff --git a/lib/model/message.dart b/lib/model/message.dart
@@ -409,20 +409,75 @@ class MessageStoreImpl extends HasChannelStore with MessageStore, _OutboxMessage
     // from the event queue, so there's inherently a race.
     //
     // If the fetched message reflects changes we haven't yet heard from the
-    // event queue, then it doesn't much matter which version we use: we'll
-    // soon get the corresponding events and apply the changes anyway.
-    // But if it lacks changes we've already heard from the event queue, then
-    // we won't hear those events again; the only way to wind up with an
+    // event queue, then normally (see [1]) it doesn't much matter which version
+    // we use: we'll soon get the corresponding events and apply the changes
+    // anyway. But if it lacks changes we've already heard from the event queue,
+    // then we won't hear those events again; the only way to wind up with an
     // updated message is to use the version we have, that already reflects
-    // those events' changes.  So we always stick with the version we have.
+    // those events' changes.  So we always stick with the version we have. [1]
+    //
+    // [1] With one exception: if the version we have was in an unsubscribed
+    //     channel when we got it or sometime since, we take the fetched version
+    //     instead. That's because our version might be stale; we don't expect
+    //     update events for messages in unsubscribed channels.
     for (int i = 0; i < messages.length; i++) {
       final message = messages[i];
-      messages[i] = this.messages.putIfAbsent(message.id, () {
-        message.matchContent = null;
-        message.matchTopic = null;
-        return message;
-      });
+
+      messages[i] = this.messages.update(message.id,
+        (current) => _reconcileWhenPresent(current, message),
+        ifAbsent: () => _reconcileWhenAbsent(message));
+    }
+  }
+
+  Message _reconcileWhenPresent(Message current, Message incoming) {
+    bool currentIsMaybeStale = false;
+    if (incoming is StreamMessage) {
+      if (subscriptions[incoming.streamId] != null) {
+        // The message is in a subscribed channel. Remove it from
+        // _maybeStaleChannelMessages if it was there.
+        currentIsMaybeStale = _maybeStaleChannelMessages.remove(incoming.id);
+      } else {
+        currentIsMaybeStale = _maybeStaleChannelMessages.contains(incoming.id);
+      }
+    }
+
+    return currentIsMaybeStale
+      ? _stripMatchFields(incoming)
+      : current;
+  }
+
+  Message _reconcileWhenAbsent(Message incoming) {
+    if (
+      incoming is StreamMessage
+      && subscriptions[incoming.streamId] == null
+    ) {
+      // The message is in an unsubscribed (or unknown) channel.
+      // Add it to _maybeStaleChannelMessages.
+      _maybeStaleChannelMessages.add(incoming.id);
     }
+    return _stripMatchFields(incoming);
+  }
+
+  /// Messages in [messages] whose data stream is or was presumably broken
+  /// by the message being in an unsubscribed channel.
+  ///
+  /// This is the subset of [messages] where the message was
+  /// in an unsubscribed channel when we added it or sometime since.
+  ///
+  /// We don't expect update events for messages in unsubscribed channels,
+  /// so if some of these maybe-stale messages appear in a fetch,
+  /// we'll always clobber our stored version with the fetched version.
+  /// See implementation comment in [reconcileMessages].
+  ///
+  /// (We have seen a few such events, actually --
+  /// maybe because the channel only recently became unsubscribed? --
+  /// but not consistently, and we're not supposed to rely on them.)
+  final Set<int> _maybeStaleChannelMessages = {};
+
+  Message _stripMatchFields(Message message) {
+    message.matchContent = null;
+    message.matchTopic = null;
+    return message;
   }
 
   @override
@@ -489,6 +544,29 @@ class MessageStoreImpl extends HasChannelStore with MessageStore, _OutboxMessage
     );
   }
 
+  void handleChannelDeleteEvent(ChannelDeleteEvent event) {
+    final channelIds = event.streams.map((channel) => channel.streamId);
+    _handleSubscriptionsRemoved(channelIds);
+  }
+
+  void handleSubscriptionRemoveEvent(SubscriptionRemoveEvent event) {
+    _handleSubscriptionsRemoved(event.streamIds);
+  }
+
+  void _handleSubscriptionsRemoved(Iterable<int> channelIds) {
+    if (channelIds.length > 1) {
+      assert(channelIds is! Set);
+      channelIds = Set.from(channelIds); // optimization
+    }
+
+    // Linear in [messages].
+    final affectedKnownMessageIds = messages.values
+      .where((message) => message is StreamMessage && channelIds.contains(message.streamId))
+      .map((message) => message.id);
+
+    _maybeStaleChannelMessages.addAll(affectedKnownMessageIds);
+  }
+
   void handleUserTopicEvent(UserTopicEvent event) {
     for (final view in _messageListViews) {
       view.handleUserTopicEvent(event);
@@ -502,10 +580,18 @@ class MessageStoreImpl extends HasChannelStore with MessageStore, _OutboxMessage
   }
 
   void handleMessageEvent(MessageEvent event) {
+    final message = event.message;
+
     // If the message is one we already know about (from a fetch),
     // clobber it with the one from the event system.
     // See [reconcileMessages] for reasoning.
-    messages[event.message.id] = event.message;
+    messages[event.message.id] = message;
+
+    if (message is StreamMessage && subscriptions[message.streamId] == null) {
+      // We didn't expect this event, because the channel is unsubscribed. But
+      // that doesn't mean we should expect future events about this message.
+      _maybeStaleChannelMessages.add(message.id);
+    }
 
     _handleMessageEventOutbox(event);
 
@@ -594,6 +680,12 @@ class MessageStoreImpl extends HasChannelStore with MessageStore, _OutboxMessage
         // See [StreamConversation.displayRecipient] on why the invalidation is
         // needed.
         message.conversation.displayRecipient = null;
+
+        if (subscriptions[newStreamId] == null) {
+          // The message was moved into an unsubscribed (or unknown) channel,
+          // which means we expect our data on it to get stale.
+          _maybeStaleChannelMessages.add(messageId);
+        }
       }
 
       if (newTopic != origTopic) {
@@ -616,6 +708,7 @@ class MessageStoreImpl extends HasChannelStore with MessageStore, _OutboxMessage
   void handleDeleteMessageEvent(DeleteMessageEvent event) {
     for (final messageId in event.messageIds) {
       messages.remove(messageId);
+      _maybeStaleChannelMessages.remove(messageId);
       _editMessageRequests.remove(messageId);
     }
     for (final view in _messageListViews) {
diff --git a/lib/model/store.dart b/lib/model/store.dart
@@ -821,11 +821,17 @@ class PerAccountStore extends PerAccountStoreBase with
 
       case ChannelEvent():
         assert(debugLog("server event: stream/${event.op}"));
+        if (event is ChannelDeleteEvent) {
+          _messages.handleChannelDeleteEvent(event);
+        }
         _channels.handleChannelEvent(event);
         notifyListeners();
 
       case SubscriptionEvent():
         assert(debugLog("server event: subscription/${event.op}"));
+        if (event is SubscriptionRemoveEvent) {
+          _messages.handleSubscriptionRemoveEvent(event);
+        }
         _channels.handleSubscriptionEvent(event);
         notifyListeners();
 
diff --git a/test/model/message_test.dart b/test/model/message_test.dart
@@ -36,7 +36,7 @@ void main() {
 
   // These "late" variables are the common state operated on by each test.
   // Each test case calls [prepare] to initialize them.
-  late Subscription subscription;
+  late Subscription? subscription;
   late PerAccountStore store;
   late FakeApiConnection connection;
   // [messageList] is here only for the sake of checking when it notifies.
@@ -54,15 +54,18 @@ void main() {
   /// Initialize [store] and the rest of the test state.
   Future<void> prepare({
     ZulipStream? stream,
+    bool isChannelSubscribed = true,
     int? zulipFeatureLevel,
   }) async {
     stream ??= eg.stream(streamId: eg.defaultStreamMessageStreamId);
-    subscription = eg.subscription(stream);
     final selfAccount = eg.selfAccount.copyWith(zulipFeatureLevel: zulipFeatureLevel);
     store = eg.store(account: selfAccount,
       initialSnapshot: eg.initialSnapshot(zulipFeatureLevel: zulipFeatureLevel));
     await store.addStream(stream);
-    await store.addSubscription(subscription);
+    if (isChannelSubscribed) {
+      subscription = eg.subscription(stream);
+      await store.addSubscription(subscription!);
+    }
     connection = store.connection as FakeApiConnection;
     notifiedCount = 0;
     messageList = MessageListView.init(store: store,
@@ -533,18 +536,141 @@ void main() {
       });
     });
 
-    test('on ID collision, new message does not clobber old in store.messages', () async {
-      await prepare();
-      final message = eg.streamMessage(id: 1, content: '<p>foo</p>');
-      await addMessages([message]);
-      check(store.messages).deepEquals({1: message});
-      final newMessage = eg.streamMessage(id: 1, content: '<p>bar</p>');
-      final messages = [newMessage];
-      store.reconcileMessages(messages);
-      check(messages).deepEquals(
-        // (We'll check more messages in an upcoming commit.)
-        [message].map(conditionIdentical));
-      check(store.messages).deepEquals({1: message});
+    group('fetched message with ID already in store.messages', () {
+      late Message messageCopy;
+
+      /// Makes a copy of the single message in [MessageStore.messages]
+      /// by round-tripping through [Message.fromJson] and [Message.toJson].
+      ///
+      /// If that message's [StreamMessage.conversation.displayRecipient]
+      /// is null, callers must provide a non-null [displayRecipient]
+      /// to allow [StreamConversation.fromJson] to complete without throwing.
+      Message copyStoredMessage({String? displayRecipient}) {
+        final message = store.messages.values.single;
+
+        Map<String, dynamic> json = message.toJson();
+        if (
+          message is StreamMessage
+          && message.conversation.displayRecipient == null
+        ) {
+          if (displayRecipient == null) throw ArgumentError();
+          json['display_recipient'] = displayRecipient;
+        }
+
+        return Message.fromJson(json);
+      }
+
+      /// Checks if the single message in [MessageStore.messages]
+      /// is identical to [message].
+      void checkStoredMessageIdenticalTo(Message message) {
+        check(store.messages)
+          .deepEquals({message.id: conditionIdentical(message)});
+      }
+
+      test('DM', () async {
+        await prepare();
+        final message = eg.dmMessage(id: 1, from: eg.otherUser, to: [eg.selfUser]);
+
+        store.reconcileMessages([message]);
+        checkStoredMessageIdenticalTo(message);
+        store.reconcileMessages([copyStoredMessage()]);
+        // Not clobbering, because the first call didn't mark stale.
+        checkStoredMessageIdenticalTo(message);
+      });
+
+      group('channel message; chooses correctly whether to clobber the stored version', () {
+        // Exercise the ways we move the message in and out of the "maybe stale"
+        // state. These include reconcileMessage itself, so sometimes we test
+        // repeated calls to that with nothing else happening in between.
+
+        test('various conditions', () async {
+          final channel = eg.stream();
+          await prepare(stream: channel, isChannelSubscribed: true);
+          final message = eg.streamMessage(id: 1, stream: channel);
+
+          final otherChannel = eg.stream();
+          await store.addStream(otherChannel);
+
+          store.reconcileMessages([message]);
+          checkStoredMessageIdenticalTo(message);
+          store.reconcileMessages([copyStoredMessage()]);
+          // Not clobbering, because the first call didn't mark stale,
+          // because the message was in a subscribed channel.
+          checkStoredMessageIdenticalTo(message);
+
+          await store.removeSubscription(channel.streamId);
+          messageCopy = copyStoredMessage();
+          store.reconcileMessages([messageCopy]);
+          // Clobbering because the unsubscribe event marked the message stale.
+          checkStoredMessageIdenticalTo(messageCopy);
+          messageCopy = copyStoredMessage();
+          store.reconcileMessages([messageCopy]);
+          // (Check that reconcileMessage itself didn't unmark as stale.)
+          checkStoredMessageIdenticalTo(messageCopy);
+
+          await store.addSubscription(eg.subscription(channel));
+          messageCopy = copyStoredMessage();
+          store.reconcileMessages([messageCopy]);
+          // The channel became subscribed,
+          // but the message's data hasn't been refreshed, so clobber…
+          checkStoredMessageIdenticalTo(messageCopy);
+
+          store.reconcileMessages([copyStoredMessage()]);
+          // …Now it's been refreshed, by reconcileMessages, so don't clobber.
+          checkStoredMessageIdenticalTo(messageCopy);
+
+          check(store.subscriptions[otherChannel.streamId]).isNull();
+          await store.handleEvent(
+            eg.updateMessageEventMoveFrom(origMessages: [message],
+            newStreamId: otherChannel.streamId));
+          messageCopy = copyStoredMessage(displayRecipient: otherChannel.name);
+          store.reconcileMessages([messageCopy]);
+          // Message was moved to an unsubscribed channel, so clobber.
+          checkStoredMessageIdenticalTo(messageCopy);
+          messageCopy = copyStoredMessage();
+          store.reconcileMessages([messageCopy]);
+          // (Check that reconcileMessage itself didn't unmark as stale.)
+          checkStoredMessageIdenticalTo(messageCopy);
+
+          // Subscribe, to mark message as not-stale, setting up another check…
+          await store.addSubscription(eg.subscription(otherChannel));
+          store.reconcileMessages([copyStoredMessage()]);
+          await store.handleEvent(ChannelDeleteEvent(id: 1, streams: [otherChannel]));
+          messageCopy = copyStoredMessage();
+          store.reconcileMessages([messageCopy]);
+          // Message was in a channel that became unknown, so clobber.
+          checkStoredMessageIdenticalTo(messageCopy);
+        });
+
+        test('in unsubscribed channel on first call', () async {
+          await prepare(isChannelSubscribed: false);
+          final message = eg.streamMessage(id: 1);
+
+          store.reconcileMessages([message]);
+          checkStoredMessageIdenticalTo(message);
+
+          messageCopy = copyStoredMessage();
+          store.reconcileMessages([messageCopy]);
+          checkStoredMessageIdenticalTo(messageCopy);
+          messageCopy = copyStoredMessage();
+          store.reconcileMessages([messageCopy]);
+          checkStoredMessageIdenticalTo(messageCopy);
+        });
+
+        test('new-message event when in unsubscribed channel', () async {
+          await prepare(isChannelSubscribed: false);
+          final message = eg.streamMessage(id: 1);
+
+          await store.handleEvent(eg.messageEvent(message));
+
+          messageCopy = copyStoredMessage();
+          store.reconcileMessages([messageCopy]);
+          checkStoredMessageIdenticalTo(messageCopy);
+          messageCopy = copyStoredMessage();
+          store.reconcileMessages([messageCopy]);
+          checkStoredMessageIdenticalTo(messageCopy);
+        });
+      });
     });
 
     test('matchContent and matchTopic are removed', () async {
