recent_senders: Add the new MessageIdTracker and RecentSenders data structures

MessageIdTracker data structure is used to keep track of message ids in
an ascending sorted list. It is used in RecentSenders data structure.
RecentSenders data structure is used to keep track of user messages in topics
and streams.

Much of this code is transcribed from Zulip web; in particular,
from:
  https://github.com/zulip/zulip/blob/bd04a30bbc6dc5bd7c20940a3d1d34cf8c8c6f28/web/src/recent_senders.ts

Author: sm-sayedi

lib/model/message_list.dart

@@ -403,6 +403,7 @@ class MessageListView with ChangeNotifier, _MessageSequence {
       numAfter: 0,
     );
     store.reconcileMessages(result.messages);
+    store.recentSenders.handleMessages(result.messages); // TODO(#824)
     for (final message in result.messages) {
       if (_messageVisible(message)) {
         _addMessage(message);
@@ -439,6 +440,7 @@ class MessageListView with ChangeNotifier, _MessageSequence {
       }
 
       store.reconcileMessages(result.messages);
+      store.recentSenders.handleMessages(result.messages); // TODO(#824)
 
       final fetchedMessages = _allMessagesVisible
         ? result.messages // Avoid unnecessarily copying the list.

lib/model/recent_senders.dart

@@ -0,0 +1,154 @@
+import 'package:collection/collection.dart';
+import 'package:flutter/foundation.dart';
+
+import '../api/model/events.dart';
+import '../api/model/model.dart';
+import 'algorithms.dart';
+
+/// A data structure to keep track of stream and topic messages of users (senders).
+///
+/// Use [latestMessageIdOfSenderInStream] and [latestMessageIdOfSenderInTopic]
+/// to get the relevant data.
+class RecentSenders {
+  // streamSenders[streamId][senderId] = MessageIdTracker
+  @visibleForTesting
+  final Map<int, Map<int, MessageIdTracker>> streamSenders = {};
+
+  // topicSenders[streamId][topic][senderId] = MessageIdTracker
+  @visibleForTesting
+  final Map<int, Map<String, Map<int, MessageIdTracker>>> topicSenders = {};
+
+  int? latestMessageIdOfSenderInStream({
+    required int streamId,
+    required int senderId,
+  }) => streamSenders[streamId]?[senderId]?.maxId;
+
+  int? latestMessageIdOfSenderInTopic({
+    required int streamId,
+    required String topic,
+    required int senderId,
+  }) => topicSenders[streamId]?[topic]?[senderId]?.maxId;
+
+  /// Records the necessary data from each message if it is a [StreamMessage].
+  ///
+  /// [messages] should be sorted by [id] ascendingly, which are, the way app
+  /// receives and handles messages.
+  void handleMessages(List<Message> messages) {
+    final messagesByUserInStream = <(int, int), List<int>>{};
+    final messagesByUserInTopic = <(int, String, int), List<int>>{};
+    for (final message in messages) {
+      if (message is! StreamMessage) continue;
+      final StreamMessage(:streamId, :topic, :senderId, id: int messageId) = message;
+      (messagesByUserInStream[(streamId, senderId)] ??= []).add(messageId);
+      (messagesByUserInTopic[(streamId, topic, senderId)] ??= []).add(messageId);
+    }
+
+    for (final entry in messagesByUserInStream.entries) {
+      final (streamId, senderId) = entry.key;
+      ((streamSenders[streamId] ??= {})
+        [senderId] ??= MessageIdTracker()).addAll(entry.value);
+    }
+    for (final entry in messagesByUserInTopic.entries) {
+      final (streamId, topic, senderId) = entry.key;
+      (((topicSenders[streamId] ??= {})[topic] ??= {})
+        [senderId] ??= MessageIdTracker()).addAll(entry.value);
+    }
+  }
+
+  /// Records the necessary data from [message] if it is a [StreamMessage].
+  ///
+  /// If [message] is not a [StreamMessage], this is a no-op.
+  void handleMessage(Message message) {
+    if (message is! StreamMessage) return;
+    final StreamMessage(:streamId, :topic, :senderId, id: int messageId) = message;
+    ((streamSenders[streamId] ??= {})
+      [senderId] ??= MessageIdTracker()).add(messageId);
+    (((topicSenders[streamId] ??= {})[topic] ??= {})
+      [senderId] ??= MessageIdTracker()).add(messageId);
+  }
+
+  void handleDeleteMessageEvent(DeleteMessageEvent event, Map<int, Message> cachedMessages) {
+    if (event.messageType != MessageType.stream) return;
+    final messagesByUserInStream = <(int, int), List<int>>{};
+    final messagesByUserInTopic = <(int, String, int), List<int>>{};
+
+    final DeleteMessageEvent(:streamId!, :topic!) = event;
+    for (final id in event.messageIds) {
+      final message = cachedMessages[id] as StreamMessage?;
+      if (message == null) continue;
+      final senderId = message.senderId;
+      (messagesByUserInStream[(streamId, senderId)] ??= []).add(id);
+      (messagesByUserInTopic[(streamId, topic, senderId)] ??= []).add(id);
+    }
+
+    for (final entry in messagesByUserInStream.entries) {
+      final MapEntry(key: (streamId, senderId), value: messages) = entry;
+      final sendersMap = streamSenders[streamId];
+      final idTracker = sendersMap?[senderId];
+
+      idTracker?.removeAll(messages);
+      if (idTracker?.maxId == null) sendersMap?.remove(senderId);
+      if (sendersMap?.isEmpty ?? false) streamSenders.remove(streamId);
+    }
+
+    for (final entry in messagesByUserInTopic.entries) {
+      final MapEntry(key: (streamId, topic, senderId), value: messages) = entry;
+      final topicsMap = topicSenders[streamId];
+      final sendersMap = topicsMap?[topic];
+      final idTracker = sendersMap?[senderId];
+
+      idTracker?.removeAll(messages);
+      if (idTracker?.maxId == null) sendersMap?.remove(senderId);
+      if (sendersMap?.isEmpty ?? false) topicsMap?.remove(topic);
+      if (topicsMap?.isEmpty ?? false) topicSenders.remove(streamId);
+    }
+  }
+}
+
+class MessageIdTracker {
+  /// A list of distinct message IDs, sorted ascendingly.
+  @visibleForTesting
+  QueueList<int> ids = QueueList.from([]);
+
+  /// The maximum id in the tracker list, or `null` if the list is empty.
+  int? get maxId => ids.lastOrNull;
+
+  /// Add the message ID to the tracker list at the proper place, if not present.
+  ///
+  /// Optimized, taking O(log n) time for the case where that place is the end,
+  /// because that's the common case for a message that is received through
+  /// [PerAccountStore.handleEvent]. May take O(n) time in some rare cases.
+  void add(int id) {
+    final i = lowerBound(ids, id);
+    if (i < ids.length && ids[i] == id) {
+      // The ID is already present. Nothing to do.
+      return;
+    }
+    if (i == ids.length) {
+      ids.addLast(id);
+    } else {
+      ids.insert(i, id);
+    }
+  }
+
+  /// Add the messages IDs to the tracker list at the proper place, if not present.
+  ///
+  /// [newIds] should be sorted ascendingly.
+  void addAll(List<int> newIds) {
+    if (ids.isEmpty) {
+      ids = QueueList.from(newIds);
+      return;
+    }
+    ids = setUnion(ids, newIds);
+  }
+
+  void removeAll(List<int> idsToRemove) {
+    ids.removeWhere((id) {
+      final i = lowerBound(idsToRemove, id);
+      return i < idsToRemove.length && idsToRemove[i] == id;
+    });
+  }
+
+  @override
+  String toString() => ids.toString();
+}

lib/model/store.dart

@@ -23,6 +23,7 @@ import 'database.dart';
 import 'message.dart';
 import 'message_list.dart';
 import 'recent_dm_conversations.dart';
+import 'recent_senders.dart';
 import 'stream.dart';
 import 'typing_status.dart';
 import 'unreads.dart';
@@ -256,6 +257,7 @@ class PerAccountStore extends ChangeNotifier with StreamStore, MessageStore {
       ),
       recentDmConversationsView: RecentDmConversationsView(
         initial: initialSnapshot.recentPrivateConversations, selfUserId: account.userId),
+      recentSenders: RecentSenders(),
     );
   }
 
@@ -276,6 +278,7 @@ class PerAccountStore extends ChangeNotifier with StreamStore, MessageStore {
     required MessageStoreImpl messages,
     required this.unreads,
     required this.recentDmConversationsView,
+    required this.recentSenders,
   }) : assert(selfUserId == globalStore.getAccount(accountId)!.userId),
        assert(realmUrl == globalStore.getAccount(accountId)!.realmUrl),
        assert(realmUrl == connection.realmUrl),
@@ -369,6 +372,8 @@ class PerAccountStore extends ChangeNotifier with StreamStore, MessageStore {
 
   final RecentDmConversationsView recentDmConversationsView;
 
+  final RecentSenders recentSenders;
+
   ////////////////////////////////
   // Other digests of data.
 
@@ -480,6 +485,7 @@ class PerAccountStore extends ChangeNotifier with StreamStore, MessageStore {
       _messages.handleMessageEvent(event);
       unreads.handleMessageEvent(event);
       recentDmConversationsView.handleMessageEvent(event);
+      recentSenders.handleMessage(event.message); // TODO(#824)
       // When adding anything here (to handle [MessageEvent]),
       // it probably belongs in [reconcileMessages] too.
     } else if (event is UpdateMessageEvent) {
@@ -488,6 +494,11 @@ class PerAccountStore extends ChangeNotifier with StreamStore, MessageStore {
       unreads.handleUpdateMessageEvent(event);
     } else if (event is DeleteMessageEvent) {
       assert(debugLog("server event: delete_message ${event.messageIds}"));
+      // This should be called before [_messages.handleDeleteMessageEvent(event)],
+      // as we need to know about each message for [event.messageIds],
+      // specifically, their `senderId`s. By calling this after the
+      // aforementioned line, we'll lose reference to those messages.
+      recentSenders.handleDeleteMessageEvent(event, messages);
       _messages.handleDeleteMessageEvent(event);
       unreads.handleDeleteMessageEvent(event);
     } else if (event is UpdateMessageFlagsEvent) {

test/model/message_list_test.dart

@@ -18,6 +18,7 @@ import '../api/model/model_checks.dart';
 import '../example_data.dart' as eg;
 import '../stdlib_checks.dart';
 import 'content_checks.dart';
+import 'recent_senders_test.dart' as recent_senders_test;
 import 'test_store.dart';
 
 void main() {
@@ -141,6 +142,25 @@ void main() {
       ..haveOldest.isTrue();
   });
 
+  // TODO(#824): move this test
+  test('fetchInitial, recent senders track all the messages', () async {
+    const narrow = CombinedFeedNarrow();
+    await prepare(narrow: narrow);
+    final messages = [
+      eg.streamMessage(),
+      // Not subscribed to the stream with id 10.
+      eg.streamMessage(stream: eg.stream(streamId: 10)),
+    ];
+    connection.prepare(json: newestResult(
+      foundOldest: false,
+      messages: messages,
+    ).toJson());
+    await model.fetchInitial();
+
+    check(model).messages.length.equals(1);
+    recent_senders_test.checkMatchesMessages(store.recentSenders, messages);
+  });
+
   test('fetchOlder', () async {
     const narrow = CombinedFeedNarrow();
     await prepare(narrow: narrow);
@@ -233,6 +253,27 @@ void main() {
       ..messages.length.equals(200);
   });
 
+  // TODO(#824): move this test
+  test('fetchOlder, recent senders track all the messages', () async {
+    const narrow = CombinedFeedNarrow();
+    await prepare(narrow: narrow);
+    final initialMessages = List.generate(10, (i) => eg.streamMessage(id: 100 + i));
+    await prepareMessages(foundOldest: false, messages: initialMessages);
+
+    final oldMessages = List.generate(10, (i) => eg.streamMessage(id: 89 + i))
+      // Not subscribed to the stream with id 10.
+      ..add(eg.streamMessage(id: 99, stream: eg.stream(streamId: 10)));
+    connection.prepare(json: olderResult(
+      anchor: 100, foundOldest: false,
+      messages: oldMessages,
+    ).toJson());
+    await model.fetchOlder();
+
+    check(model).messages.length.equals(20);
+    recent_senders_test.checkMatchesMessages(store.recentSenders,
+      [...initialMessages, ...oldMessages]);
+  });
+
   test('MessageEvent', () async {
     final stream = eg.stream();
     await prepare(narrow: StreamNarrow(stream.streamId));