Skip to content

unreads: Add/use locatorMap, to efficiently locate messages #1703

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 1 commit into
base: main
Choose a base branch
from
Open

unreads: Add/use locatorMap, to efficiently locate messages #1703

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
134 changes: 74 additions & 60 deletions lib/model/unreads.dart
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,7 @@ class Unreads extends PerAccountStoreBase with ChangeNotifier {
required ChannelStore channelStore,
required UnreadMessagesSnapshot initial,
}) {
final locatorMap = <int, SendableNarrow>{};
final streams = <int, TopicKeyedMap<QueueList<int>>>{};
final dms = <DmNarrow, QueueList<int>>{};
final mentions = Set.of(initial.mentions);
Expand All @@ -57,23 +58,34 @@ class Unreads extends PerAccountStoreBase with ChangeNotifier {
// TODO(server-10) simplify away
(value) => setUnion(value, unreadChannelSnapshot.unreadMessageIds),
ifAbsent: () => QueueList.from(unreadChannelSnapshot.unreadMessageIds));
final narrow = TopicNarrow(streamId, topic);
for (final messageId in unreadChannelSnapshot.unreadMessageIds) {
locatorMap[messageId] = narrow;
}
}

for (final unreadDmSnapshot in initial.dms) {
final otherUserId = unreadDmSnapshot.otherUserId;
final narrow = DmNarrow.withUser(otherUserId, selfUserId: core.selfUserId);
dms[narrow] = QueueList.from(unreadDmSnapshot.unreadMessageIds);
for (final messageId in dms[narrow]!) {
locatorMap[messageId] = narrow;
}
}

for (final unreadHuddleSnapshot in initial.huddles) {
final narrow = DmNarrow.ofUnreadHuddleSnapshot(unreadHuddleSnapshot,
selfUserId: core.selfUserId);
dms[narrow] = QueueList.from(unreadHuddleSnapshot.unreadMessageIds);
for (final messageId in dms[narrow]!) {
locatorMap[messageId] = narrow;
}
}

return Unreads._(
core: core,
channelStore: channelStore,
locatorMap: locatorMap,
streams: streams,
dms: dms,
mentions: mentions,
Expand All @@ -84,6 +96,7 @@ class Unreads extends PerAccountStoreBase with ChangeNotifier {
Unreads._({
required super.core,
required this.channelStore,
required this.locatorMap,
required this.streams,
required this.dms,
required this.mentions,
Expand All @@ -92,6 +105,12 @@ class Unreads extends PerAccountStoreBase with ChangeNotifier {

final ChannelStore channelStore;

/// All unread messages, as: message ID → narrow ([TopicNarrow] or [DmNarrow]).
///
/// Enables efficient [isUnread] and efficient lookups in [streams] and [dms].
@visibleForTesting
final Map<int, SendableNarrow> locatorMap;

// TODO excluded for now; would need to handle nuances around muting etc.
// int count;

Expand Down Expand Up @@ -233,11 +252,8 @@ class Unreads extends PerAccountStoreBase with ChangeNotifier {
/// The unread state for [messageId], or null if unknown.
///
/// May be unknown only if [oldUnreadsMissing].
///
/// This is inefficient; it iterates through [dms] and [channels].
// TODO implement efficiently
bool? isUnread(int messageId) {
final isPresent = _slowIsPresentInDms(messageId) || _slowIsPresentInStreams(messageId);
final isPresent = locatorMap.containsKey(messageId);
if (oldUnreadsMissing && !isPresent) return null;
return isPresent;
}
Expand All @@ -250,9 +266,12 @@ class Unreads extends PerAccountStoreBase with ChangeNotifier {

switch (message) {
case StreamMessage():
final narrow = TopicNarrow.ofMessage(message);
locatorMap[event.message.id] = narrow;
_addLastInStreamTopic(message.id, message.streamId, message.topic);
case DmMessage():
final narrow = DmNarrow.ofMessage(message, selfUserId: selfUserId);
locatorMap[event.message.id] = narrow;
_addLastInDm(message.id, narrow);
}
if (
Expand Down Expand Up @@ -346,24 +365,35 @@ class Unreads extends PerAccountStoreBase with ChangeNotifier {
// 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
for (final messageId in messageToMoveIds) {
locatorMap.remove(messageId);
}
return true;
}

final narrow = TopicNarrow(newStreamId, newTopic);
for (final messageId in messageToMoveIds) {
locatorMap[messageId] = narrow;
}
_addAllInStreamTopic(messageToMoveIds, newStreamId, newTopic);

return true;
}

void handleDeleteMessageEvent(DeleteMessageEvent event) {
mentions.removeAll(event.messageIds);
final messageIdsSet = Set.of(event.messageIds);
switch (event.messageType) {
case MessageType.stream:
// All the messages are in [event.streamId] and [event.topic],
// so we can be more efficient than _removeAllInStreamsAndDms.
final streamId = event.streamId!;
final topic = event.topic!;
_removeAllInStreamTopic(messageIdsSet, streamId, topic);
_removeAllInStreamTopic(Set.of(event.messageIds), streamId, topic);
case MessageType.direct:
_slowRemoveAllInDms(messageIdsSet);
_removeAllInStreamsAndDms(event.messageIds, expectOnlyDms: true);
}
for (final messageId in event.messageIds) {
locatorMap.remove(messageId);
}

// TODO skip notifyListeners if unchanged?
Expand Down Expand Up @@ -405,15 +435,18 @@ class Unreads extends PerAccountStoreBase with ChangeNotifier {
switch (event) {
case UpdateMessageFlagsAddEvent():
if (event.all) {
locatorMap.clear();
streams.clear();
dms.clear();
mentions.clear();
oldUnreadsMissing = false;
} else {
final messageIdsSet = Set.of(event.messages);
mentions.removeAll(messageIdsSet);
_slowRemoveAllInStreams(messageIdsSet);
_slowRemoveAllInDms(messageIdsSet);
final messageIds = event.messages;
mentions.removeAll(messageIds);
_removeAllInStreamsAndDms(messageIds);
for (final messageId in messageIds) {
locatorMap.remove(messageId);
}
}
case UpdateMessageFlagsRemoveEvent():
final newlyUnreadInStreams = <int, TopicKeyedMap<QueueList<int>>>{};
Expand All @@ -431,12 +464,15 @@ class Unreads extends PerAccountStoreBase with ChangeNotifier {
}
switch (detail.type) {
case MessageType.stream:
final topics = (newlyUnreadInStreams[detail.streamId!] ??= makeTopicKeyedMap());
final messageIds = (topics[detail.topic!] ??= QueueList());
final UpdateMessageFlagsMessageDetail(:streamId, :topic) = detail;
locatorMap[messageId] = TopicNarrow(streamId!, topic!);
final topics = (newlyUnreadInStreams[streamId] ??= makeTopicKeyedMap());
final messageIds = (topics[topic] ??= QueueList());
messageIds.add(messageId);
case MessageType.direct:
final narrow = DmNarrow.ofUpdateMessageFlagsMessageDetail(selfUserId: selfUserId,
detail);
locatorMap[messageId] = narrow;
(newlyUnreadInDms[narrow] ??= QueueList())
.add(messageId);
}
Expand Down Expand Up @@ -489,15 +525,6 @@ class Unreads extends PerAccountStoreBase with ChangeNotifier {
notifyListeners();
}

// TODO use efficient lookups
bool _slowIsPresentInStreams(int messageId) {
return streams.values.any(
(topics) => topics.values.any(
(messageIds) => messageIds.contains(messageId),
),
);
}

void _addLastInStreamTopic(int messageId, int streamId, TopicName topic) {
((streams[streamId] ??= makeTopicKeyedMap())[topic] ??= QueueList())
.addLast(messageId);
Expand All @@ -517,26 +544,32 @@ class Unreads extends PerAccountStoreBase with ChangeNotifier {
);
}

// TODO use efficient model lookups
void _slowRemoveAllInStreams(Set<int> idsToRemove) {
final newlyEmptyStreams = <int>[];
for (final MapEntry(key: streamId, value: topics) in streams.entries) {
final newlyEmptyTopics = <TopicName>[];
for (final MapEntry(key: topic, value: messageIds) in topics.entries) {
messageIds.removeWhere((id) => idsToRemove.contains(id));
if (messageIds.isEmpty) {
newlyEmptyTopics.add(topic);
}
}
for (final topic in newlyEmptyTopics) {
topics.remove(topic);
}
if (topics.isEmpty) {
newlyEmptyStreams.add(streamId);
}
/// Remove [idsToRemove] from [streams] and [dms].
void _removeAllInStreamsAndDms(Iterable<int> idsToRemove, {bool expectOnlyDms = false}) {
final idsPresentByNarrow = <SendableNarrow, Set<int>>{};
for (final id in idsToRemove) {
final narrow = locatorMap[id];
if (narrow == null) continue;
(idsPresentByNarrow[narrow] ??= {}).add(id);
}
for (final streamId in newlyEmptyStreams) {
streams.remove(streamId);

for (final MapEntry(key: narrow, value: ids) in idsPresentByNarrow.entries) {
switch (narrow) {
case TopicNarrow():
if (expectOnlyDms) {
// TODO(log)?
}
_removeAllInStreamTopic(ids, narrow.streamId, narrow.topic);
case DmNarrow():
final messageIds = dms[narrow];
if (messageIds == null) return;

// ([QueueList] doesn't have a `removeAll`)
messageIds.removeWhere((id) => ids.contains(id));
if (messageIds.isEmpty) {
dms.remove(narrow);
}
}
}
}

Expand Down Expand Up @@ -599,11 +632,6 @@ class Unreads extends PerAccountStoreBase with ChangeNotifier {
return poppedMessageIds;
}

// TODO use efficient model lookups
bool _slowIsPresentInDms(int messageId) {
return dms.values.any((ids) => ids.contains(messageId));
}

void _addLastInDm(int messageId, DmNarrow narrow) {
(dms[narrow] ??= QueueList()).addLast(messageId);
}
Expand All @@ -618,18 +646,4 @@ class Unreads extends PerAccountStoreBase with ChangeNotifier {
(existing) => setUnion(existing, messageIds),
);
}

// TODO use efficient model lookups
void _slowRemoveAllInDms(Set<int> idsToRemove) {
final newlyEmptyDms = <DmNarrow>[];
for (final MapEntry(key: dmNarrow, value: messageIds) in dms.entries) {
messageIds.removeWhere((id) => idsToRemove.contains(id));
if (messageIds.isEmpty) {
newlyEmptyDms.add(dmNarrow);
}
}
for (final dmNarrow in newlyEmptyDms) {
dms.remove(dmNarrow);
}
}
}
1 change: 1 addition & 0 deletions test/model/unreads_checks.dart
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@ import 'package:zulip/model/narrow.dart';
import 'package:zulip/model/unreads.dart';

extension UnreadsChecks on Subject<Unreads> {
Subject<Map<int, SendableNarrow>> get locatorMap => has((u) => u.locatorMap, 'locatorMap');
Subject<Map<int, Map<TopicName, QueueList<int>>>> get streams => has((u) => u.streams, 'streams');
Subject<Map<DmNarrow, QueueList<int>>> get dms => has((u) => u.dms, 'dms');
Subject<Set<int>> get mentions => has((u) => u.mentions, 'mentions');
Expand Down
4 changes: 4 additions & 0 deletions test/model/unreads_test.dart
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -78,6 +78,7 @@ void main() {
assert(Set.of(messages.map((m) => m.id)).length == messages.length,
'checkMatchesMessages: duplicate messages in test input');

final Map<int, SendableNarrow> expectedLocatorMap = {};
final Map<int, TopicKeyedMap<QueueList<int>>> expectedStreams = {};
final Map<DmNarrow, QueueList<int>> expectedDms = {};
final Set<int> expectedMentions = {};
Expand All @@ -87,10 +88,12 @@ void main() {
}
switch (message) {
case StreamMessage():
expectedLocatorMap[message.id] = TopicNarrow.ofMessage(message);
final perTopic = expectedStreams[message.streamId] ??= makeTopicKeyedMap();
final messageIds = perTopic[message.topic] ??= QueueList();
messageIds.add(message.id);
case DmMessage():
expectedLocatorMap[message.id] = DmNarrow.ofMessage(message, selfUserId: store.selfUserId);
final narrow = DmNarrow.ofMessage(message, selfUserId: eg.selfUser.userId);
final messageIds = expectedDms[narrow] ??= QueueList();
messageIds.add(message.id);
Expand All @@ -112,6 +115,7 @@ void main() {
}

check(model)
..locatorMap.deepEquals(expectedLocatorMap)
..streams.deepEquals(expectedStreams)
..dms.deepEquals(expectedDms)
..mentions.unorderedEquals(expectedMentions);
Expand Down