message: Store all messages centrally

In itself, this exacerbates #455 by making it trigger in more
circumstances.  That's OK for the moment, because this also gives us
a foundation on which to fix that issue soon.

A key part of testing this is the line added in message_list_test's
`checkInvariants`, which checks that all messages tracked by the
MessageListView are present in the central message store.

Co-authored-by: Chris Bobbe <[email protected]>

Author: gnprice

lib/model/message.dart

@@ -1,14 +1,36 @@
 import '../api/model/events.dart';
+import '../api/model/model.dart';
 import 'message_list.dart';
 
 /// The portion of [PerAccountStore] for messages and message lists.
 mixin MessageStore {
+  /// All known messages, indexed by [Message.id].
+  Map<int, Message> get messages;
+
   void registerMessageList(MessageListView view);
   void unregisterMessageList(MessageListView view);
+
+  /// Reconcile a batch of just-fetched messages with the store,
+  /// mutating the list.
+  ///
+  /// This is called after a [getMessages] request to report the result
+  /// to the store.
+  ///
+  /// The list's length will not change, but some entries may be replaced
+  /// by a different [Message] object with the same [Message.id].
+  /// All [Message] objects in the resulting list will be present in
+  /// [this.messages].
+  void reconcileMessages(List<Message> messages);
 }
 
 class MessageStoreImpl with MessageStore {
-  MessageStoreImpl();
+  MessageStoreImpl()
+    // There are no messages in InitialSnapshot, so we don't have
+    // a use case for initializing MessageStore with nonempty [messages].
+    : messages = {};
+
+  @override
+  final Map<int, Message> messages;
 
   final Set<MessageListView> _messageListViews = {};
 
@@ -36,31 +58,59 @@ class MessageStoreImpl with MessageStore {
     }
   }
 
+  @override
+  void reconcileMessages(List<Message> messages) {
+    // What to do when some of the just-fetched messages are already known?
+    // This is common and normal: in particular it happens when one message list
+    // 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.
+    for (int i = 0; i < messages.length; i++) {
+      final message = messages[i];
+      messages[i] = this.messages.putIfAbsent(message.id, () => message);
+    }
+  }
+
   void handleMessageEvent(MessageEvent event) {
+    // If the message is one we already know about (from a fetch),
+    // clobber it with the one from the event system.
+    // See [fetchedMessages] for reasoning.
+    messages[event.message.id] = event.message;
+
     for (final view in _messageListViews) {
       view.handleMessageEvent(event);
     }
   }
 
   void handleUpdateMessageEvent(UpdateMessageEvent event) {
     for (final view in _messageListViews) {
-      view.maybeUpdateMessage(event);
+      view.maybeUpdateMessage(event); // TODO update mainly in [messages] instead
     }
   }
 
   void handleDeleteMessageEvent(DeleteMessageEvent event) {
-    // TODO handle DeleteMessageEvent in MessageListView
+    // TODO handle DeleteMessageEvent, particularly in MessageListView
   }
 
   void handleUpdateMessageFlagsEvent(UpdateMessageFlagsEvent event) {
     for (final view in _messageListViews) {
-      view.maybeUpdateMessageFlags(event);
+      view.maybeUpdateMessageFlags(event); // TODO update mainly in [messages] instead
     }
   }
 
   void handleReactionEvent(ReactionEvent event) {
     for (final view in _messageListViews) {
-      view.maybeUpdateMessageReactions(event);
+      view.maybeUpdateMessageReactions(event); // TODO update mainly in [messages] instead
     }
   }
 }

lib/model/message_list.dart

@@ -378,6 +378,7 @@ class MessageListView with ChangeNotifier, _MessageSequence {
       numBefore: kMessageListFetchBatchSize,
       numAfter: 0,
     );
+    store.reconcileMessages(result.messages);
     for (final message in result.messages) {
       if (_messageVisible(message)) {
         _addMessage(message);
@@ -413,6 +414,8 @@ class MessageListView with ChangeNotifier, _MessageSequence {
         result.messages.removeLast();
       }
 
+      store.reconcileMessages(result.messages);
+
       final fetchedMessages = _allMessagesVisible
         ? result.messages // Avoid unnecessarily copying the list.
         : result.messages.where(_messageVisible);

lib/model/store.dart

@@ -340,12 +340,17 @@ class PerAccountStore extends ChangeNotifier with StreamStore, MessageStore {
   ////////////////////////////////
   // Messages, and summaries of messages.
 
+  @override
+  Map<int, Message> get messages => _messages.messages;
   @override
   void registerMessageList(MessageListView view) =>
     _messages.registerMessageList(view);
   @override
   void unregisterMessageList(MessageListView view) =>
     _messages.unregisterMessageList(view);
+  @override
+  void reconcileMessages(List<Message> messages) =>
+    _messages.reconcileMessages(messages);
 
   final MessageStoreImpl _messages;
 

test/model/message_list_test.dart

@@ -933,6 +933,7 @@ void checkInvariants(MessageListView model) {
   }
 
   for (final message in model.messages) {
+    check(model.store.messages)[message.id].isNotNull().identicalTo(message);
     check(model.narrow.containsMessage(message)).isTrue();
 
     if (message is! StreamMessage) continue;

test/model/message_test.dart

@@ -0,0 +1,116 @@
+import 'package:checks/checks.dart';
+import 'package:test/scaffolding.dart';
+import 'package:zulip/api/model/events.dart';
+import 'package:zulip/api/model/model.dart';
+import 'package:zulip/model/store.dart';
+
+import '../example_data.dart' as eg;
+
+void main() {
+  // These variables are the common state operated on by each test.
+  // Each test case calls [prepare] to initialize them.
+  late PerAccountStore store;
+
+  /// Initialize [store] and the rest of the test state.
+  void prepare() {
+    store = eg.store();
+  }
+
+  Future<void> addMessages(Iterable<Message> messages) async {
+    for (final m in messages) {
+      await store.handleEvent(MessageEvent(id: 0, message: m));
+    }
+  }
+
+  group('reconcileMessages', () {
+    test('from empty', () async {
+      prepare();
+      check(store.messages).isEmpty();
+      final message1 = eg.streamMessage();
+      final message2 = eg.streamMessage();
+      final message3 = eg.dmMessage(from: eg.otherUser, to: [eg.selfUser]);
+      final messages = [message1, message2, message3];
+      store.reconcileMessages(messages);
+      check(messages).deepEquals(
+        [message1, message2, message3]
+          .map((m) => (Subject<Object?> it) => it.identicalTo(m)));
+      check(store.messages).deepEquals({
+        for (final m in messages) m.id: m,
+      });
+    });
+
+    test('from not-empty', () async {
+      prepare();
+      final message1 = eg.streamMessage();
+      final message2 = eg.streamMessage();
+      final message3 = eg.dmMessage(from: eg.otherUser, to: [eg.selfUser]);
+      final messages = [message1, message2, message3];
+      await addMessages(messages);
+      final newMessage = eg.streamMessage();
+      store.reconcileMessages([newMessage]);
+      check(messages).deepEquals(
+        [message1, message2, message3]
+          .map((m) => (Subject<Object?> it) => it.identicalTo(m)));
+      check(store.messages).deepEquals({
+        for (final m in messages) m.id: m,
+        newMessage.id: newMessage,
+      });
+    });
+
+    test('on ID collision, new message does not clobber old in store.messages', () async {
+      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).single.identicalTo(message);
+      check(store.messages).deepEquals({1: message});
+    });
+  });
+
+  group('handleMessageEvent', () {
+    test('from empty', () async {
+      prepare();
+      check(store.messages).isEmpty();
+
+      final newMessage = eg.streamMessage();
+      await store.handleEvent(MessageEvent(id: 1, message: newMessage));
+      check(store.messages).deepEquals({
+        newMessage.id: newMessage,
+      });
+    });
+
+    test('from not-empty', () async {
+      prepare();
+      final messages = [
+        eg.streamMessage(),
+        eg.streamMessage(),
+        eg.dmMessage(from: eg.otherUser, to: [eg.selfUser]),
+      ];
+      await addMessages(messages);
+      check(store.messages).deepEquals({
+        for (final m in messages) m.id: m,
+      });
+
+      final newMessage = eg.streamMessage();
+      await store.handleEvent(MessageEvent(id: 1, message: newMessage));
+      check(store.messages).deepEquals({
+        for (final m in messages) m.id: m,
+        newMessage.id: newMessage,
+      });
+    });
+
+    test('new message clobbers old on ID collision', () async {
+      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>');
+      await store.handleEvent(MessageEvent(id: 1, message: newMessage));
+      check(store.messages).deepEquals({1: newMessage});
+    });
+  });
+}

test/model/store_checks.dart

@@ -42,6 +42,7 @@ extension PerAccountStoreChecks on Subject<PerAccountStore> {
   Subject<Map<int, ZulipStream>> get streams => has((x) => x.streams, 'streams');
   Subject<Map<String, ZulipStream>> get streamsByName => has((x) => x.streamsByName, 'streamsByName');
   Subject<Map<int, Subscription>> get subscriptions => has((x) => x.subscriptions, 'subscriptions');
+  Subject<Map<int, Message>> get messages => has((x) => x.messages, 'messages');
   Subject<Unreads> get unreads => has((x) => x.unreads, 'unreads');
   Subject<RecentDmConversationsView> get recentDmConversationsView => has((x) => x.recentDmConversationsView, 'recentDmConversationsView');
   Subject<AutocompleteViewManager> get autocompleteViewManager => has((x) => x.autocompleteViewManager, 'autocompleteViewManager');

test/model/test_store.dart

@@ -154,4 +154,14 @@ extension PerAccountStoreTestExtension on PerAccountStore {
       visibilityPolicy: visibilityPolicy,
     ));
   }
+
+  Future<void> addMessage(Message message) async {
+    await handleEvent(MessageEvent(id: 1, message: message));
+  }
+
+  Future<void> addMessages(Iterable<Message> messages) async {
+    for (final message in messages) {
+      await addMessage(message);
+    }
+  }
 }