Skip to content

Commit c02565c

Browse files
PIG208PIG208
committed
recent_senders: Handle moved messages
This is similar to how we add support to handling moves for unreads (commit 34e6201), especially the optimizations to avoid making unnecessary copies of message IDs when the entire conversation is moved (e.g. resolve/unresolve topic). An alternative approach to this is extracting helpers from handleMessages and handleDeleteMessageEvent and combining the two to handle moves, like web does (https://github.com/zulip/zulip/blob/bd04a30b/web/src/recent_senders.ts#L165-L190). Compared to that, creating a dedicated helper (this commit) makes it more straightforward to optimize for our performance needs. (The tests do not have to use PerAccountStore, but this setup makes it a bit more integrated.) Fixes: #901
1 parent 25781a1 commit c02565c

File tree

3 files changed

+241
-0
lines changed

3 files changed

+241
-0
lines changed

‎lib/model/recent_senders.dart

+79
Original file line numberDiff line numberDiff line change
@@ -68,6 +68,52 @@ class RecentSenders {
6868
[senderId] ??= MessageIdTracker()).add(messageId);
6969
}
7070

71+
/// Handles channel/topic updates when messages are moved.
72+
///
73+
/// This is a no-op if no message move happened.
74+
void handleUpdateMessageEvent(UpdateMessageEvent event, Map<int, Message> cachedMessages) {
75+
if (event.moveData == null) {
76+
return;
77+
}
78+
final UpdateMessageMoveData(
79+
:origStreamId, :newStreamId, :origTopic, :newTopic) = event.moveData!;
80+
81+
final messagesBySender = _groupStreamMessageIdsBySender(event.messageIds, cachedMessages);
82+
final sendersInChannel = streamSenders[origStreamId];
83+
final topicsInChannel = topicSenders[origStreamId];
84+
final sendersInTopic = topicsInChannel?[origTopic];
85+
for (final MapEntry(key: senderId, value: messages) in messagesBySender.entries) {
86+
// The later `popAll` calls require the message IDs to be sorted in
87+
// ascending order. Only sort as many as we need: the message IDs
88+
// with the same sender, instead of all of them in `event.messageIds`.
89+
messages.sort();
90+
91+
if (newStreamId != origStreamId) {
92+
final streamTracker = sendersInChannel?[senderId];
93+
final movedMessagesInStreamTracker = streamTracker?.popAll(messages);
94+
if (streamTracker?.maxId == null) sendersInChannel?.remove(senderId);
95+
if (movedMessagesInStreamTracker != null) {
96+
((streamSenders[newStreamId] ??= {})
97+
[senderId] ??= MessageIdTracker()).addAll(movedMessagesInStreamTracker);
98+
}
99+
}
100+
101+
// This does not need a check like the stream trackers one above,
102+
// because the conversation is guaranteed to have moved. This is an
103+
// invariant [UpdateMessageMoveData] offers.
104+
final topicTracker = sendersInTopic?[senderId];
105+
final movedMessagesInTopicTracker = topicTracker?.popAll(messages);
106+
if (topicTracker?.maxId == null) sendersInTopic?.remove(senderId);
107+
if (movedMessagesInTopicTracker != null) {
108+
(((topicSenders[newStreamId] ??= {})[newTopic] ??= {})
109+
[senderId] ??= MessageIdTracker()).addAll(movedMessagesInTopicTracker);
110+
}
111+
}
112+
if (sendersInChannel?.isEmpty ?? false) streamSenders.remove(origStreamId);
113+
if (sendersInTopic?.isEmpty ?? false) topicsInChannel?.remove(origTopic);
114+
if (topicsInChannel?.isEmpty ?? false) topicSenders.remove(origStreamId);
115+
}
116+
71117
void handleDeleteMessageEvent(DeleteMessageEvent event, Map<int, Message> cachedMessages) {
72118
if (event.messageType != MessageType.stream) return;
73119

@@ -153,6 +199,39 @@ class MessageIdTracker {
153199
ids.removeWhere((id) => binarySearch(idsToRemove, id) != -1);
154200
}
155201

202+
/// Remove message IDs found in [idsToRemove] from the tracker list.
203+
///
204+
/// Returns the removed message IDs sorted in ascending order, or `null` if
205+
/// nothing is removed.
206+
///
207+
/// [idsToRemove] should be sorted ascending.
208+
///
209+
/// Consider using [removeAll] if the returned message IDs are not needed.
210+
// Part of this is adapted from [ListBase.removeWhere].
211+
QueueList<int>? popAll(List<int> idsToRemove) {
212+
assert(isSortedWithoutDuplicates(idsToRemove));
213+
final retainedMessageIds =
214+
ids.where((id) => binarySearch(idsToRemove, id) == -1).toList();
215+
216+
if (retainedMessageIds.isEmpty) {
217+
// All message IDs in this tracker are removed; this is an optimization
218+
// to clear all ids and return the removed ones without making a new copy.
219+
final result = ids;
220+
ids = QueueList();
221+
return result;
222+
}
223+
224+
QueueList<int>? poppedMessageIds;
225+
if (retainedMessageIds.length != ids.length) {
226+
poppedMessageIds = QueueList.from(
227+
ids.where((id) => binarySearch(idsToRemove, id) != -1));
228+
ids.setRange(0, retainedMessageIds.length, retainedMessageIds);
229+
ids.length = retainedMessageIds.length;
230+
assert(isSortedWithoutDuplicates(poppedMessageIds));
231+
}
232+
return poppedMessageIds;
233+
}
234+
156235
@override
157236
String toString() => ids.toString();
158237
}

‎lib/model/store.dart

+1
Original file line numberDiff line numberDiff line change
@@ -767,6 +767,7 @@ class PerAccountStore extends ChangeNotifier with EmojiStore, UserStore, Channel
767767

768768
case UpdateMessageEvent():
769769
assert(debugLog("server event: update_message ${event.messageId}"));
770+
recentSenders.handleUpdateMessageEvent(event, messages);
770771
_messages.handleUpdateMessageEvent(event);
771772
unreads.handleUpdateMessageEvent(event);
772773

‎test/model/recent_senders_test.dart

+161
Original file line numberDiff line numberDiff line change
@@ -2,12 +2,15 @@ import 'package:checks/checks.dart';
22
import 'package:flutter_test/flutter_test.dart';
33
import 'package:zulip/api/model/model.dart';
44
import 'package:zulip/model/recent_senders.dart';
5+
import 'package:zulip/model/store.dart';
56
import '../example_data.dart' as eg;
7+
import 'test_store.dart';
68

79
/// [messages] should be sorted by [id] ascending.
810
void checkMatchesMessages(RecentSenders model, List<Message> messages) {
911
final Map<int, Map<int, Set<int>>> messagesByUserInStream = {};
1012
final Map<int, Map<TopicName, Map<int, Set<int>>>> messagesByUserInTopic = {};
13+
messages.sort((a, b) => a.id - b.id);
1114
for (final message in messages) {
1215
if (message is! StreamMessage) {
1316
throw UnsupportedError('Message of type ${message.runtimeType} is not expected.');
@@ -142,6 +145,164 @@ void main() {
142145
});
143146
});
144147

148+
group('RecentSenders.handleUpdateMessageEvent', () {
149+
late PerAccountStore store;
150+
late RecentSenders model;
151+
152+
final origChannel = eg.stream(); final newChannel = eg.stream();
153+
final origTopic = 'origTopic'; final newTopic = 'newTopic';
154+
final userX = eg.user(); final userY = eg.user();
155+
156+
Future<void> prepare(List<Message> messages) async {
157+
store = eg.store();
158+
await store.addMessages(messages);
159+
await store.addStreams([origChannel, newChannel]);
160+
await store.addUsers([userX, userY]);
161+
model = store.recentSenders;
162+
}
163+
164+
List<StreamMessage> copyMessagesWith(Iterable<StreamMessage> messages, {
165+
ZulipStream? newChannel,
166+
String? newTopic,
167+
}) {
168+
assert(newChannel != null || newTopic != null);
169+
return messages.map((message) => StreamMessage.fromJson(
170+
message.toJson()
171+
..['stream_id'] = newChannel?.streamId ?? message.streamId
172+
// See [StreamMessage.displayRecipient] for why this is needed.
173+
..['display_recipient'] = newChannel?.name ?? message.displayRecipient!
174+
175+
..['subject'] = newTopic ?? message.topic
176+
)).toList();
177+
}
178+
179+
test('move a conversation entirely, with additional unknown messages', () async {
180+
final messages = List.generate(10, (i) => eg.streamMessage(
181+
stream: origChannel, topic: origTopic, sender: userX));
182+
await prepare(messages);
183+
final unknownMessages = List.generate(10, (i) => eg.streamMessage(
184+
stream: origChannel, topic: origTopic, sender: userX));
185+
checkMatchesMessages(model, messages);
186+
187+
final messageIdsByUserInStreamBefore =
188+
model.streamSenders[origChannel.streamId]![userX.userId]!.ids;
189+
final messageIdsByUserInTopicBefore =
190+
model.topicSenders[origChannel.streamId]![eg.t(origTopic)]![userX.userId]!.ids;
191+
192+
await store.handleEvent(eg.updateMessageEventMoveFrom(
193+
origMessages: messages + unknownMessages,
194+
newStreamId: newChannel.streamId));
195+
checkMatchesMessages(model, copyMessagesWith(
196+
messages, newChannel: newChannel));
197+
198+
// Check we avoided creating a new list for the moved message IDs.
199+
check(messageIdsByUserInStreamBefore).identicalTo(
200+
model.streamSenders[newChannel.streamId]![userX.userId]!.ids);
201+
check(messageIdsByUserInTopicBefore).identicalTo(
202+
model.topicSenders[newChannel.streamId]![eg.t(origTopic)]![userX.userId]!.ids);
203+
});
204+
205+
test('move a conversation exactly', () async {
206+
final messages = List.generate(10, (i) => eg.streamMessage(
207+
stream: origChannel, topic: origTopic, sender: userX));
208+
await prepare(messages);
209+
210+
final messageIdsByUserInStreamBefore =
211+
model.streamSenders[origChannel.streamId]![userX.userId]!.ids;
212+
final messageIdsByUserInTopicBefore =
213+
model.topicSenders[origChannel.streamId]![eg.t(origTopic)]![userX.userId]!.ids;
214+
215+
await store.handleEvent(eg.updateMessageEventMoveFrom(
216+
origMessages: messages,
217+
newStreamId: newChannel.streamId,
218+
newTopicStr: newTopic));
219+
checkMatchesMessages(model, copyMessagesWith(
220+
messages, newChannel: newChannel, newTopic: newTopic));
221+
222+
// Check we avoided creating a new list for the moved message IDs.
223+
check(messageIdsByUserInStreamBefore).identicalTo(
224+
model.streamSenders[newChannel.streamId]![userX.userId]!.ids);
225+
check(messageIdsByUserInTopicBefore).identicalTo(
226+
model.topicSenders[newChannel.streamId]![eg.t(newTopic)]![userX.userId]!.ids);
227+
});
228+
229+
test('move a conversation partially to a different channel', () async {
230+
final messages = List.generate(10, (i) => eg.streamMessage(
231+
stream: origChannel, topic: origTopic));
232+
final movedMessages = messages.take(5).toList();
233+
final otherMessages = messages.skip(5);
234+
await prepare(messages);
235+
236+
await store.handleEvent(eg.updateMessageEventMoveFrom(
237+
origMessages: movedMessages,
238+
newStreamId: newChannel.streamId));
239+
checkMatchesMessages(model, [
240+
...copyMessagesWith(movedMessages, newChannel: newChannel),
241+
...otherMessages,
242+
]);
243+
});
244+
245+
test('move a conversation partially to a different topic, within the same channel', () async {
246+
final messages = List.generate(10, (i) => eg.streamMessage(
247+
stream: origChannel, topic: origTopic, sender: userX));
248+
final movedMessages = messages.take(5).toList();
249+
final otherMessages = messages.skip(5);
250+
await prepare(messages);
251+
252+
final messageIdsByUserInStreamBefore =
253+
model.streamSenders[origChannel.streamId]![userX.userId]!.ids;
254+
255+
await store.handleEvent(eg.updateMessageEventMoveFrom(
256+
origMessages: movedMessages,
257+
newTopicStr: newTopic));
258+
checkMatchesMessages(model, [
259+
...copyMessagesWith(movedMessages, newTopic: newTopic),
260+
...otherMessages,
261+
]);
262+
263+
// Check that we did not touch stream message IDs tracker
264+
// when there wasn't a stream move.
265+
check(messageIdsByUserInStreamBefore).identicalTo(
266+
model.streamSenders[origChannel.streamId]![userX.userId]!.ids);
267+
});
268+
269+
test('move a conversation with multiple senders', () async {
270+
final messages = [
271+
eg.streamMessage(stream: origChannel, topic: origTopic, sender: userX),
272+
eg.streamMessage(stream: origChannel, topic: origTopic, sender: userX),
273+
eg.streamMessage(stream: origChannel, topic: origTopic, sender: userY),
274+
];
275+
await prepare(messages);
276+
277+
await store.handleEvent(eg.updateMessageEventMoveFrom(
278+
origMessages: messages,
279+
newStreamId: newChannel.streamId));
280+
checkMatchesMessages(model, copyMessagesWith(
281+
messages, newChannel: newChannel));
282+
});
283+
284+
test('move a converstion, but message IDs from the event are not sorted in ascending order', () async {
285+
final messages = List.generate(10, (i) => eg.streamMessage(
286+
id: 100-i, stream: origChannel, topic: origTopic));
287+
await prepare(messages);
288+
289+
await store.handleEvent(eg.updateMessageEventMoveFrom(
290+
origMessages: messages,
291+
newStreamId: newChannel.streamId));
292+
checkMatchesMessages(model,
293+
copyMessagesWith(messages, newChannel: newChannel));
294+
});
295+
296+
test('message edit update without move', () async {
297+
final messages = List.generate(10, (i) => eg.streamMessage(
298+
stream: origChannel, topic: origTopic));
299+
await prepare(messages);
300+
301+
await store.handleEvent(eg.updateMessageEditEvent(messages[0]));
302+
checkMatchesMessages(model, messages);
303+
});
304+
});
305+
145306
test('RecentSenders.handleDeleteMessageEvent', () {
146307
final model = RecentSenders();
147308
final stream = eg.stream();

0 commit comments

Comments
 (0)