fix(llc): update watcher count from realtime events and ignore events after dispose

packages/stream_chat/CHANGELOG.md

@@ -1,5 +1,9 @@
 ## Upcoming
 
+✅ Added
+
+- Added `Event.watcherCount`, exposing the server-provided `watcher_count` field on events (e.g. `user.watching.start`, `user.watching.stop`, `message.new`).
+
 🔄 Changed
 
 - Raised the minimum `rate_limiter` version to `^1.1.1`.
@@ -12,6 +16,10 @@
 - Fixed truncated channels dropping to the bottom of the list when sorting by `last_updated`.
 - Fixed pinned channels appearing at the bottom of the list when sorting by `pinned_at` descending.
 - Fixed channels without messages appearing at the top of the list when sorting by `last_message_at` descending.
+- Fixed `ChannelClientState.watcherCount` staying stale during a session.
+- Fixed watchers not being removed from `ChannelClientState.watchers` on `user.watching.stop`.
+- Fixed a `StateError` (`Cannot add new events after calling close`) thrown when the client is disposed while a reconnect recovery is still in flight.
+
 
 ## 9.27.0
 

packages/stream_chat/lib/src/client/channel.dart

@@ -2438,6 +2438,7 @@ class ChannelClientState {
               watcher,
               ...?existingWatchers?.where((user) => user.id != watcher.id),
             ],
+            watcherCount: event.watcherCount,
           ));
         }
       }),
@@ -2449,12 +2450,13 @@ class ChannelClientState {
       _channel.on(EventType.userWatchingStop).listen((event) {
         final watcher = event.user;
         if (watcher != null) {
-          final existingWatchers = channelState.watchers;
-          updateChannelState(channelState.copyWith(
-            watchers: [
-              ...?existingWatchers?.where((user) => user.id != watcher.id)
-            ],
-          ));
+          final existingWatchers = channelState.watchers ?? const <User>[];
+          _channelState = channelState.copyWith(
+            watchers: existingWatchers
+                .where((user) => user.id != watcher.id)
+                .toList(),
+            watcherCount: event.watcherCount,
+          );
         }
       }),
     );
@@ -2887,6 +2889,13 @@ class ChannelClientState {
       }
 
       _client.channelDeliveryReporter.submitForDelivery([_channel]);
+
+      // Only message.new carries a reliable watcher count;
+      // notification.message_new targets non-watchers and reports 0.
+      if (event.watcherCount case final watcherCount?
+          when event.type == EventType.messageNew) {
+        updateChannelState(channelState.copyWith(watcherCount: watcherCount));
+      }
     }));
   }
 

packages/stream_chat/lib/src/client/client.dart

@@ -38,6 +38,7 @@ import 'package:stream_chat/src/core/models/poll_vote.dart';
 import 'package:stream_chat/src/core/models/push_preference.dart';
 import 'package:stream_chat/src/core/models/thread.dart';
 import 'package:stream_chat/src/core/models/user.dart';
+import 'package:stream_chat/src/core/util/extension.dart';
 import 'package:stream_chat/src/core/util/in_flight_cache.dart';
 import 'package:stream_chat/src/core/util/utils.dart';
 import 'package:stream_chat/src/db/chat_persistence_client.dart';
@@ -540,11 +541,14 @@ class StreamChatClient {
 
   /// Method called to add a new event to the [_eventController].
   void handleEvent(Event event) {
+    // Ignore events that arrive after the client has been disposed.
+    if (_eventController.isClosed) return;
+
     if (event.type == EventType.healthCheck) {
       return _handleHealthCheckEvent(event);
     }
     state.updateUser(event.user);
-    return _eventController.add(event);
+    return _eventController.safeAdd(event);
   }
 
   void _onConnectionStatusChanged(

packages/stream_chat/lib/src/core/models/event.dart

@@ -44,6 +44,7 @@ class Event {
     this.pushPreference,
     this.channelPushPreference,
     this.channelMessageCount,
+    this.watcherCount,
     this.lastDeliveredAt,
     this.lastDeliveredMessageId,
     this.extraData = const {},
@@ -168,6 +169,12 @@ class Event {
   /// The total number of messages in the channel.
   final int? channelMessageCount;
 
+  /// The number of users currently watching the channel.
+  ///
+  /// Sent with `user.watching.start`, `user.watching.stop` and `message.new`
+  /// events, reflecting the authoritative watcher count after the change.
+  final int? watcherCount;
+
   /// The date of the last delivered message.
   final DateTime? lastDeliveredAt;
 
@@ -216,6 +223,7 @@ class Event {
     'push_preference',
     'channel_push_preference',
     'channel_message_count',
+    'watcher_count',
     'last_delivered_at',
     'last_delivered_message_id',
   ];
@@ -262,6 +270,7 @@ class Event {
     PushPreference? pushPreference,
     ChannelPushPreference? channelPushPreference,
     int? channelMessageCount,
+    int? watcherCount,
     DateTime? lastDeliveredAt,
     String? lastDeliveredMessageId,
     Map<String, Object?>? extraData,
@@ -303,6 +312,7 @@ class Event {
         channelPushPreference:
             channelPushPreference ?? this.channelPushPreference,
         channelMessageCount: channelMessageCount ?? this.channelMessageCount,
+        watcherCount: watcherCount ?? this.watcherCount,
         lastDeliveredAt: lastDeliveredAt ?? this.lastDeliveredAt,
         lastDeliveredMessageId:
             lastDeliveredMessageId ?? this.lastDeliveredMessageId,

packages/stream_chat/test/fixtures/event.json

@@ -30,5 +30,6 @@
   "ai_message": "Some message",
   "unread_thread_messages": 2,
   "unread_threads": 3,
-  "channel_last_message_at": "2019-03-27T17:40:17.155892Z"
+  "channel_last_message_at": "2019-03-27T17:40:17.155892Z",
+  "watcher_count": 12
 }

packages/stream_chat/test/src/client/channel_test.dart

@@ -5173,6 +5173,198 @@ void main() {
       );
     });
 
+    group('Watching Events', () {
+      const channelId = 'test-channel-id';
+      const channelType = 'test-channel-type';
+      late Channel channel;
+
+      setUp(() {
+        final channelState = _generateChannelState(
+          channelId,
+          channelType,
+          mockChannelConfig: true,
+          ownCapabilities: const [ChannelCapability.readEvents],
+        );
+        channel = Channel.fromState(client, channelState);
+      });
+
+      tearDown(() => channel.dispose());
+
+      test(
+        '${EventType.userWatchingStart} adds the watcher and updates watcherCount',
+        () async {
+          final watcher = User(id: 'watcher-1');
+
+          client.addEvent(
+            Event(
+              cid: channel.cid,
+              type: EventType.userWatchingStart,
+              user: watcher,
+              watcherCount: 3,
+            ),
+          );
+
+          // Wait for the event to get processed
+          await Future.delayed(Duration.zero);
+
+          expect(channel.state!.watcherCount, 3);
+          expect(
+            channel.state!.channelState.watchers?.map((it) => it.id),
+            contains('watcher-1'),
+          );
+        },
+      );
+
+      test(
+        '${EventType.userWatchingStop} removes the watcher and updates watcherCount',
+        () async {
+          final watcher = User(id: 'watcher-1');
+
+          // The watcher starts watching first (count = 2).
+          client.addEvent(
+            Event(
+              cid: channel.cid,
+              type: EventType.userWatchingStart,
+              user: watcher,
+              watcherCount: 2,
+            ),
+          );
+          await Future.delayed(Duration.zero);
+          expect(channel.state!.watcherCount, 2);
+          expect(
+            channel.state!.channelState.watchers?.map((it) => it.id),
+            contains('watcher-1'),
+          );
+
+          // Then stops watching (count = 1).
+          client.addEvent(
+            Event(
+              cid: channel.cid,
+              type: EventType.userWatchingStop,
+              user: watcher,
+              watcherCount: 1,
+            ),
+          );
+          await Future.delayed(Duration.zero);
+
+          expect(channel.state!.watcherCount, 1);
+          expect(
+            channel.state!.channelState.watchers?.map((it) => it.id),
+            isNot(contains('watcher-1')),
+          );
+        },
+      );
+
+      test(
+        'watching event without watcherCount preserves the existing count',
+        () async {
+          // Seed an initial watcher count.
+          channel.state!.updateChannelState(
+            channel.state!.channelState.copyWith(watcherCount: 5),
+          );
+          expect(channel.state!.watcherCount, 5);
+
+          // A watching event that omits watcher_count must not wipe the count.
+          client.addEvent(
+            Event(
+              cid: channel.cid,
+              type: EventType.userWatchingStart,
+              user: User(id: 'watcher-2'),
+            ),
+          );
+          await Future.delayed(Duration.zero);
+
+          expect(channel.state!.watcherCount, 5);
+          expect(
+            channel.state!.channelState.watchers?.map((it) => it.id),
+            contains('watcher-2'),
+          );
+        },
+      );
+
+      test(
+        '${EventType.messageNew} updates watcherCount from the event',
+        () async {
+          expect(channel.state!.watcherCount, isNull);
+
+          final message = Message(
+            id: 'test-message-id',
+            user: client.state.currentUser,
+            createdAt: DateTime.now(),
+          );
+
+          client.addEvent(
+            Event(
+              cid: channel.cid,
+              type: EventType.messageNew,
+              message: message,
+              watcherCount: 7,
+            ),
+          );
+          await Future.delayed(Duration.zero);
+
+          expect(channel.state!.watcherCount, 7);
+        },
+      );
+
+      test(
+        '${EventType.messageNew} without watcherCount preserves the existing count',
+        () async {
+          // Seed an initial watcher count.
+          channel.state!.updateChannelState(
+            channel.state!.channelState.copyWith(watcherCount: 4),
+          );
+          expect(channel.state!.watcherCount, 4);
+
+          // A local/optimistic message.new without watcher_count must not
+          // reset the count.
+          client.addEvent(
+            Event(
+              cid: channel.cid,
+              type: EventType.messageNew,
+              message: Message(
+                id: 'test-message-id-2',
+                user: client.state.currentUser,
+                createdAt: DateTime.now(),
+              ),
+            ),
+          );
+          await Future.delayed(Duration.zero);
+
+          expect(channel.state!.watcherCount, 4);
+        },
+      );
+
+      test(
+        '${EventType.notificationMessageNew} does not overwrite watcherCount',
+        () async {
+          // Seed a known watcher count.
+          channel.state!.updateChannelState(
+            channel.state!.channelState.copyWith(watcherCount: 5),
+          );
+          expect(channel.state!.watcherCount, 5);
+
+          // notification.message_new is delivered to non-watchers and reports
+          // watcher_count: 0; it must not clobber the real count.
+          client.addEvent(
+            Event(
+              cid: channel.cid,
+              type: EventType.notificationMessageNew,
+              message: Message(
+                id: 'notif-message-id',
+                user: User(id: 'other-user'),
+                createdAt: DateTime.now(),
+              ),
+              watcherCount: 0,
+            ),
+          );
+          await Future.delayed(Duration.zero);
+
+          expect(channel.state!.watcherCount, 5);
+        },
+      );
+    });
+
     group('Read Events', () {
       const channelId = 'test-channel-id';
       const channelType = 'test-channel-type';