message: Consider unsubscribed/unknown channels in reconcileMessages

This fixes the "fourth buggy behavior" in #1798:
  #1798 (comment)

Fixes-partly: #1798

Author: chrisbobbe

lib/model/message.dart

@@ -417,6 +417,14 @@ class MessageStoreImpl extends HasChannelStore with MessageStore, _OutboxMessage
   }
 
   Message _reconcileUnrecognizedMessage(Message incoming) {
+    if (
+      incoming is StreamMessage
+      && subscriptions[incoming.streamId] == null
+    ) {
+      // The message is in an unsubscribed channel. It might grow stale;
+      // add it to _maybeStaleChannelMessages.
+      _maybeStaleChannelMessages.add(incoming.id);
+    }
     return _stripMatchFields(incoming);
   }
 
@@ -426,20 +434,57 @@ class MessageStoreImpl extends HasChannelStore with MessageStore, _OutboxMessage
     // overlaps another, e.g. a stream and a topic within it.
     //
     // Most often, the just-fetched message will look just like the one we
-    // already have.  But they can differ: message fetching happens out of band
-    // 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
-    // updated message is to use the version we have, that already reflects
-    // those events' changes.  So we always stick with the version we have.
-    // TODO(#1798) consider unsubscribed channels
-    return current;
+    // already have. But not always, and we can choose intelligently whether
+    // to keep the stored version or clobber it with the incoming one.
+
+    bool currentIsMaybeStale = false;
+    if (incoming is StreamMessage) {
+      if (subscriptions[incoming.streamId] != null) {
+        // The incoming version won't grow stale; it's in a subscribed channel.
+        // Remove it from _maybeStaleChannelMessages if it was there.
+        currentIsMaybeStale = _maybeStaleChannelMessages.remove(incoming.id);
+      } else {
+        assert(_maybeStaleChannelMessages.contains(incoming.id));
+        currentIsMaybeStale = true;
+      }
+    }
+
+    if (currentIsMaybeStale) {
+      // The event queue is unreliable for this message; the message was in an
+      // unsubscribed channel when we stored it or sometime since, so the stored
+      // version might be stale. Refresh it with the fetched version.
+      return _stripMatchFields(incoming);
+    } else {
+      // Message fetching happens out of band 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
+      // updated message is to use the version we have, that already reflects
+      // those events' changes.  So, stick with the version we have.
+      return current;
+    }
   }
 
+  /// 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 [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;
@@ -510,6 +555,30 @@ 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 > 8) {
+      assert(channelIds is! Set);
+      // optimization: https://github.com/zulip/zulip-flutter/pull/1912#discussion_r2479350329
+      channelIds = Set.from(channelIds);
+    }
+
+    // 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);
@@ -523,10 +592,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[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);
 
@@ -615,6 +692,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 channel, which means
+          // we expect our data on it to get stale.
+          _maybeStaleChannelMessages.add(messageId);
+        }
       }
 
       if (newTopic != origTopic) {
@@ -637,6 +720,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) {

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();
 

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,20 @@ 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!);
+    } else {
+      subscription = null;
+    }
     connection = store.connection as FakeApiConnection;
     notifiedCount = 0;
     messageList = MessageListView.init(store: store,
@@ -533,18 +538,137 @@ 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', () {
+      /// 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;
+
+        final 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)});
+      }
+
+      void checkClobber({Message? withMessageCopy}) {
+        final messageCopy = withMessageCopy ?? copyStoredMessage();
+        store.reconcileMessages([messageCopy]);
+        checkStoredMessageIdenticalTo(messageCopy);
+      }
+
+      void checkNoClobber() {
+        final messageBefore = store.messages.values.single;
+        store.reconcileMessages([copyStoredMessage()]);
+        checkStoredMessageIdenticalTo(messageBefore);
+      }
+
+      test('DM', () async {
+        await prepare();
+        final message = eg.dmMessage(id: 1, from: eg.otherUser, to: [eg.selfUser]);
+
+        store.reconcileMessages([message]);
+
+        // Not clobbering, because the first call didn't mark stale.
+        checkNoClobber();
+      });
+
+      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(stream: channel);
+
+          store.reconcileMessages([message]);
+
+          // Not clobbering, because the first call didn't mark stale,
+          // because the message was in a subscribed channel.
+          checkNoClobber();
+
+          await store.removeSubscription(channel.streamId);
+          // Clobbering because the unsubscribe event marked the message stale.
+          checkClobber();
+          // (Check that reconcileMessage itself didn't unmark as stale.)
+          checkClobber();
+
+          await store.addSubscription(eg.subscription(channel));
+          // The channel became subscribed,
+          // but the message's data hasn't been refreshed, so clobber…
+          checkClobber();
+
+          // …Now it's been refreshed, by reconcileMessages, so don't clobber.
+          checkNoClobber();
+
+          final otherChannel = eg.stream();
+          await store.addStream(otherChannel);
+          check(store.subscriptions[otherChannel.streamId]).isNull();
+          await store.handleEvent(
+            eg.updateMessageEventMoveFrom(origMessages: [message],
+              newStreamId: otherChannel.streamId));
+          // Message was moved to an unsubscribed channel, so clobber.
+          checkClobber(
+            withMessageCopy: copyStoredMessage(displayRecipient: otherChannel.name));
+          // (Check that reconcileMessage itself didn't unmark as stale.)
+          checkClobber();
+
+          // Subscribe, to mark message as not-stale, setting up another check…
+          await store.addSubscription(eg.subscription(otherChannel));
+
+          await store.handleEvent(ChannelDeleteEvent(id: 1, streams: [otherChannel]));
+          // Message was in a channel that became unknown, so clobber.
+          checkClobber();
+        });
+
+        test('in unsubscribed channel on first call', () async {
+          await prepare(isChannelSubscribed: false);
+          final message = eg.streamMessage();
+
+          store.reconcileMessages([message]);
+
+          checkClobber();
+          checkClobber();
+        });
+
+        test('new-message event when in unsubscribed channel', () async {
+          await prepare(isChannelSubscribed: false);
+          final message = eg.streamMessage();
+
+          await store.handleEvent(eg.messageEvent(message));
+
+          checkClobber();
+          checkClobber();
+        });
+
+        test('new-message event when in a subscribed channel', () async {
+          await prepare(isChannelSubscribed: true);
+          final message = eg.streamMessage();
+
+          await store.handleEvent(eg.messageEvent(message));
+
+          checkNoClobber();
+          checkNoClobber();
+        });
+      });
     });
 
     test('matchContent and matchTopic are removed', () async {