diff --git a/packages/stream_chat_flutter_core/CHANGELOG.md b/packages/stream_chat_flutter_core/CHANGELOG.md index f88ba7b602..1e32081584 100644 --- a/packages/stream_chat_flutter_core/CHANGELOG.md +++ b/packages/stream_chat_flutter_core/CHANGELOG.md @@ -13,6 +13,7 @@ - Fixed `StreamDraftListController.deleteDraft` removing the wrong draft. - Fixed `StreamPollVoteListController` adding votes and comments from other polls in the same channel. +- Fixed `StreamChannelListController` querying channels again for every new message in a watched channel its filter excludes, which could hit the API rate limit; once a refresh leaves such a channel out, further messages in it refresh the list again only after a backoff of 30 seconds that doubles up to 10 minutes. ## 10.5.0 diff --git a/packages/stream_chat_flutter_core/lib/src/stream_channel_list_event_handler.dart b/packages/stream_chat_flutter_core/lib/src/stream_channel_list_event_handler.dart index 7cff5f488a..dcb7a48c7f 100644 --- a/packages/stream_chat_flutter_core/lib/src/stream_channel_list_event_handler.dart +++ b/packages/stream_chat_flutter_core/lib/src/stream_channel_list_event_handler.dart @@ -1,4 +1,6 @@ +import 'package:clock/clock.dart'; import 'package:stream_chat/stream_chat.dart' show Event, Member; + import 'stream_channel_list_controller.dart'; /// Contains handlers that are called from [StreamChannelListController] for @@ -6,6 +8,12 @@ import 'stream_channel_list_controller.dart'; /// /// This class can be mixed in or extended to create custom overrides. mixin class StreamChannelListEventHandler { + final _missingChannelRefreshers = Expando<_MissingChannelRefresher>(); + + _MissingChannelRefresher _missingChannelRefresherFor(StreamChannelListController controller) { + return _missingChannelRefreshers[controller] ??= _MissingChannelRefresher(controller); + } + /// Function which gets called for the event /// [EventType.channelDeleted]. /// @@ -111,6 +119,7 @@ mixin class StreamChannelListEventHandler { Event event, StreamChannelListController controller, ) { + _missingChannelRefreshers[controller]?.reset(); controller.refresh(); } @@ -120,20 +129,27 @@ mixin class StreamChannelListEventHandler { /// we are currently watching. /// /// By default, this moves the channel to the top of the list. + /// + /// When the channel is not in the list, for example because it was hidden + /// and the new message made it visible again, this refreshes the list + /// without resetting it. If the refreshed list still does not contain the + /// channel, the list's query is assumed to exclude it, and further messages + /// in that channel do not refresh the list again until a backoff of 30 + /// seconds has passed. The backoff doubles on each refresh that still leaves + /// the channel out, up to 10 minutes, and is cleared when the connection + /// recovers. At most one refresh runs at a time; messages arriving meanwhile + /// share a single follow-up refresh. void onMessageNew(Event event, StreamChannelListController controller) async { final channelCid = event.cid; if (channelCid == null) return; final channels = [...controller.currentItems]; + final missingChannelRefresher = _missingChannelRefresherFor(controller); final channelIndex = channels.indexWhere((it) => it.cid == channelCid); - if (channelIndex < 0) { - // If the channel is not in the list, It might be hidden. - // So, we just refresh the list. - await controller.refresh(resetValue: false); - return; - } + if (channelIndex < 0) return missingChannelRefresher.refreshFor(channelCid); + missingChannelRefresher.forget(channelCid); final channel = channels.removeAt(channelIndex); channels.insert(0, channel); @@ -238,3 +254,82 @@ mixin class StreamChannelListEventHandler { controller.channels = [...channels]; } } + +// Refreshes a channel list on behalf of channels that received a message but +// are not listed. +// +// A refresh that comes back without the channel means the list's query +// excludes it, so the channel is backed off: further requests for it are +// ignored until the backoff expires, and each refresh that still leaves it out +// doubles the backoff. Requests made while a refresh runs share one trailing +// refresh, so at most one query is in flight per controller. +class _MissingChannelRefresher { + _MissingChannelRefresher(this._controller); + + static const _initialBackoff = Duration(seconds: 30); + static const _maxBackoff = Duration(minutes: 10); + + final StreamChannelListController _controller; + final _backoffs = {}; + final _pendingCids = {}; + Future? _inFlight; + Future? _trailing; + + Future refreshFor(String cid) async { + if (_pendingCids.contains(cid)) return; + + final backoff = _backoffs[cid]; + if (backoff != null && clock.now().isBefore(backoff.expiresAt)) return; + + _pendingCids.add(cid); + try { + await _scheduleRefresh(); + } finally { + _pendingCids.remove(cid); + } + + if (_controller.currentItems.any((it) => it.cid == cid)) { + _backoffs.remove(cid); + return; + } + + final duration = _nextBackoff(backoff); + _backoffs[cid] = _Backoff(duration: duration, expiresAt: clock.now().add(duration)); + } + + Duration _nextBackoff(_Backoff? previous) { + if (previous == null) return _initialBackoff; + final doubled = previous.duration * 2; + return doubled > _maxBackoff ? _maxBackoff : doubled; + } + + // Drops the backoff for [cid], for when the channel shows up in the list. + void forget(String cid) => _backoffs.remove(cid); + + // Drops every backoff, for when the list may have changed in ways no event + // reported. + void reset() => _backoffs.clear(); + + Future _scheduleRefresh() { + if (_trailing case final trailing?) return trailing; + + final inFlight = _inFlight; + if (inFlight == null) return _refresh(); + + return _trailing = inFlight.then((_) { + _trailing = null; + return _refresh(); + }); + } + + Future _refresh() { + return _inFlight = _controller.refresh(resetValue: false).whenComplete(() => _inFlight = null); + } +} + +class _Backoff { + const _Backoff({required this.duration, required this.expiresAt}); + + final Duration duration; + final DateTime expiresAt; +} diff --git a/packages/stream_chat_flutter_core/pubspec.yaml b/packages/stream_chat_flutter_core/pubspec.yaml index 71330760e9..9b78f20e3b 100644 --- a/packages/stream_chat_flutter_core/pubspec.yaml +++ b/packages/stream_chat_flutter_core/pubspec.yaml @@ -22,6 +22,7 @@ environment: flutter: ">=3.44.0" dependencies: + clock: ^1.1.2 collection: ^1.19.1 connectivity_plus: ^7.1.1 device_info_plus: ">=12.4.0 <14.0.0" diff --git a/packages/stream_chat_flutter_core/test/stream_channel_list_controller_message_new_test.dart b/packages/stream_chat_flutter_core/test/stream_channel_list_controller_message_new_test.dart new file mode 100644 index 0000000000..05c10b6a96 --- /dev/null +++ b/packages/stream_chat_flutter_core/test/stream_channel_list_controller_message_new_test.dart @@ -0,0 +1,67 @@ +import 'dart:async'; + +import 'package:flutter_test/flutter_test.dart'; +import 'package:mocktail/mocktail.dart'; +import 'package:stream_chat/stream_chat.dart' hide Success; +import 'package:stream_chat_flutter_core/src/stream_channel_list_controller.dart'; + +import 'mocks.dart'; + +void main() { + setUpAll(() { + registerFallbackValue(const PaginationParams()); + }); + + test('queries channels once for a burst of messages in a watched channel the filter excludes', () async { + final client = MockClient(); + final events = StreamController.broadcast(); + addTearDown(events.close); + when(client.on).thenAnswer((_) => events.stream); + _stubQueryChannels(client, channels: [_mockChannel('messaging:listed')]); + final controller = StreamChannelListController(client: client, channelStateSort: null); + addTearDown(controller.dispose); + await controller.doInitialLoad(); + clearInteractions(client); + + for (var i = 0; i < 50; i++) { + events.add(Event(type: EventType.messageNew, cid: 'livestream:excluded')); + await pumpEventQueue(); + } + + verify( + () => client.queryChannelsWithResult( + filter: any(named: 'filter'), + channelStateSort: any(named: 'channelStateSort'), + predefinedFilter: any(named: 'predefinedFilter'), + filterValues: any(named: 'filterValues'), + sortValues: any(named: 'sortValues'), + memberLimit: any(named: 'memberLimit'), + messageLimit: any(named: 'messageLimit'), + presence: any(named: 'presence'), + paginationParams: any(named: 'paginationParams'), + ), + ).called(1); + }); +} + +MockChannel _mockChannel(String cid) { + final channel = MockChannel(); + when(() => channel.cid).thenReturn(cid); + return channel; +} + +void _stubQueryChannels(MockClient client, {required List channels}) { + when( + () => client.queryChannelsWithResult( + filter: any(named: 'filter'), + channelStateSort: any(named: 'channelStateSort'), + predefinedFilter: any(named: 'predefinedFilter'), + filterValues: any(named: 'filterValues'), + sortValues: any(named: 'sortValues'), + memberLimit: any(named: 'memberLimit'), + messageLimit: any(named: 'messageLimit'), + presence: any(named: 'presence'), + paginationParams: any(named: 'paginationParams'), + ), + ).thenAnswer((_) => Stream.value(QueryChannelsResult(channels: channels))); +} diff --git a/packages/stream_chat_flutter_core/test/stream_channel_list_event_handler_message_new_test.dart b/packages/stream_chat_flutter_core/test/stream_channel_list_event_handler_message_new_test.dart new file mode 100644 index 0000000000..0ea968e96b --- /dev/null +++ b/packages/stream_chat_flutter_core/test/stream_channel_list_event_handler_message_new_test.dart @@ -0,0 +1,196 @@ +import 'dart:async'; + +import 'package:fake_async/fake_async.dart'; +import 'package:flutter_test/flutter_test.dart'; +import 'package:mocktail/mocktail.dart'; +import 'package:stream_chat/stream_chat.dart' hide Success; +import 'package:stream_chat_flutter_core/src/stream_channel_list_event_handler.dart'; + +import 'mocks.dart'; + +void main() { + group('when the message is in a channel the refreshed list leaves out', () { + test('refreshes the list once for many messages', () { + fakeAsync((async) { + final list = _FakeChannelList([_mockChannel('messaging:listed')]); + final handler = StreamChannelListEventHandler(); + + for (var i = 0; i < 20; i++) { + _receiveMessage(async, handler, list, 'livestream:excluded'); + } + + expect(list.refreshCount, 1); + }); + }); + + test('refreshes the list again once the backoff has passed', () { + fakeAsync((async) { + final list = _FakeChannelList([_mockChannel('messaging:listed')]); + final handler = StreamChannelListEventHandler(); + + _receiveMessage(async, handler, list, 'livestream:excluded'); + async.elapse(const Duration(seconds: 30)); + _receiveMessage(async, handler, list, 'livestream:excluded'); + + expect(list.refreshCount, 2); + }); + }); + + test('doubles the backoff after each refresh that still leaves the channel out', () { + fakeAsync((async) { + final list = _FakeChannelList([_mockChannel('messaging:listed')]); + final handler = StreamChannelListEventHandler(); + + _receiveMessage(async, handler, list, 'livestream:excluded'); + async.elapse(const Duration(seconds: 30)); + _receiveMessage(async, handler, list, 'livestream:excluded'); + async.elapse(const Duration(seconds: 59)); + _receiveMessage(async, handler, list, 'livestream:excluded'); + final refreshCountWithinBackoff = list.refreshCount; + async.elapse(const Duration(seconds: 1)); + _receiveMessage(async, handler, list, 'livestream:excluded'); + + expect(refreshCountWithinBackoff, 2); + expect(list.refreshCount, 3); + }); + }); + + test('caps the backoff at 10 minutes', () { + fakeAsync((async) { + final list = _FakeChannelList([_mockChannel('messaging:listed')]); + final handler = StreamChannelListEventHandler(); + + // The backoff grows 30 s → 1 → 2 → 4 → 8 min; without the cap the sixth + // refresh would back off 16 minutes and skip the seventh message. + for (var i = 0; i < 7; i++) { + _receiveMessage(async, handler, list, 'livestream:excluded'); + async.elapse(const Duration(minutes: 10)); + } + + expect(list.refreshCount, 7); + }); + }); + + test('still refreshes the list for a message in another missing channel', () { + fakeAsync((async) { + final list = _FakeChannelList([_mockChannel('messaging:listed')]); + final handler = StreamChannelListEventHandler(); + + _receiveMessage(async, handler, list, 'livestream:excluded'); + _receiveMessage(async, handler, list, 'messaging:other'); + + expect(list.refreshCount, 2); + }); + }); + + test('refreshes the list again after the connection recovers', () { + fakeAsync((async) { + final list = _FakeChannelList([_mockChannel('messaging:listed')]); + final handler = StreamChannelListEventHandler(); + + _receiveMessage(async, handler, list, 'livestream:excluded'); + handler.onConnectionRecovered(Event(type: EventType.connectionRecovered), list.controller); + _receiveMessage(async, handler, list, 'livestream:excluded'); + + expect(list.refreshCount, 2); + }); + }); + }); + + test('refreshes the list for a channel that was listed and then removed', () { + fakeAsync((async) { + final listed = _mockChannel('messaging:listed'); + final hidden = _mockChannel('messaging:hidden'); + final list = _FakeChannelList([listed], channelsAfterRefresh: [hidden, listed]); + final handler = StreamChannelListEventHandler(); + + _receiveMessage(async, handler, list, 'messaging:hidden'); + list.items = [listed]; + _receiveMessage(async, handler, list, 'messaging:hidden'); + + expect(list.refreshCount, 2); + }); + }); + + test('refreshes the list for a backed-off channel that was listed in the meantime and then removed', () { + fakeAsync((async) { + final listed = _mockChannel('messaging:listed'); + final hidden = _mockChannel('messaging:hidden'); + final list = _FakeChannelList([listed]); + final handler = StreamChannelListEventHandler(); + + _receiveMessage(async, handler, list, 'messaging:hidden'); + list.items = [hidden, listed]; + _receiveMessage(async, handler, list, 'messaging:hidden'); + list.items = [listed]; + _receiveMessage(async, handler, list, 'messaging:hidden'); + + expect(list.refreshCount, 2); + }); + }); + + test('shares one trailing refresh between messages arriving while a refresh runs', () async { + final list = _FakeChannelList([_mockChannel('messaging:listed')], completeRefreshesManually: true); + final handler = StreamChannelListEventHandler(); + + for (final cid in ['messaging:first', 'messaging:second', 'messaging:third']) { + handler.onMessageNew(_messageNew(cid), list.controller); + } + await pumpEventQueue(); + list.completeRefresh(); + await pumpEventQueue(); + list.completeRefresh(); + await pumpEventQueue(); + + expect(list.refreshCount, 2); + }); +} + +// Delivers a `message.new` event for [cid] and lets the handler finish with it. +void _receiveMessage( + FakeAsync async, + StreamChannelListEventHandler handler, + _FakeChannelList list, + String cid, +) { + handler.onMessageNew(_messageNew(cid), list.controller); + async.flushMicrotasks(); +} + +Event _messageNew(String cid) => Event(type: EventType.messageNew, cid: cid); + +MockChannel _mockChannel(String cid) { + final channel = MockChannel(); + when(() => channel.cid).thenReturn(cid); + return channel; +} + +// A mocked channel list whose items change only when the test says so, and +// which counts the refreshes it is asked for. +class _FakeChannelList { + _FakeChannelList( + this.items, { + List? channelsAfterRefresh, + bool completeRefreshesManually = false, + }) { + when(() => controller.currentItems).thenAnswer((_) => items); + when(controller.refresh).thenAnswer((_) async {}); + when(() => controller.refresh(resetValue: false)).thenAnswer((_) { + refreshCount += 1; + if (completeRefreshesManually) { + final completer = Completer(); + _pendingRefreshes.add(completer); + return completer.future; + } + if (channelsAfterRefresh != null) items = channelsAfterRefresh; + return Future.value(); + }); + } + + final controller = MockStreamChannelListController(); + final _pendingRefreshes = >[]; + List items; + int refreshCount = 0; + + void completeRefresh() => _pendingRefreshes.removeAt(0).complete(); +}