recent_senders: Add the new MessageIdTracker and RecentSenders data structures

sm-sayedi · sm-sayedi · commit 0289061746a2 · 2024-07-12T01:32:27.000+04:30
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:
  web\src\recent_senders.ts
diff --git a/lib/model/message_list.dart b/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);
     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, inReverse: true);
 
       final fetchedMessages = _allMessagesVisible
         ? result.messages // Avoid unnecessarily copying the list.
diff --git a/lib/model/recent_senders.dart b/lib/model/recent_senders.dart
@@ -0,0 +1,167 @@
+import 'package:collection/collection.dart';
+import 'package:flutter/foundation.dart';
+
+import '../api/model/events.dart';
+import '../api/model/model.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;
+
+  /// Calls [handleMessage] for each of the message.
+  ///
+  /// [inReverse] is used for the optimization of the underlying
+  /// [MessageIdTracker.add] method. It should be `false` (default) when
+  /// [messages] are the initial messages in a narrow (fetched through
+  /// [MessageListView.fetchInitial]), and `true` when they are the older
+  /// messages in a narrow (fetched through [MessageListView.fetchOlder]).
+  void handleMessages(List<Message> messages, {bool inReverse = false}) {
+    for (int i = inReverse ? messages.length - 1 : 0;
+                 inReverse ? i >= 0              : i < messages.length;
+                 inReverse ? i--                 : i++) {
+      handleMessage(messages[i]);
+    }
+  }
+
+  /// 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;
+
+    // Track message in stream.
+    var sendersMap = streamSenders[streamId] ??= {};
+    var idTracker = sendersMap[senderId] ??= MessageIdTracker();
+    idTracker.add(messageId);
+
+    // Track message in topic.
+    final topicsMap = topicSenders[streamId] ??= {};
+    sendersMap = topicsMap[topic] ??= {};
+    idTracker = sendersMap[senderId] ??= MessageIdTracker();
+    idTracker.add(messageId);
+  }
+
+  void handleDeleteMessageEvent(DeleteMessageEvent event, Map<int, Message> cachedMessages) {
+    if (event.messageType != MessageType.stream) return;
+
+    for (final id in event.messageIds) {
+      final message = cachedMessages[id] as StreamMessage?;
+      if (message == null) break;
+      final StreamMessage(:streamId, :topic, :senderId, id: int messageId) = message;
+
+    _removeMessageInStream(
+      streamId: streamId, senderId: senderId, messageId: messageId);
+
+    _removeMessageInTopic(streamId: streamId, topic: topic, senderId: senderId,
+      messageId: messageId);
+    }
+  }
+
+  void _removeMessageInStream({
+    required int streamId,
+    required int senderId,
+    required int messageId,
+  }) {
+    if (streamSenders.isEmpty) return;
+
+    final sendersMap = streamSenders[streamId];
+    if (sendersMap == null) return;
+
+    final idTracker = sendersMap[senderId];
+    if (idTracker == null) return;
+
+    idTracker.remove(messageId);
+    if (idTracker.maxId == null) sendersMap.remove(senderId);
+    if (sendersMap.isEmpty) streamSenders.remove(streamId);
+  }
+
+  void _removeMessageInTopic({
+    required int streamId,
+    required String topic,
+    required int senderId,
+    required int messageId,
+  }) {
+    if (topicSenders.isEmpty) return;
+
+    final topicsMap = topicSenders[streamId];
+    if (topicsMap == null) return;
+
+    final sendersMap = topicsMap[topic];
+    if (sendersMap == null) return;
+
+    final idTracker = sendersMap[senderId];
+    if (idTracker == null) return;
+
+    idTracker.remove(messageId);
+    if (idTracker.maxId == null) sendersMap.remove(senderId);
+    if (sendersMap.isEmpty) topicsMap.remove(topic);
+    if (topicsMap.isEmpty) topicSenders.remove(streamId);
+  }
+}
+
+class MessageIdTracker {
+  /// A list of distinct message IDs, sorted ascendingly.
+  final 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 already present.
+  ///
+  /// Optimized, taking O(1) time, for the cases where that place is the start
+  /// (message fetched through [MessageListView.fetchOlder]) or the end (message
+  /// fetched through [MessageListView.fetchInitial] or
+  /// [PerAccountStore.handleEvent]), because those are the common cases for a
+  /// message the app receives. 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 == 0) {
+      _ids.addFirst(id);
+    } else if (i == _ids.length) {
+      _ids.addLast(id);
+    } else {
+      _ids.insert(i, id);
+    }
+  }
+
+  void remove(int id) => _ids.remove(id);
+
+  @override
+  bool operator ==(covariant MessageIdTracker other) {
+    if (identical(this, other)) return true;
+
+    return _ids.equals(other._ids);
+  }
+
+  @override
+  int get hashCode => Object.hashAll(_ids);
+
+  @override
+  String toString() => _ids.toString();
+}
diff --git a/lib/model/store.dart b/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 'unreads.dart';
 
@@ -251,6 +252,7 @@ class PerAccountStore extends ChangeNotifier with StreamStore, MessageStore {
       ),
       recentDmConversationsView: RecentDmConversationsView(
         initial: initialSnapshot.recentPrivateConversations, selfUserId: account.userId),
+      recentSenders: RecentSenders(),
     );
   }
 
@@ -270,6 +272,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),
@@ -361,6 +364,8 @@ class PerAccountStore extends ChangeNotifier with StreamStore, MessageStore {
 
   final RecentDmConversationsView recentDmConversationsView;
 
+  final RecentSenders recentSenders;
+
   ////////////////////////////////
   // Other digests of data.
 
@@ -471,6 +476,7 @@ class PerAccountStore extends ChangeNotifier with StreamStore, MessageStore {
       _messages.handleMessageEvent(event);
       unreads.handleMessageEvent(event);
       recentDmConversationsView.handleMessageEvent(event);
+      recentSenders.handleMessage(event.message);
       // When adding anything here (to handle [MessageEvent]),
       // it probably belongs in [reconcileMessages] too.
     } else if (event is UpdateMessageEvent) {
@@ -479,6 +485,8 @@ 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)].
+      recentSenders.handleDeleteMessageEvent(event, messages);
       _messages.handleDeleteMessageEvent(event);
       unreads.handleDeleteMessageEvent(event);
     } else if (event is UpdateMessageFlagsEvent) {
diff --git a/test/model/message_list_test.dart b/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';
 import 'test_store.dart';
 
 void main() {
@@ -141,6 +142,24 @@ void main() {
       ..haveOldest.isTrue();
   });
 
+  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);
+    checkMatchesMessages(messages, store.recentSenders);
+  });
+
   test('fetchOlder', () async {
     const narrow = CombinedFeedNarrow();
     await prepare(narrow: narrow);
@@ -233,6 +252,27 @@ void main() {
       ..messages.length.equals(200);
   });
 
+
+  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: 90 + i))
+        // Not subscribed to the stream with id 10.
+        ..add(eg.streamMessage(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);
+    checkMatchesMessages([...initialMessages, ...oldMessages], store.recentSenders);
+  });
+
   test('MessageEvent', () async {
     final stream = eg.stream();
     await prepare(narrow: StreamNarrow(stream.streamId));
diff --git a/test/model/recent_senders_checks.dart b/test/model/recent_senders_checks.dart
@@ -0,0 +1,29 @@
+import 'package:checks/checks.dart';
+import 'package:zulip/model/recent_senders.dart';
+
+extension RecentSendersChecks on Subject<RecentSenders> {
+  Subject<Map<int, Map<int, MessageIdTracker>>> get streamSenders =>
+    has((rs) => rs.streamSenders, 'streamSenders');
+
+  Subject<Map<int, Map<String, Map<int, MessageIdTracker>>>> get topicSenders =>
+    has((rs) => rs.topicSenders, 'topicSenders');
+
+  Subject<int?> latestMessageIdOfSenderInStream({
+    required int streamId,
+    required int senderId,
+  }) => has(
+          (rs) => rs.latestMessageIdOfSenderInStream(streamId: streamId,
+            senderId: senderId),
+          'latestMessageIdOfSenderInStream',
+        );
+
+  Subject<int?> latestMessageIdOfSenderInTopic({
+    required int streamId,
+    required String topic,
+    required int senderId,
+  }) => has(
+          (rs) => rs.latestMessageIdOfSenderInTopic(streamId: streamId,
+            topic: topic, senderId: senderId),
+          'latestMessageIdOfSenderInTopic',
+        );
+}
diff --git a/test/model/recent_senders_test.dart b/test/model/recent_senders_test.dart
