diff --git a/packages/stream_chat/CHANGELOG.md b/packages/stream_chat/CHANGELOG.md index 910a7bbd3b..92725853d4 100644 --- a/packages/stream_chat/CHANGELOG.md +++ b/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 diff --git a/packages/stream_chat/lib/src/client/channel.dart b/packages/stream_chat/lib/src/client/channel.dart index 9e9906238c..3ed717da9a 100644 --- a/packages/stream_chat/lib/src/client/channel.dart +++ b/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 []; + _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)); + } })); } diff --git a/packages/stream_chat/lib/src/client/client.dart b/packages/stream_chat/lib/src/client/client.dart index a251b15ed7..670f7b2235 100644 --- a/packages/stream_chat/lib/src/client/client.dart +++ b/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( diff --git a/packages/stream_chat/lib/src/core/models/event.dart b/packages/stream_chat/lib/src/core/models/event.dart index 57d2aaeec2..4b1b672691 100644 --- a/packages/stream_chat/lib/src/core/models/event.dart +++ b/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? 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, diff --git a/packages/stream_chat/lib/src/core/models/event.g.dart b/packages/stream_chat/lib/src/core/models/event.g.dart index 7bc820b33e..863de9844c 100644 --- a/packages/stream_chat/lib/src/core/models/event.g.dart +++ b/packages/stream_chat/lib/src/core/models/event.g.dart @@ -77,6 +77,7 @@ Event _$EventFromJson(Map json) => Event( : ChannelPushPreference.fromJson( json['channel_push_preference'] as Map), channelMessageCount: (json['channel_message_count'] as num?)?.toInt(), + watcherCount: (json['watcher_count'] as num?)?.toInt(), lastDeliveredAt: json['last_delivered_at'] == null ? null : DateTime.parse(json['last_delivered_at'] as String), @@ -131,6 +132,7 @@ Map _$EventToJson(Event instance) => { 'channel_push_preference': value, if (instance.channelMessageCount case final value?) 'channel_message_count': value, + if (instance.watcherCount case final value?) 'watcher_count': value, if (instance.lastDeliveredAt?.toIso8601String() case final value?) 'last_delivered_at': value, if (instance.lastDeliveredMessageId case final value?) diff --git a/packages/stream_chat/test/fixtures/event.json b/packages/stream_chat/test/fixtures/event.json index 4bcb908693..b164b3aa59 100644 --- a/packages/stream_chat/test/fixtures/event.json +++ b/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 } \ No newline at end of file diff --git a/packages/stream_chat/test/src/client/channel_test.dart b/packages/stream_chat/test/src/client/channel_test.dart index ef499e6e2c..87ddcbcc13 100644 --- a/packages/stream_chat/test/src/client/channel_test.dart +++ b/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'; diff --git a/packages/stream_chat/test/src/client/client_test.dart b/packages/stream_chat/test/src/client/client_test.dart index 9040449f89..91ae6144b2 100644 --- a/packages/stream_chat/test/src/client/client_test.dart +++ b/packages/stream_chat/test/src/client/client_test.dart @@ -1,5 +1,7 @@ // ignore_for_file: avoid_redundant_argument_values +import 'dart:async'; + import 'package:mocktail/mocktail.dart'; import 'package:stream_chat/src/core/http/token.dart'; import 'package:stream_chat/stream_chat.dart'; @@ -4434,4 +4436,82 @@ void main() { }); }); }); + + group('dispose during reconnect recovery', () { + const apiKey = 'test-api-key'; + final user = User(id: 'test-user-id'); + final token = Token.development(user.id).rawValue; + + late FakeChatApi api; + late FakeWebSocket ws; + late StreamChatClient client; + var disposed = false; + + setUpAll(() { + registerFallbackValue(const PaginationParams()); + registerFallbackValue(Filter.equal('cid', '')); + }); + + setUp(() { + api = FakeChatApi(); + ws = FakeWebSocket(); + disposed = false; + }); + + // The test disposes the client itself; avoid disposing it a second time. + tearDown(() async { + if (!disposed) await client.dispose(); + }); + + // Disposing the client while a reconnect is still recovering must complete + // cleanly: recovery work that finishes after disposal is discarded, never + // surfacing as an error. + test('disposing mid-recovery does not surface a late recovery event', + () async { + // Keep the recovery's channel query pending so the client is still + // mid-recovery at the moment it is disposed. + final pendingQuery = Completer(); + when( + () => api.channel.queryChannels( + filter: any(named: 'filter'), + sort: any(named: 'sort'), + state: any(named: 'state'), + watch: any(named: 'watch'), + presence: any(named: 'presence'), + memberLimit: any(named: 'memberLimit'), + messageLimit: any(named: 'messageLimit'), + paginationParams: any(named: 'paginationParams'), + ), + ).thenAnswer((_) => pendingQuery.future); + + client = StreamChatClient(apiKey, chatApi: api, ws: ws); + await client.connectUser(user, token); + await delay(300); + + // Track a channel so reconnecting triggers channel recovery, which then + // blocks on the pending query above. + final channel = Channel.fromState( + client, + ChannelState(channel: ChannelModel(cid: 'messaging:c1')), + ); + client.state.addChannels({'messaging:c1': channel}); + + // Drop then restore the connection to start a reconnect recovery. + ws.connectionStatus = ConnectionStatus.disconnected; + await delay(100); + ws.connectionStatus = ConnectionStatus.connected; + await delay(100); + + // Dispose while the recovery is still in flight. + await client.dispose(); + disposed = true; + + // Let the now-orphaned recovery finish. Its trailing work must be + // discarded silently instead of thrown as an unhandled async error. + pendingQuery.complete(QueryChannelsResponse()..channels = []); + await delay(300); + + expect(client.wsConnectionStatus, ConnectionStatus.disconnected); + }); + }); } diff --git a/packages/stream_chat/test/src/core/models/event_test.dart b/packages/stream_chat/test/src/core/models/event_test.dart index f7847c4c60..ab5c78d2fd 100644 --- a/packages/stream_chat/test/src/core/models/event_test.dart +++ b/packages/stream_chat/test/src/core/models/event_test.dart @@ -19,6 +19,7 @@ void main() { expect(event.unreadThreadMessages, 2); expect(event.unreadThreads, 3); expect(event.channelLastMessageAt, isA()); + expect(event.watcherCount, 12); expect(event.lastReadAt, null); expect(event.unreadMessages, null); expect(event.lastReadMessageId, null); @@ -62,6 +63,7 @@ void main() { unreadThreadMessages: 2, unreadThreads: 3, channelLastMessageAt: DateTime.parse('2019-03-27T17:40:17.155892Z'), + watcherCount: 9, lastReadAt: DateTime.parse('2020-02-10T10:00:00.000Z'), unreadMessages: 5, lastReadMessageId: 'last-read-message-id', @@ -107,6 +109,7 @@ void main() { 'unread_thread_messages': 2, 'unread_threads': 3, 'channel_last_message_at': '2019-03-27T17:40:17.155892Z', + 'watcher_count': 9, 'last_read_at': '2020-02-10T10:00:00.000Z', 'unread_messages': 5, 'last_read_message_id': 'last-read-message-id', @@ -145,6 +148,7 @@ void main() { expect(newEvent.unreadThreadMessages, 2); expect(newEvent.unreadThreads, 3); expect(newEvent.channelLastMessageAt, isA()); + expect(newEvent.watcherCount, 12); expect(newEvent.lastReadAt, null); expect(newEvent.unreadMessages, null); expect(newEvent.lastReadMessageId, null); @@ -171,6 +175,7 @@ void main() { unreadThreadMessages: 6, unreadThreads: 7, channelLastMessageAt: DateTime.parse('2020-01-29T03:22:47.636130Z'), + watcherCount: 21, lastReadAt: DateTime.parse('2020-02-10T10:00:00.000000Z'), unreadMessages: 5, lastReadMessageId: 'last-read-message-id', @@ -196,6 +201,7 @@ void main() { DateTime.parse('2020-02-10T10:00:00.000000Z'), ); expect(newEvent.unreadMessages, 5); + expect(newEvent.watcherCount, 21); expect(newEvent.lastReadMessageId, 'last-read-message-id'); expect(newEvent.draft, isNotNull); expect(newEvent.draft, equals(draft));