msglist: Apply stream and topic muting to MessageListView

This is the last step for implementing stream and topic muting,
i.e. #346.  (Except for what's been split off to follow-up issues,
notably #421.)

Fixes: #346

lib/model/message_list.dart

@@ -307,6 +307,31 @@ class MessageListView with ChangeNotifier, _MessageSequence {
   final PerAccountStore store;
   final Narrow narrow;
 
+  /// Whether [message] should actually appear in this message list,
+  /// given that it does belong to the narrow.
+  ///
+  /// This depends in particular on whether the message is muted in
+  /// one way or another.
+  bool _messageVisible(Message message) {
+    switch (narrow) {
+      case AllMessagesNarrow():
+        return switch (message) {
+          StreamMessage() =>
+            store.isTopicVisible(message.streamId, message.subject),
+          DmMessage() => true,
+        };
+
+      case StreamNarrow(:final streamId):
+        assert(message is StreamMessage && message.streamId == streamId);
+        if (message is! StreamMessage) return false;
+        return store.isTopicVisibleInStream(streamId, message.subject);
+
+      case TopicNarrow():
+      case DmNarrow():
+        return true;
+    }
+  }
+
   /// Fetch messages, starting from scratch.
   Future<void> fetchInitial() async {
     // TODO(#80): fetch from anchor firstUnread, instead of newest
@@ -321,7 +346,9 @@ class MessageListView with ChangeNotifier, _MessageSequence {
       numAfter: 0,
     );
     for (final message in result.messages) {
-      _addMessage(message);
+      if (_messageVisible(message)) {
+        _addMessage(message);
+      }
     }
     _fetched = true;
     _haveOldest = result.foundOldest;
@@ -353,7 +380,7 @@ class MessageListView with ChangeNotifier, _MessageSequence {
         result.messages.removeLast();
       }
 
-      _insertAllMessages(0, result.messages);
+      _insertAllMessages(0, result.messages.where(_messageVisible));
       _haveOldest = result.foundOldest;
     } finally {
       _fetchingOlder = false;
@@ -366,7 +393,7 @@ class MessageListView with ChangeNotifier, _MessageSequence {
   ///
   /// Called in particular when we get a [MessageEvent].
   void maybeAddMessage(Message message) {
-    if (!narrow.containsMessage(message)) {
+    if (!narrow.containsMessage(message) || !_messageVisible(message)) {
       return;
     }
     if (!_fetched) {

lib/model/narrow.dart

@@ -11,8 +11,14 @@ sealed class Narrow {
   /// This const constructor allows subclasses to have const constructors.
   const Narrow();
 
-  // TODO implement muting; will need containsMessage to take more params
-  //   This means stream muting, topic un/muting, and user muting.
+  /// Whether this message satisfies the filters of this narrow.
+  ///
+  /// This is true just when the server would be expected to include the message
+  /// in a [getMessages] request for this narrow, given appropriate anchor etc.
+  ///
+  /// This does not necessarily mean the message list would show this message
+  /// when navigated to this narrow; in particular it does not address the
+  /// question of whether the stream or topic, or the sending user, is muted.
   bool containsMessage(Message message);
 
   /// This narrow, expressed as an [ApiNarrow].

lib/model/stream.dart

@@ -180,6 +180,7 @@ class StreamStoreImpl with StreamStore {
           case SubscriptionProperty.color:
             subscription.color                  = event.value as int;
           case SubscriptionProperty.isMuted:
+            // TODO(#421) update [MessageListView] if affected
             subscription.isMuted                = event.value as bool;
           case SubscriptionProperty.inHomeView:
             subscription.isMuted                = !(event.value as bool);
@@ -211,6 +212,7 @@ class StreamStoreImpl with StreamStore {
     if (_warnInvalidVisibilityPolicy(visibilityPolicy)) {
       visibilityPolicy = UserTopicVisibilityPolicy.none;
     }
+    // TODO(#421) update [MessageListView] if affected
     if (visibilityPolicy == UserTopicVisibilityPolicy.none) {
       // This is the "zero value" for this type, which our data structure
       // represents by leaving the topic out entirely.

test/api/model/model_checks.dart

@@ -16,6 +16,7 @@ extension StreamColorSwatchChecks on Subject<StreamColorSwatch> {
 }
 
 extension MessageChecks on Subject<Message> {
+  Subject<int> get id => has((e) => e.id, 'id');
   Subject<String> get content => has((e) => e.content, 'content');
   Subject<bool> get isMeMessage => has((e) => e.isMeMessage, 'isMeMessage');
   Subject<int?> get lastEditTimestamp => has((e) => e.lastEditTimestamp, 'lastEditTimestamp');

test/model/message_list_test.dart

@@ -17,10 +17,12 @@ import '../api/model/model_checks.dart';
 import '../example_data.dart' as eg;
 import '../stdlib_checks.dart';
 import 'content_checks.dart';
+import 'test_store.dart';
 
 void main() async {
   // These variables are the common state operated on by each test.
   // Each test case calls [prepare] to initialize them.
+  late Subscription subscription;
   late PerAccountStore store;
   late FakeApiConnection connection;
   late MessageListView model;
@@ -35,7 +37,10 @@ void main() async {
 
   /// Initialize [model] and the rest of the test state.
   void prepare({Narrow narrow = const AllMessagesNarrow()}) {
-    store = eg.store();
+    final stream = eg.stream();
+    subscription = eg.subscription(stream);
+    store = eg.store()
+      ..addStream(stream)..addSubscription(subscription);
     connection = store.connection as FakeApiConnection;
     notifiedCount = 0;
     model = MessageListView.init(store: store, narrow: narrow)
@@ -548,6 +553,141 @@ void main() async {
     check(model.contents[0]).equalsNode(correctContent);
   });
 
+  group('stream/topic muting', () {
+    test('in AllMessagesNarrow', () async {
+      final stream1 = eg.stream(streamId: 1, name: 'stream 1');
+      final stream2 = eg.stream(streamId: 2, name: 'stream 2');
+      prepare(narrow: const AllMessagesNarrow());
+      store.addStreams([stream1, stream2]);
+      store.addSubscription(eg.subscription(stream1));
+      store.addUserTopic(stream1, 'B', UserTopicVisibilityPolicy.muted);
+      store.addSubscription(eg.subscription(stream2, isMuted: true));
+      store.addUserTopic(stream2, 'C', UserTopicVisibilityPolicy.unmuted);
+
+      // Check filtering on fetchInitial…
+      await prepareMessages(foundOldest: false, messages: [
+        eg.streamMessage(id: 201, stream: stream1, topic: 'A'),
+        eg.streamMessage(id: 202, stream: stream1, topic: 'B'),
+        eg.streamMessage(id: 203, stream: stream2, topic: 'C'),
+        eg.streamMessage(id: 204, stream: stream2, topic: 'D'),
+        eg.dmMessage(    id: 205, from: eg.otherUser, to: [eg.selfUser]),
+      ]);
+      final expected = <int>[];
+      check(model.messages.map((m) => m.id))
+        .deepEquals(expected..addAll([201, 203, 205]));
+
+      // … and on fetchOlder…
+      connection.prepare(json: olderResult(
+        anchor: 201, foundOldest: true, messages: [
+          eg.streamMessage(id: 101, stream: stream1, topic: 'A'),
+          eg.streamMessage(id: 102, stream: stream1, topic: 'B'),
+          eg.streamMessage(id: 103, stream: stream2, topic: 'C'),
+          eg.streamMessage(id: 104, stream: stream2, topic: 'D'),
+          eg.dmMessage(    id: 105, from: eg.otherUser, to: [eg.selfUser]),
+        ]).toJson());
+      await model.fetchOlder();
+      checkNotified(count: 2);
+      check(model.messages.map((m) => m.id))
+        .deepEquals(expected..insertAll(0, [101, 103, 105]));
+
+      // … and on maybeAddMessage.
+      model.maybeAddMessage(eg.streamMessage(id: 301, stream: stream1, topic: 'A'));
+      checkNotifiedOnce();
+      check(model.messages.map((m) => m.id)).deepEquals(expected..add(301));
+
+      model.maybeAddMessage(eg.streamMessage(id: 302, stream: stream1, topic: 'B'));
+      checkNotNotified();
+      check(model.messages.map((m) => m.id)).deepEquals(expected);
+
+      model.maybeAddMessage(eg.streamMessage(id: 303, stream: stream2, topic: 'C'));
+      checkNotifiedOnce();
+      check(model.messages.map((m) => m.id)).deepEquals(expected..add(303));
+
+      model.maybeAddMessage(eg.streamMessage(id: 304, stream: stream2, topic: 'D'));
+      checkNotNotified();
+      check(model.messages.map((m) => m.id)).deepEquals(expected);
+
+      model.maybeAddMessage(eg.dmMessage(id: 305, from: eg.otherUser, to: [eg.selfUser]));
+      checkNotifiedOnce();
+      check(model.messages.map((m) => m.id)).deepEquals(expected..add(305));
+    });
+
+    test('in StreamNarrow', () async {
+      final stream = eg.stream(streamId: 1, name: 'stream 1');
+      prepare(narrow: StreamNarrow(stream.streamId));
+      store.addStream(stream);
+      store.addSubscription(eg.subscription(stream, isMuted: true));
+      store.addUserTopic(stream, 'A', UserTopicVisibilityPolicy.unmuted);
+      store.addUserTopic(stream, 'C', UserTopicVisibilityPolicy.muted);
+
+      // Check filtering on fetchInitial…
+      await prepareMessages(foundOldest: false, messages: [
+        eg.streamMessage(id: 201, stream: stream, topic: 'A'),
+        eg.streamMessage(id: 202, stream: stream, topic: 'B'),
+        eg.streamMessage(id: 203, stream: stream, topic: 'C'),
+      ]);
+      final expected = <int>[];
+      check(model.messages.map((m) => m.id))
+        .deepEquals(expected..addAll([201, 202]));
+
+      // … and on fetchOlder…
+      connection.prepare(json: olderResult(
+        anchor: 201, foundOldest: true, messages: [
+          eg.streamMessage(id: 101, stream: stream, topic: 'A'),
+          eg.streamMessage(id: 102, stream: stream, topic: 'B'),
+          eg.streamMessage(id: 103, stream: stream, topic: 'C'),
+        ]).toJson());
+      await model.fetchOlder();
+      checkNotified(count: 2);
+      check(model.messages.map((m) => m.id))
+        .deepEquals(expected..insertAll(0, [101, 102]));
+
+      // … and on maybeAddMessage.
+      model.maybeAddMessage(eg.streamMessage(id: 301, stream: stream, topic: 'A'));
+      checkNotifiedOnce();
+      check(model.messages.map((m) => m.id)).deepEquals(expected..add(301));
+
+      model.maybeAddMessage(eg.streamMessage(id: 302, stream: stream, topic: 'B'));
+      checkNotifiedOnce();
+      check(model.messages.map((m) => m.id)).deepEquals(expected..add(302));
+
+      model.maybeAddMessage(eg.streamMessage(id: 303, stream: stream, topic: 'C'));
+      checkNotNotified();
+      check(model.messages.map((m) => m.id)).deepEquals(expected);
+    });
+
+    test('in TopicNarrow', () async {
+      final stream = eg.stream(streamId: 1, name: 'stream 1');
+      prepare(narrow: TopicNarrow(stream.streamId, 'A'));
+      store.addStream(stream);
+      store.addSubscription(eg.subscription(stream, isMuted: true));
+      store.addUserTopic(stream, 'A', UserTopicVisibilityPolicy.muted);
+
+      // Check filtering on fetchInitial…
+      await prepareMessages(foundOldest: false, messages: [
+        eg.streamMessage(id: 201, stream: stream, topic: 'A'),
+      ]);
+      final expected = <int>[];
+      check(model.messages.map((m) => m.id))
+        .deepEquals(expected..addAll([201]));
+
+      // … and on fetchOlder…
+      connection.prepare(json: olderResult(
+        anchor: 201, foundOldest: true, messages: [
+          eg.streamMessage(id: 101, stream: stream, topic: 'A'),
+        ]).toJson());
+      await model.fetchOlder();
+      checkNotified(count: 2);
+      check(model.messages.map((m) => m.id))
+        .deepEquals(expected..insertAll(0, [101]));
+
+      // … and on maybeAddMessage.
+      model.maybeAddMessage(eg.streamMessage(id: 301, stream: stream, topic: 'A'));
+      checkNotifiedOnce();
+      check(model.messages.map((m) => m.id)).deepEquals(expected..add(301));
+    });
+  });
+
   test('recipient headers are maintained consistently', () async {
     // This tests the code that maintains the invariant that recipient headers
     // are present just where [canShareRecipientHeader] requires them.
@@ -772,6 +912,22 @@ void checkInvariants(MessageListView model) {
     check(model).fetchingOlder.isFalse();
   }
 
+  for (final message in model.messages) {
+    check(model.narrow.containsMessage(message)).isTrue();
+
+    if (message is! StreamMessage) continue;
+    switch (model.narrow) {
+      case AllMessagesNarrow():
+        check(model.store.isTopicVisible(message.streamId, message.subject))
+          .isTrue();
+      case StreamNarrow():
+        check(model.store.isTopicVisibleInStream(message.streamId, message.subject))
+          .isTrue();
+      case TopicNarrow():
+      case DmNarrow():
+    }
+  }
+
   for (int i = 0; i < model.messages.length - 1; i++) {
     check(model.messages[i].id).isLessThan(model.messages[i+1].id);
   }

test/widgets/action_sheet_test.dart

@@ -37,7 +37,8 @@ Future<void> setupToMessageActionSheet(WidgetTester tester, {
   final store = await testBinding.globalStore.perAccount(eg.selfAccount.id);
   store.addUser(eg.user(userId: message.senderId));
   if (message is StreamMessage) {
-    store.addStream(eg.stream(streamId: message.streamId));
+    final stream = eg.stream(streamId: message.streamId);
+    store..addStream(stream)..addSubscription(eg.subscription(stream));
   }
   final connection = store.connection as FakeApiConnection;
 

test/widgets/message_list_test.dart

@@ -214,7 +214,7 @@ void main() {
       testWidgets('show stream name in AllMessagesNarrow', (tester) async {
         await setupMessageListPage(tester,
           narrow: const AllMessagesNarrow(),
-          messages: [message], streams: [stream]);
+          messages: [message], subscriptions: [eg.subscription(stream)]);
         await tester.pump();
         check(findInMessageList('stream name')).length.equals(1);
         check(findInMessageList('topic name')).length.equals(1);
@@ -268,7 +268,7 @@ void main() {
         final stream = eg.stream(isWebPublic: false, inviteOnly: false);
         await setupMessageListPage(tester,
           messages: [eg.streamMessage(stream: stream)],
-          streams: [stream]);
+          subscriptions: [eg.subscription(stream)]);
         await tester.pump();
         check(find.descendant(
           of: find.byType(StreamMessageRecipientHeader),
@@ -280,7 +280,7 @@ void main() {
         final stream = eg.stream(isWebPublic: true);
         await setupMessageListPage(tester,
           messages: [eg.streamMessage(stream: stream)],
-          streams: [stream]);
+          subscriptions: [eg.subscription(stream)]);
         await tester.pump();
         check(find.descendant(
           of: find.byType(StreamMessageRecipientHeader),
@@ -292,7 +292,7 @@ void main() {
         final stream = eg.stream(inviteOnly: true);
         await setupMessageListPage(tester,
           messages: [eg.streamMessage(stream: stream)],
-          streams: [stream]);
+          subscriptions: [eg.subscription(stream)]);
         await tester.pump();
         check(find.descendant(
           of: find.byType(StreamMessageRecipientHeader),
@@ -304,6 +304,9 @@ void main() {
     testWidgets('show stream name from message when stream unknown', (tester) async {
       // This can perfectly well happen, because message fetches can race
       // with events.
+      // … Though not actually with AllMessagesNarrow, because that shows
+      // stream messages only in subscribed streams, hence only known streams.
+      // See skip comment below.
       final stream = eg.stream(name: 'stream name');
       await setupMessageListPage(tester,
         narrow: const AllMessagesNarrow(),
@@ -313,7 +316,7 @@ void main() {
         ]);
       await tester.pump();
       tester.widget(find.text('stream name'));
-    });
+    }, skip: true); // TODO(#252) could repro this with search narrows, once we have those
 
     testWidgets('show stream name from stream data when known', (tester) async {
       final streamBefore = eg.stream(name: 'old stream name');