unreads: Handle updates when there are moved messages

We could have extend `_removeAllInStreamTopic` to fit the needs of
handling moved unreads, but it is used in a code path where messages
are marked as read (which will become hotter after #81).

The new helper is designed to avoid making copies of message IDs
when possible.  As for name, we use "pop" (inspired by Python) to
indicate that the removed items are returned, since "removeAll" in
Dart doesn't carry that meaning.

Signed-off-by: Zixuan James Li <[email protected]>

Author: PIG208

lib/model/unreads.dart

@@ -259,10 +259,8 @@ class Unreads extends ChangeNotifier {
       (f) => f == MessageFlag.mentioned || f == MessageFlag.wildcardMentioned,
     );
 
-    // We assume this event can't signal a change in a message's 'read' flag.
-    // TODO can it actually though, when it's about messages being moved into an
-    //   unsubscribed stream?
-    //   https://chat.zulip.org/#narrow/stream/378-api-design/topic/mark-as-read.20events.20with.20message.20moves.3F/near/1639957
+    // We expect the event's 'read' flag to be boring,
+    // matching the message's local unread state.
     final bool isRead = event.flags.contains(MessageFlag.read);
     assert(() {
       final isUnreadLocally = isUnread(messageId);
@@ -272,6 +270,17 @@ class Unreads extends ChangeNotifier {
       // We were going to check something but can't; shrug.
       if (isUnreadLocally == null) return true;
 
+      final newChannelId = event.moveData?.newStreamId;
+      if (newChannelId != null && !channelStore.subscriptions.containsKey(newChannelId)) {
+        // When unread messages are moved to an unsubscribed channel, the server
+        // marks them as read without sending a mark-as-read event. Clients are
+        // asked to special-case this by marking them as read, which we do in
+        // _handleMessageMove. That contract is clear enough and doesn't involve
+        // this event's 'read' flag, so don't bother logging about the flag;
+        // its behavior seems like an implementation detail that could change.
+        return true;
+      }
+
       if (isUnreadLocally != isUnreadInEvent) {
         // If this happens, then either:
         // - the server and client have been out of sync about the message's
@@ -296,13 +305,40 @@ class Unreads extends ChangeNotifier {
         madeAnyUpdate |= mentions.add(messageId);
     }
 
-    // TODO(#901) handle moved messages
+    madeAnyUpdate |= _handleMessageMove(event);
 
     if (madeAnyUpdate) {
       notifyListeners();
     }
   }
 
+  bool _handleMessageMove(UpdateMessageEvent event) {
+    if (event.moveData == null) {
+      // No moved messages.
+      return false;
+    }
+    final UpdateMessageMoveData(
+      :origStreamId, :newStreamId, :propagateMode, :origTopic, :newTopic,
+    ) = event.moveData!;
+
+    final messageToMoveIds = _popAllInStreamTopic(
+      event.messageIds.toSet(), origStreamId, origTopic)?..sort();
+
+    if (messageToMoveIds == null || messageToMoveIds.isEmpty) return false;
+    assert(event.messageIds.toSet().containsAll(messageToMoveIds));
+
+    if (!channelStore.subscriptions.containsKey(newStreamId)) {
+      // Unreads moved to an unsubscribed channel; just drop them.
+      // See also:
+      //   https://chat.zulip.org/#narrow/channel/378-api-design/topic/mark-as-read.20events.20with.20message.20moves.3F/near/2101926
+      return true;
+    }
+
+    _addAllInStreamTopic(messageToMoveIds, newStreamId, newTopic);
+
+    return true;
+  }
+
   void handleDeleteMessageEvent(DeleteMessageEvent event) {
     mentions.removeAll(event.messageIds);
     final messageIdsSet = Set.of(event.messageIds);
@@ -506,6 +542,53 @@ class Unreads extends ChangeNotifier {
     }
   }
 
+  /// Remove unread stream messages contained in `incomingMessageIds`, with
+  /// the matching `streamId` and `topic`.
+  ///
+  /// Returns the removed message IDs, or `null` if no messages are affected.
+  ///
+  /// Use [_removeAllInStreamTopic] if the removed message IDs are not needed.
+  // Part of this is adapted from [ListBase.removeWhere].
+  QueueList<int>? _popAllInStreamTopic(Set<int> incomingMessageIds, int streamId, TopicName topic) {
+    final topics = streams[streamId];
+    if (topics == null) return null;
+    final messageIds = topics[topic];
+    if (messageIds == null) return null;
+
+    final retainedMessageIds = <int>[];
+    for (final id in messageIds) {
+      if (!incomingMessageIds.contains(id)) {
+        retainedMessageIds.add(id);
+      }
+    }
+
+    if (retainedMessageIds.isEmpty) {
+      // This is an optimization for the case when all messages in the
+      // conversation are removed, which avoids making a copy of `messageId`
+      // unnecessarily.
+      topics.remove(topic);
+      if (topics.isEmpty) {
+        streams.remove(streamId);
+      }
+      return messageIds;
+    }
+
+    QueueList<int>? poppedMessageIds;
+    if (retainedMessageIds.length != messageIds.length) {
+      poppedMessageIds = QueueList.from(
+        messageIds.where((id) => incomingMessageIds.contains(id)));
+      messageIds.setRange(0, retainedMessageIds.length, retainedMessageIds);
+      messageIds.length = retainedMessageIds.length;
+    }
+    if (messageIds.isEmpty) {
+      topics.remove(topic);
+      if (topics.isEmpty) {
+        streams.remove(streamId);
+      }
+    }
+    return poppedMessageIds;
+  }
+
   // TODO use efficient model lookups
   bool _slowIsPresentInDms(int messageId) {
     return dms.values.any((ids) => ids.contains(messageId));

test/model/unreads_test.dart

@@ -469,6 +469,145 @@ void main() {
         }
       }
     });
+
+    group('moves', () {
+      final origChannel = eg.stream();
+      const origTopic = 'origTopic';
+      const newTopic = 'newTopic';
+
+      final readMessages = List<StreamMessage>.generate(10,
+        (_) => eg.streamMessage(
+          stream: origChannel, topic: origTopic, flags: [MessageFlag.read]));
+      final unreadMessages = List<StreamMessage>.generate(10,
+        (_) => eg.streamMessage(stream: origChannel, topic: origTopic));
+
+      Future<void> prepareStore() async {
+        prepare();
+        await channelStore.addStream(origChannel);
+        await channelStore.addSubscription(eg.subscription(origChannel));
+      }
+
+      List<StreamMessage> copyMessagesWith(Iterable<StreamMessage> messages, {
+        ZulipStream? newChannel,
+        String? newTopic,
+      }) {
+        assert(newChannel != null || newTopic != null);
+        return messages.map((message) => StreamMessage.fromJson(
+          message.toJson()
+            ..['stream_id'] = newChannel?.streamId ?? message.streamId
+            ..['subject'] = newTopic ?? message.topic
+        )).toList();
+      }
+
+      test('smoke', () async {
+        await prepareStore();
+        final newChannel = eg.stream();
+        await channelStore.addStream(newChannel);
+        await channelStore.addSubscription(eg.subscription(newChannel));
+        fillWithMessages(unreadMessages);
+
+        model.handleUpdateMessageEvent(eg.updateMessageEventMoveFrom(
+          origMessages: unreadMessages,
+          newStreamId: newChannel.streamId,
+          newTopicStr: newTopic));
+        checkNotifiedOnce();
+        checkMatchesMessages(copyMessagesWith(unreadMessages,
+          newChannel: newChannel, newTopic: newTopic));
+      });
+
+      test('moving some read and unread messages from a conversation', () async {
+        await prepareStore();
+        final messagesToMove = [unreadMessages.first, readMessages.first];
+        fillWithMessages(unreadMessages + readMessages);
+
+        model.handleUpdateMessageEvent(eg.updateMessageEventMoveFrom(
+          origMessages: messagesToMove,
+          newTopicStr: newTopic));
+        checkNotifiedOnce();
+        checkMatchesMessages([
+          ...copyMessagesWith(unreadMessages.take(1), newTopic: newTopic),
+          ...unreadMessages.skip(1),
+        ]);
+      });
+
+      test('moving all read and unread messages from a conversation', () async {
+        await prepareStore();
+        final allMessages = unreadMessages + readMessages;
+        fillWithMessages(allMessages);
+
+        model.handleUpdateMessageEvent(eg.updateMessageEventMoveFrom(
+          origMessages: allMessages,
+          newTopicStr: newTopic));
+        checkNotifiedOnce();
+        checkMatchesMessages(copyMessagesWith(unreadMessages, newTopic: newTopic));
+      });
+
+      test('moving to unsubscribed channels drops the unreads', () async {
+        await prepareStore();
+        final unsubscribedChannel = eg.stream();
+        await channelStore.addStream(unsubscribedChannel);
+        assert(!channelStore.subscriptions.containsKey(
+          unsubscribedChannel.streamId));
+        fillWithMessages(unreadMessages);
+
+        model.handleUpdateMessageEvent(eg.updateMessageEventMoveFrom(
+          origMessages: unreadMessages,
+          newStreamId: unsubscribedChannel.streamId));
+        checkNotifiedOnce();
+        checkMatchesMessages([]);
+      });
+
+      test('tolerates unsorted messages', () async {
+        await prepareStore();
+        final unreadMessages = List.generate(10, (i) =>
+          eg.streamMessage(
+            id: 1000 - i, stream: origChannel, topic: origTopic));
+        fillWithMessages(unreadMessages);
+
+        model.handleUpdateMessageEvent(eg.updateMessageEventMoveFrom(
+          origMessages: unreadMessages,
+          newTopicStr: newTopic));
+        checkNotifiedOnce();
+        checkMatchesMessages(copyMessagesWith(unreadMessages, newTopic: newTopic));
+      });
+
+      test('tolerates unreads unknown to the model', () async {
+        await prepareStore();
+        fillWithMessages(unreadMessages);
+
+        final unknownChannel = eg.stream();
+        assert(!channelStore.streams.containsKey(unknownChannel.streamId));
+        final unknownUnreadMessage = eg.streamMessage(
+          stream: unknownChannel, topic: origTopic);
+
+        model.handleUpdateMessageEvent(eg.updateMessageEventMoveFrom(
+          origMessages: [unknownUnreadMessage],
+          newTopicStr: newTopic));
+        checkNotNotified();
+        checkMatchesMessages(unreadMessages);
+      });
+
+      test('moving read messages has no effect', () async {
+        await prepareStore();
+        fillWithMessages(unreadMessages + readMessages);
+
+        model.handleUpdateMessageEvent(eg.updateMessageEventMoveFrom(
+          origMessages: readMessages,
+          newTopicStr: newTopic));
+        checkNotNotified();
+        checkMatchesMessages(unreadMessages);
+      });
+
+      test('message edit but no move', () async {
+        await prepareStore();
+        fillWithMessages(unreadMessages);
+
+        model.handleUpdateMessageEvent(eg.updateMessageEditEvent(
+          unreadMessages.first));
+        checkNotNotified();
+        checkMatchesMessages(unreadMessages);
+      });
+    });
   });