diff --git a/packages/stream_chat/CHANGELOG.md b/packages/stream_chat/CHANGELOG.md index 68347485d..f57de6c14 100644 --- a/packages/stream_chat/CHANGELOG.md +++ b/packages/stream_chat/CHANGELOG.md @@ -1,7 +1,18 @@ ## Upcoming +✅ Added + +- Added a getter for `StreamChatClient.recoverStateOnReconnect`, which was previously write-only. + +🔄 Changed + +- Reconnecting no longer replays very large event backlogs; the offline cache is reset and the affected channels are re-queried instead, so a long spell offline does not stall the app on reconnect. + 🐞 Fixed +- Fixed reconnecting with more than 255 channels clearing the offline cache and skipping the events missed while offline. +- Fixed reconnect recovery refreshing only the first 30 active channels. +- Fixed reconnect catch-up covering an arbitrary subset of channels when more are active than one request holds; the most recently active are now covered first. - Fixed `CurrentPlatform` throwing `UnimplementedError` on WebAssembly builds. - Fixed live location expiry emitting repeated `location.expired` events for the same expired location. diff --git a/packages/stream_chat/lib/src/client/client.dart b/packages/stream_chat/lib/src/client/client.dart index b979af801..25a641b04 100644 --- a/packages/stream_chat/lib/src/client/client.dart +++ b/packages/stream_chat/lib/src/client/client.dart @@ -1,6 +1,5 @@ import 'dart:async'; -import 'package:collection/collection.dart'; import 'package:dio/dio.dart'; import 'package:logging/logging.dart'; import 'package:meta/meta.dart'; @@ -60,6 +59,7 @@ import 'event_resolvers.dart' as event_resolvers; import 'live_location_expiration_scheduler.dart'; import 'query_channels_result.dart'; import 'retry_policy.dart'; +import 'sync_manager.dart'; /// Handler function used for logging records. Function requires a single /// [LogRecord] as the only parameter. @@ -102,7 +102,7 @@ class StreamChatClient { AttachmentFileUploaderProvider attachmentFileUploaderProvider = StreamAttachmentFileUploader.new, Iterable? chatApiInterceptors, HttpClientAdapter? httpClientAdapter, - this._recoverStateOnReconnect = true, + this.recoverStateOnReconnect = true, this.isLocalUnreadCountEnabled = false, }) { logger.info('Initiating new StreamChatClient'); @@ -245,17 +245,18 @@ class StreamChatClient { /// Whether the client should automatically refresh local state from the /// server when the WebSocket connection recovers. /// - /// When `true` (default), the client re-queries the active channels on - /// reconnect (capped at 30, ordered by `state.channels.keys`). The set of - /// state recovered on reconnect may grow in the future to cover threads, - /// reminders, etc. + /// When `true` (default), the client re-queries the channels that were + /// active before the connection was lost. /// /// Setting this to `false` disables that client-level recovery. Consumers /// that opt out are responsible for refreshing their own state when the /// [EventType.connectionRecovered] event fires — for example, by re-running /// their channel list query. - set recoverStateOnReconnect(bool value) => _recoverStateOnReconnect = value; - bool _recoverStateOnReconnect; + /// + /// Replaying the events missed while offline is not affected either way: it + /// runs whenever a persistence client is connected, and the channels it + /// cannot replay are refreshed regardless of this flag. + bool recoverStateOnReconnect; /// By default the Chat client will write all messages with level Warn or /// Error to stdout. @@ -588,6 +589,12 @@ class StreamChatClient { return _eventController.safeAdd(event); } + late final _syncManager = SyncManager( + client: this, + logger: logger, + fetchMissedEvents: _chatApi.general.sync, + ); + void _onConnectionStatusChanged( ConnectionStatus prevStatus, ConnectionStatus currStatus, @@ -599,46 +606,13 @@ class StreamChatClient { final isConnected = currStatus == ConnectionStatus.connected; // Notify the connection status change event - handleEvent( - Event( - type: EventType.connectionChanged, - online: isConnected, - ), - ); + handleEvent(Event(type: EventType.connectionChanged, online: isConnected)); final connectionRecovered = !wasConnected && isConnected; + if (!connectionRecovered) return; - if (connectionRecovered) { - // connection recovered - final cids = [...state.channels.keys.toSet()]; - if (cids.isNotEmpty) { - // Recovery is best-effort: the connection can drop again while it is - // in flight. Nothing awaits this method, so an error here would - // surface as an unhandled crash instead of reaching the app. - try { - // Sync the persistence client if available - if (persistenceEnabled) await sync(cids: cids); - - // Recover the channels that were active before the connection was lost, - // only if the client is configured to do so. - if (_recoverStateOnReconnect) { - await queryChannelsOnline( - filter: Filter.in_('cid', cids), - paginationParams: const PaginationParams(limit: 30), - ); - } - } catch (e, stk) { - logger.warning('Error recovering state on reconnect', e, stk); - } - } - - handleEvent( - Event( - type: EventType.connectionRecovered, - online: true, - ), - ); - } + await _syncManager.recoverState(); + handleEvent(Event(type: EventType.connectionRecovered, online: true)); } /// Stream of [Event] coming from [_ws] connection @@ -656,60 +630,19 @@ class StreamChatClient { ); } - // Lock to make sure only one sync process is running at a time. - final _syncLock = Lock(); - - /// Get the events missed while offline to sync the offline storage - /// Will automatically fetch [cids] and [lastSyncedAt] if [persistenceEnabled] + /// Replays the events missed while offline, applying them to client state and + /// to the offline storage. + /// + /// [cids] and [lastSyncAt] both fall back to the values held by the + /// persistence client when omitted. + /// + /// A window that cannot be replayed — too many events, or refused by the + /// server — is given up on, and the channels it covered are re-queried in its + /// place. + /// + /// Never throws: a failed catch-up is logged and left for the next one. Future sync({List? cids, DateTime? lastSyncAt}) { - return _syncLock.synchronized(() async { - final channels = cids ?? await chatPersistenceClient?.getChannelCids(); - if (channels == null || channels.isEmpty) return; - - final syncAt = lastSyncAt ?? await chatPersistenceClient?.getLastSyncAt(); - if (syncAt == null) { - logger.info('Fresh sync start: lastSyncAt initialized to now.'); - return chatPersistenceClient?.updateLastSyncAt(DateTime.now()); - } - - try { - logger.info('Syncing events since $syncAt for channels: $channels'); - - final res = await _chatApi.general.sync(channels, syncAt); - final events = res.events.sorted( - (a, b) => a.createdAt.compareTo(b.createdAt), - ); - - for (final event in events) { - logger.fine('Syncing event: ${event.type}'); - handleEvent(event); - } - - final updatedSyncAt = events.lastOrNull?.createdAt ?? DateTime.now(); - return await chatPersistenceClient?.updateLastSyncAt(updatedSyncAt); - } catch (error, stk) { - // If we got a 400 error, it means that either the sync time is too - // old or the channel list is too long or too many events need to be - // synced. In this case, we should just flush the persistence client - // and start over. - if (error is StreamChatNetworkError && error.statusCode == 400) { - logger.warning( - 'Failed to sync events due to stale or oversized state. ' - 'Resetting the persistence client to enable a fresh start.', - ); - - try { - await chatPersistenceClient?.flush(); - return await chatPersistenceClient?.updateLastSyncAt(DateTime.now()); - } catch (resetError, resetStk) { - logger.warning('Error resetting the persistence client', resetError, resetStk); - return; - } - } - - logger.warning('Error syncing events', error, stk); - } - }); + return _syncManager.sync(cids: cids, lastSyncAt: lastSyncAt); } final _queryChannelsCache = InFlightCache(); diff --git a/packages/stream_chat/lib/src/client/sync_manager.dart b/packages/stream_chat/lib/src/client/sync_manager.dart new file mode 100644 index 000000000..3657d9dd8 --- /dev/null +++ b/packages/stream_chat/lib/src/client/sync_manager.dart @@ -0,0 +1,310 @@ +import 'package:clock/clock.dart'; +import 'package:collection/collection.dart'; +import 'package:logging/logging.dart'; +import 'package:meta/meta.dart'; +import 'package:synchronized/synchronized.dart'; + +import '../core/api/requests.dart'; +import '../core/api/responses.dart'; +import '../core/error/error.dart'; +import '../core/models/event.dart'; +import '../core/models/filter.dart'; +import '../db/chat_persistence_client.dart'; +import 'client.dart'; + +/// Fetches the events missed on [cids] since [lastSyncAt]. +typedef FetchMissedEvents = Future Function(List cids, DateTime lastSyncAt); + +/// Catches a client up on the state it missed while it was disconnected. +/// +/// Created and driven by `StreamChatClient`; not intended to be constructed +/// directly. +class SyncManager { + /// Instantiate a new SyncManager object. + SyncManager({ + required this.client, + required this.fetchMissedEvents, + this.logger, + @visibleForTesting this.maxReplayEvents = _defaultMaxReplayEvents, + }); + + // The endpoint rejects more than 255, counted before duplicates collapse. + // Well under it because the 2000-event ceiling is shared across every channel + // asked about rather than applied per channel, so fewer channels buys more + // events each before the window is refused — and a refusal drops the store. + // + // Channels past the cap are left to [recoverState]; a direct [sync] leaves + // them as they were. + static const _maxSyncCids = 100; + + // The endpoint returns up to 2000 events. Replaying that many runs a state + // update and a persistence write for each, on the reconnect path, while the + // app is trying to render. + static const _defaultMaxReplayEvents = 250; + + // A `queryChannels` response holds at most 30 channels. + static const _channelPageSize = 30; + + /// The client this manager catches up. + final StreamChatClient client; + + /// Fetches the missed events, passed separately because the client does not + /// expose the endpoint itself. + final FetchMissedEvents fetchMissedEvents; + + /// The logger associated to this manager. + final Logger? logger; + + /// How many events may be replayed from one window before it is given up on + /// and its channels refreshed instead. + final int maxReplayEvents; + + // Only one catch-up runs at a time. + final _syncLock = Lock(); + + ChatPersistenceClient? get _store => client.chatPersistenceClient; + + // Records how far a catch-up got. + // + // Swallows a write failure: there is nowhere else to keep the checkpoint, and + // a catch-up that applied its events but could not write it down is not a + // failed catch-up. + Future _recordLastSyncAt(DateTime to) async { + try { + await _store?.updateLastSyncAt(to); + } catch (error, stk) { + logger?.warning('Failed to record lastSyncAt as $to', error, stk); + } + } + + // Drops everything the local store holds, lastSyncAt included, so every + // caller has to write the checkpoint afterwards. + // + // Reports whether it worked rather than throwing: a store that could not be + // dropped still holds the state the discarded events would have updated, so + // a caller may not want to advance past them. The refresh that repopulates + // it still has to run either way. + Future _flushStore() async { + try { + await _store?.flush(); + return true; + } catch (error, stk) { + logger?.warning('Failed to reset the persistence client', error, stk); + return false; + } + } + + /// Replays the events missed since [lastSyncAt] for [cids], both falling back + /// to the values held by the persistence client. + /// + /// A window that cannot be replayed — too many events, or refused by the + /// server — is given up on, and the channels it covered are refreshed in its + /// place. Does nothing when there are no channels to catch up on. + /// + /// Those channels are returned so a caller doing its own recovery can skip + /// them; the set is empty when the events were replayed normally. + /// + /// Never throws: a failed catch-up is logged and left for the next one. + Future> sync({List? cids, DateTime? lastSyncAt}) { + return _syncLock.synchronized(() async { + final List? channelCids; + final DateTime? syncAt; + try { + channelCids = cids ?? await _store?.getChannelCids(); + syncAt = lastSyncAt ?? await _store?.getLastSyncAt(); + } catch (error, stk) { + // Not treated as "never synced": seeding lastSyncAt off an unreadable + // store would discard a history that is still there. + logger?.warning('Could not read where the last catch-up left off', error, stk); + return const {}; + } + + if (channelCids == null || channelCids.isEmpty) return const {}; + + if (syncAt == null) { + final now = clock.now(); + logger?.info('Fresh sync start: lastSyncAt initialized to $now.'); + await _recordLastSyncAt(now); + return const {}; + } + + return _performSync(channelCids, syncAt); + }); + } + + /// Recovers the state of the channels that were active before the connection + /// was lost, re-querying the ones the replay did not already refresh. + /// + /// Best-effort and never throws: the connection can drop again while this is + /// in flight, and what did not recover is left for the next reconnect. + Future recoverState() async { + final cids = _sortActiveCidsByRecency(); + if (cids.isEmpty) return; + + // A failed replay reports no refreshed channels rather than throwing, so the + // refresh below still runs — it needs the network, not the local store. + // Guarded so the contract above holds: `persistenceEnabled` reads a + // user-supplied persistence client, and replaying an event runs the whole + // event pipeline. A throw here would stop the caller announcing recovery. + try { + var refreshed = const {}; + if (client.persistenceEnabled) refreshed = await sync(cids: cids); + + if (client.recoverStateOnReconnect) { + final stale = cids.whereNot(refreshed.contains).toList(); + if (stale.isNotEmpty) await _refreshPages(stale); + } + } catch (error, stk) { + logger?.warning('Error recovering state on reconnect', error, stk); + } + } + + // Stands in for the recency of a channel whose state was never loaded, so it + // sorts behind every channel that has one. + static final _neverActive = DateTime.fromMillisecondsSinceEpoch(0); + + // Sorts the channels held in memory, most recently active first. + // + // Ordered to match the cids the persistence client hands back, so the cap in + // [_performSync] keeps the most recently active channels whichever path the + // list arrived by rather than whichever ones a query paged through first. + // + // Recency is read off the channel state rather than through the date getters + // on `Channel`, which throw for one that was never initialized or has since + // been disposed. [recoverState] must not throw. + List _sortActiveCidsByRecency() { + final byRecency = client.state.channels.entries.sortedByCompare( + (it) => it.value.state?.channelState.channel?.lastUpdatedAt ?? _neverActive, + (a, b) => b.compareTo(a), + ); + + return byRecency.map((it) => it.key).toList(); + } + + // Refreshes [cids] a page at a time, so a set larger than one request is + // covered in full rather than truncated. + // + // Every page is attempted: one failing says nothing about the others. Returns + // the channels the server answered with — fewer when one is deleted or no + // longer visible, which is not a failure — and the first request that failed. + Future<(Set, (Object, StackTrace)?)> _refreshPages(List cids) async { + final refreshed = {}; + (Object, StackTrace)? failure; + + for (final page in cids.slices(_channelPageSize)) { + try { + final channels = await client.queryChannelsOnline( + filter: Filter.in_('cid', page), + paginationParams: PaginationParams(limit: page.length), + // Fail fast if the connection dropped again: the reconnect handler is + // waiting to announce recovery, and a sync would hold its lock. + waitForConnect: false, + ); + + refreshed.addAll(channels.map((it) => it.cid).nonNulls); + } catch (error, stk) { + logger?.warning('Failed to refresh ${page.length} channels', error, stk); + failure ??= (error, stk); + } + } + + return (refreshed, failure); + } + + // Asks for the window [lastSyncAt] opens on [cids] and applies what comes + // back. [sync] has already decided there is a window worth asking for, and + // holds the lock while this runs. + Future> _performSync(List cids, DateTime lastSyncAt) async { + // Deduplicated before capping: the endpoint counts duplicates against its + // own limit, so leaving them in would spend slots on nothing. + final cappedCids = cids.toSet().take(_maxSyncCids).toList(); + logger?.info('Syncing events since $lastSyncAt for channels: $cappedCids'); + + final List events; + try { + final res = await fetchMissedEvents(cappedCids, lastSyncAt); + // lastSyncAt becomes the newest event's date, so the order has to be ours. + events = res.events.sortedBy((it) => it.createdAt); + } catch (error, stk) { + // A 400 means the window is too old, or held too many events to return. + // The two are indistinguishable, and either way the server refusing it is + // the signal that local state is too far behind to reconcile — so the + // store is dropped and repopulated rather than reconciled. + if (error is StreamChatNetworkError && error.statusCode == 400) { + logger?.warning('Resetting local state after a refused window', error, stk); + return _discardRefusedWindow(cappedCids, to: clock.now()); + } + + // Anything else could succeed next time, so lastSyncAt stays put. + logger?.warning('Error syncing events', error, stk); + return const {}; + } + + final nextSyncAt = events.lastOrNull?.createdAt ?? clock.now(); + if (events.length > maxReplayEvents) { + logger?.warning('Skipping replay of ${events.length} events, over the $maxReplayEvents limit.'); + return _discardOversizedWindow(cappedCids, from: lastSyncAt, to: nextSyncAt); + } + + // Applying an event runs the whole event pipeline, so a listener can throw. + // lastSyncAt stays where it is when one does: the window was only partly + // applied, and the next catch-up should ask for it again. + try { + for (final event in events) { + logger?.fine('Syncing event: ${event.type}'); + client.handleEvent(event); + } + } catch (error, stk) { + logger?.warning('Stopped replaying the missed events, keeping lastSyncAt', error, stk); + return const {}; + } + + await _recordLastSyncAt(nextSyncAt); + return const {}; + } + + // Gives up on a window that arrived but held more events than may be replayed. + // + // The store is dropped and repopulated from the refresh, since what it holds + // is missing every change those events carried. lastSyncAt moves to [to] only + // once both of those worked: a channel left unrefreshed, or a store that + // would not drop, still needs the events being discarded, so the checkpoint + // goes back to [from] and the next reconnect asks for the window again. + Future> _discardOversizedWindow(List cids, {required DateTime from, required DateTime to}) async { + final flushed = await _flushStore(); + final (refreshed, failure) = await _refreshPages(cids); + + // Refreshed channels are reported even on failure, so the caller does not + // query them again. + if (!flushed || failure != null) { + logger?.warning( + 'Putting lastSyncAt back: store dropped: $flushed, ' + 'refreshed ${refreshed.length} of ${cids.length} channels', + ); + await _recordLastSyncAt(from); + return refreshed; + } + + await _recordLastSyncAt(to); + return refreshed; + } + + // Gives up on a window the server would not serve. + // + // The store is dropped and repopulated from the refresh, as it is for an + // oversized window. Only the checkpoint differs: it has nowhere to go back + // to, and moves to [to] whatever the flush and the refresh did, because a + // window refused for what it is would be refused again on every reconnect + // for as long as it is held. + // + // [to] is taken before the repopulation, so anything arriving during it is + // asked for again rather than skipped. + Future> _discardRefusedWindow(List cids, {required DateTime to}) async { + // A failed flush or page is already logged, and changes nothing here: the + // checkpoint advances either way. + await _flushStore(); + final (refreshed, _) = await _refreshPages(cids); + await _recordLastSyncAt(to); + return refreshed; + } +} diff --git a/packages/stream_chat/pubspec.yaml b/packages/stream_chat/pubspec.yaml index 334c69fde..8f95a7ebc 100644 --- a/packages/stream_chat/pubspec.yaml +++ b/packages/stream_chat/pubspec.yaml @@ -22,6 +22,7 @@ environment: dependencies: async: ^2.13.1 + clock: ^1.1.2 collection: ^1.19.1 diacritic: ^0.1.6 dio: ^5.11.0 diff --git a/packages/stream_chat/test/src/client/client_test.dart b/packages/stream_chat/test/src/client/client_test.dart index d5dc7a52f..8f2f4483f 100644 --- a/packages/stream_chat/test/src/client/client_test.dart +++ b/packages/stream_chat/test/src/client/client_test.dart @@ -5206,6 +5206,11 @@ void main() { ); group('Sync Method Tests', () { + setUpAll(() { + registerFallbackValue(const PaginationParams()); + registerFallbackValue(Filter.equal('cid', '')); + }); + test( 'should retrieve data from persistence client and sync successfully', () async { @@ -5267,6 +5272,115 @@ void main() { verify(() => api.general.sync(cids, lastSyncAt)).called(1); }); + + test( + '''should replay events and advance lastSyncAt when the payload is within the replay limit''', + () async { + final cids = ['channel1']; + final lastSyncAt = DateTime.now().subtract(const Duration(hours: 1)); + final fakeClient = FakePersistenceClient( + channelCids: cids, + lastSyncAt: lastSyncAt, + ); + + client.chatPersistenceClient = fakeClient; + final events = List.generate( + 10, + (index) => Event( + type: EventType.messageNew, + cid: 'channel1', + message: Message(id: 'message-$index'), + createdAt: lastSyncAt.add(Duration(seconds: index + 1)), + ), + ); + when(() => api.general.sync(cids, lastSyncAt)).thenAnswer( + (_) async => SyncResponse()..events = events, + ); + + final replayed = []; + final sub = client.on(EventType.messageNew).listen(replayed.add); + addTearDown(sub.cancel); + + await client.sync(); + await pumpEventQueue(); + + verify(() => api.general.sync(cids, lastSyncAt)).called(1); + // Within the limit, every event is replayed through the event handler. + expect(replayed, hasLength(events.length)); + // lastSyncAt advances to the newest replayed event date. + expect(await fakeClient.getLastSyncAt(), events.last.createdAt); + }, + ); + + test( + '''should refresh the synced channels in place of a payload that exceeds the replay limit''', + () async { + final cids = ['channel1']; + final lastSyncAt = DateTime.now().subtract(const Duration(hours: 1)); + final fakeClient = FakePersistenceClient( + channelCids: cids, + lastSyncAt: lastSyncAt, + ); + + client.chatPersistenceClient = fakeClient; + // 251 events exceeds the internal replay limit of 250. + final events = List.generate( + 251, + (index) => Event( + type: EventType.messageNew, + cid: 'channel1', + message: Message(id: 'message-$index'), + createdAt: lastSyncAt.add(Duration(seconds: index + 1)), + ), + ); + when(() => api.general.sync(cids, lastSyncAt)).thenAnswer( + (_) async => SyncResponse()..events = events, + ); + + 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((_) async => QueryChannelsResponse()..channels = []); + + final replayed = []; + final sub = client.on(EventType.messageNew).listen(replayed.add); + addTearDown(sub.cancel); + + // The group shares one api mock, so only count this test's calls. + clearInteractions(api.channel); + + await client.sync(); + await pumpEventQueue(); + + verify(() => api.general.sync(cids, lastSyncAt)).called(1); + // Replay is skipped; no events are dispatched through the handler. + expect(replayed, isEmpty); + // The channels the payload covered are refreshed in its place. + verify( + () => api.channel.queryChannels( + filter: Filter.in_('cid', cids), + 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: const PaginationParams(limit: 1), + ), + ).called(1); + // lastSyncAt moves to the newest event in the skipped payload, so + // that payload is not re-fetched while anything after it still is. + expect(await fakeClient.getLastSyncAt(), events.last.createdAt); + }, + ); }); }); @@ -5318,15 +5432,25 @@ void main() { await delay(300); } + // Recovery asks about channels most recently active first, so every fixture + // pins its own recency rather than inheriting the moment it was built. + // Both dates are set so `lastUpdatedAt` lands on [lastActiveAt] either way. + Channel channelActiveAt(String cid, DateTime lastActiveAt) { + final channel = ChannelModel(cid: cid, createdAt: lastActiveAt, lastMessageAt: lastActiveAt); + return Channel.fromState(client, ChannelState(channel: channel)); + } + test('should re-query active channels on reconnect when enabled (default)', () async { // Setup: connect with default flag, register two channels. client = StreamChatClient(apiKey, chatApi: api, ws: ws); await client.connectUser(user, token); await delay(300); - final channel1 = Channel.fromState(client, ChannelState(channel: ChannelModel(cid: 'messaging:c1'))); - final channel2 = Channel.fromState(client, ChannelState(channel: ChannelModel(cid: 'messaging:c2'))); - client.state.addChannels({'messaging:c1': channel1, 'messaging:c2': channel2}); + final now = DateTime.now(); + client.state.addChannels({ + 'messaging:c1': channelActiveAt('messaging:c1', now), + 'messaging:c2': channelActiveAt('messaging:c2', now.subtract(const Duration(minutes: 1))), + }); // Drop interactions from the initial connect's (empty-channel) recovery // so we only count the reconnect call. @@ -5334,6 +5458,7 @@ void main() { await simulateReconnect(); + // The re-query asks for exactly the channels it lists, a page at a time. verify( () => api.channel.queryChannels( filter: Filter.in_('cid', const ['messaging:c1', 'messaging:c2']), @@ -5343,7 +5468,7 @@ void main() { presence: any(named: 'presence'), memberLimit: any(named: 'memberLimit'), messageLimit: any(named: 'messageLimit'), - paginationParams: const PaginationParams(limit: 30), + paginationParams: const PaginationParams(limit: 2), ), ).called(1); }); @@ -5451,6 +5576,329 @@ void main() { ); }); + // Skipping event replay leaves the state of the synced channels behind, so + // the skip refreshes them itself, whatever this flag is set to. + test('should re-query active channels when the sync skipped event replay', () async { + client = StreamChatClient(apiKey, chatApi: api, ws: ws, recoverStateOnReconnect: false); + await client.connectUser(user, token); + await delay(300); + + const cid = 'messaging:c1'; + final channel = Channel.fromState(client, ChannelState(channel: ChannelModel(cid: cid))); + client.state.addChannels({cid: channel}); + + final lastSyncAt = DateTime.now().subtract(const Duration(hours: 1)); + final persistenceClient = FakePersistenceClient(channelCids: const [cid], lastSyncAt: lastSyncAt); + client.chatPersistenceClient = persistenceClient; + await client.openPersistenceConnection(user); + addTearDown(() => client.chatPersistenceClient = null); + + // 251 events exceeds the internal replay limit of 250. + final events = List.generate( + 251, + (index) => Event( + type: EventType.messageNew, + cid: cid, + message: Message(id: 'message-$index'), + createdAt: lastSyncAt.add(Duration(seconds: index + 1)), + ), + ); + when(() => api.general.sync(const [cid], lastSyncAt)).thenAnswer( + (_) async => SyncResponse()..events = events, + ); + + clearInteractions(api.channel); + + await simulateReconnect(); + + verify( + () => api.channel.queryChannels( + filter: Filter.in_('cid', const [cid]), + 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: const PaginationParams(limit: 1), + ), + ).called(1); + + // The pointer lands on the newest event in the skipped payload, not on + // the wall clock, so anything after it is still fetched next time. + expect(await persistenceClient.getLastSyncAt(), events.last.createdAt); + }); + + // A failed sync applied nothing and moved nothing, so the configured + // recovery still runs and the window stays outstanding for the next sync. + test('should re-query active channels when the sync fails, keeping lastSyncAt', () async { + client = StreamChatClient(apiKey, chatApi: api, ws: ws); + await client.connectUser(user, token); + await delay(300); + + const cid = 'messaging:c1'; + final channel = Channel.fromState(client, ChannelState(channel: ChannelModel(cid: cid))); + client.state.addChannels({cid: channel}); + + final lastSyncAt = DateTime.now().subtract(const Duration(hours: 1)); + final persistenceClient = FakePersistenceClient(channelCids: const [cid], lastSyncAt: lastSyncAt); + client.chatPersistenceClient = persistenceClient; + await client.openPersistenceConnection(user); + addTearDown(() => client.chatPersistenceClient = null); + + when(() => api.general.sync(const [cid], lastSyncAt)).thenThrow( + StreamChatNetworkError(ChatErrorCode.internalSystemError), + ); + + clearInteractions(api.channel); + + await simulateReconnect(); + + verify( + () => api.channel.queryChannels( + filter: Filter.in_('cid', const [cid]), + 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: const PaginationParams(limit: 1), + ), + ).called(1); + + expect(await persistenceClient.getLastSyncAt(), lastSyncAt); + }); + + // Discarding the payload is only safe once its state has been re-fetched, + // so a failed refresh keeps the checkpoint and the range is asked for again. + test('should keep lastSyncAt when the refresh after a skipped replay fails', () async { + 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'), + ), + ).thenThrow(const StreamChatError('You cannot use queryChannels without an active connection.')); + + client = StreamChatClient(apiKey, chatApi: api, ws: ws, recoverStateOnReconnect: false); + await client.connectUser(user, token); + await delay(300); + + const cid = 'messaging:c1'; + final channel = Channel.fromState(client, ChannelState(channel: ChannelModel(cid: cid))); + client.state.addChannels({cid: channel}); + + final lastSyncAt = DateTime.now().subtract(const Duration(hours: 1)); + final persistenceClient = FakePersistenceClient(channelCids: const [cid], lastSyncAt: lastSyncAt); + client.chatPersistenceClient = persistenceClient; + await client.openPersistenceConnection(user); + addTearDown(() => client.chatPersistenceClient = null); + + // 251 events exceeds the internal replay limit of 250. + final events = List.generate( + 251, + (index) => Event( + type: EventType.messageNew, + cid: cid, + message: Message(id: 'message-$index'), + createdAt: lastSyncAt.add(Duration(seconds: index + 1)), + ), + ); + when(() => api.general.sync(const [cid], lastSyncAt)).thenAnswer( + (_) async => SyncResponse()..events = events, + ); + + await simulateReconnect(); + + expect(await persistenceClient.getLastSyncAt(), lastSyncAt); + }); + + test('should re-query in batches when more channels are active than fit in one page', () async { + client = StreamChatClient(apiKey, chatApi: api, ws: ws, recoverStateOnReconnect: false); + await client.connectUser(user, token); + await delay(300); + + // 31 channels spill over the 30-channel page size into a second request. + // Listed most recently active first, which is the order recovery uses. + final now = DateTime.now(); + final cids = List.generate(31, (index) => 'messaging:c$index'); + client.state.addChannels({ + for (final (index, cid) in cids.indexed) cid: channelActiveAt(cid, now.subtract(Duration(minutes: index))), + }); + + final lastSyncAt = DateTime.now().subtract(const Duration(hours: 1)); + client.chatPersistenceClient = FakePersistenceClient(channelCids: cids, lastSyncAt: lastSyncAt); + await client.openPersistenceConnection(user); + addTearDown(() => client.chatPersistenceClient = null); + + final events = List.generate( + 251, + (index) => Event( + type: EventType.messageNew, + cid: cids.first, + message: Message(id: 'message-$index'), + createdAt: lastSyncAt.add(Duration(seconds: index + 1)), + ), + ); + when(() => api.general.sync(cids, lastSyncAt)).thenAnswer( + (_) async => SyncResponse()..events = events, + ); + + clearInteractions(api.channel); + + await simulateReconnect(); + + verify( + () => api.channel.queryChannels( + filter: Filter.in_('cid', cids.take(30).toList()), + 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: const PaginationParams(limit: 30), + ), + ).called(1); + + verify( + () => api.channel.queryChannels( + filter: Filter.in_('cid', cids.skip(30).toList()), + 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: const PaginationParams(limit: 1), + ), + ).called(1); + }); + + // Everything keyed off `connectionRecovered` — the list controllers, the + // retry queue — assumes the recovered state has already been applied when + // it fires. Emitting it before the catch-up finishes would have them act + // on state the sync has not written yet. + test('should finish recovering before `connectionRecovered` fires', () async { + client = StreamChatClient(apiKey, chatApi: api, ws: ws); + await client.connectUser(user, token); + await delay(300); + + const cid = 'messaging:c1'; + final channel = Channel.fromState(client, ChannelState(channel: ChannelModel(cid: cid))); + client.state.addChannels({cid: channel}); + + final lastSyncAt = DateTime.now().subtract(const Duration(hours: 1)); + client.chatPersistenceClient = FakePersistenceClient(channelCids: const [cid], lastSyncAt: lastSyncAt); + await client.openPersistenceConnection(user); + addTearDown(() => client.chatPersistenceClient = null); + + final calls = []; + when(() => api.general.sync(const [cid], lastSyncAt)).thenAnswer((_) async { + calls.add('sync'); + return SyncResponse()..events = []; + }); + 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((_) async { + calls.add('queryChannels'); + return QueryChannelsResponse()..channels = []; + }); + + final sub = client.on(EventType.connectionRecovered).listen((_) => calls.add('connectionRecovered')); + addTearDown(sub.cancel); + + await simulateReconnect(); + await pumpEventQueue(); + + expect(calls, ['sync', 'queryChannels', 'connectionRecovered']); + }); + + // One page failing says nothing about the others, so the rest are still + // attempted — the channels that can be refreshed are. + test('should attempt every page when one of them fails', () async { + client = StreamChatClient(apiKey, chatApi: api, ws: ws, recoverStateOnReconnect: false); + await client.connectUser(user, token); + await delay(300); + + // 31 channels spill over the 30-channel page size into a second request. + // Listed most recently active first, which is the order recovery uses. + final now = DateTime.now(); + final cids = List.generate(31, (index) => 'messaging:c$index'); + client.state.addChannels({ + for (final (index, cid) in cids.indexed) cid: channelActiveAt(cid, now.subtract(Duration(minutes: index))), + }); + + final lastSyncAt = DateTime.now().subtract(const Duration(hours: 1)); + final persistenceClient = FakePersistenceClient(channelCids: cids, lastSyncAt: lastSyncAt); + client.chatPersistenceClient = persistenceClient; + await client.openPersistenceConnection(user); + addTearDown(() => client.chatPersistenceClient = null); + + final events = List.generate( + 251, + (index) => Event( + type: EventType.messageNew, + cid: cids.first, + message: Message(id: 'message-$index'), + createdAt: lastSyncAt.add(Duration(seconds: index + 1)), + ), + ); + when(() => api.general.sync(cids, lastSyncAt)).thenAnswer( + (_) async => SyncResponse()..events = events, + ); + + // The first page fails, the second one succeeds. + when( + () => api.channel.queryChannels( + filter: Filter.in_('cid', cids.take(30).toList()), + 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'), + ), + ).thenThrow(const StreamChatError('Failed to query channels')); + + clearInteractions(api.channel); + + await simulateReconnect(); + + // Both pages are asked for, even though the first one failed. + verify( + () => api.channel.queryChannels( + filter: Filter.in_('cid', cids.skip(30).toList()), + 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: const PaginationParams(limit: 1), + ), + ).called(1); + + // The failure still keeps the checkpoint. + expect(await persistenceClient.getLastSyncAt(), lastSyncAt); + }); + test('should respect runtime toggling via the setter', () async { client = StreamChatClient(apiKey, chatApi: api, ws: ws); await client.connectUser(user, token); diff --git a/packages/stream_chat/test/src/client/sync_manager_test.dart b/packages/stream_chat/test/src/client/sync_manager_test.dart new file mode 100644 index 000000000..92d50048c --- /dev/null +++ b/packages/stream_chat/test/src/client/sync_manager_test.dart @@ -0,0 +1,695 @@ +import 'package:clock/clock.dart'; +import 'package:mocktail/mocktail.dart'; +import 'package:stream_chat/src/client/sync_manager.dart'; +import 'package:stream_chat/stream_chat.dart'; +import 'package:test/test.dart'; + +import '../fakes.dart'; +import '../mocks.dart'; + +// Records what `/sync` was asked for, and answers with events or an error. +// Callable, so it can be passed straight in as a [FetchMissedEvents]. +class _FakeSyncEndpoint { + _FakeSyncEndpoint({this.events = const [], this.error}); + + List events; + Exception? error; + + final calls = <({List cids, DateTime lastSyncAt})>[]; + + Future call(List cids, DateTime lastSyncAt) async { + calls.add((cids: cids, lastSyncAt: lastSyncAt)); + if (error case final error?) throw error; + return SyncResponse()..events = events; + } +} + +// Fails every checkpoint read, standing in for a corrupt or closed database. +class _ThrowingPersistenceClient extends Fake implements ChatPersistenceClient { + @override + Future getLastSyncAt() async => throw Exception('database is gone'); +} + +// Refuses to be reset but is otherwise readable, standing in for a store whose +// flush rolled back and left every row in place. +class _UnflushablePersistenceClient extends FakePersistenceClient { + _UnflushablePersistenceClient({super.lastSyncAt}); + + @override + Future flush() async => throw Exception('could not reset the store'); +} + +// A channel is only ever read for its cid and how recently it was active here. +// A plain fake rather than a mock: stubbing one inside another stub's +// `thenAnswer` re-enters mocktail and silently yields a null cid. +// +// Leaving [lastActiveAt] off stands for a channel whose state was never loaded. +class _FakeChannel extends Fake implements Channel { + _FakeChannel(this.cid, {DateTime? lastActiveAt}) + : state = lastActiveAt == null ? null : _FakeChannelClientState(cid, lastActiveAt); + + @override + final String? cid; + + @override + final ChannelClientState? state; +} + +// Holds just enough of a channel for `ChannelModel.lastUpdatedAt` to resolve. +// Both dates are set to [lastActiveAt] so it resolves there whichever of the +// two it picks. +class _FakeChannelClientState extends Fake implements ChannelClientState { + _FakeChannelClientState(String? cid, DateTime lastActiveAt) + : channelState = ChannelState( + channel: ChannelModel(cid: cid, createdAt: lastActiveAt, lastMessageAt: lastActiveAt), + ); + + @override + final ChannelState channelState; +} + +// The handles a test needs to observe a manager: what it replayed, and which +// channel pages it queried. +typedef _Harness = ({ + SyncManager manager, + MockStreamChatClient client, + List replayed, + List> queriedPages, +}); + +void main() { + registerFallbackValue(FakeEvent()); + registerFallbackValue(const PaginationParams()); + registerFallbackValue(const Filter.empty()); + + final t0 = DateTime.utc(2026, 3, 1, 12); + final anHourAgo = t0.subtract(const Duration(hours: 1)); + + Event eventAt(DateTime createdAt) => Event(type: EventType.messageNew, createdAt: createdAt); + + List eventsOf(int count) { + return List.generate(count, (i) => eventAt(anHourAgo.add(Duration(seconds: i + 1)))); + } + + StreamChatNetworkError badRequest() { + return StreamChatNetworkError.raw(code: 4, message: 'too many events', statusCode: 400); + } + + // [onQueryPage] decides what a page of `queryChannels` does: returning an + // Exception throws it, a List of cids answers with only those, and null + // answers with the whole page. + _Harness buildHarness({ + required _FakeSyncEndpoint api, + required ChatPersistenceClient persistence, + List activeCids = const ['messaging:a'], + Map lastActiveAt = const {}, + bool persistenceEnabled = true, + bool recoverStateOnReconnect = true, + Object? Function(List page)? onQueryPage, + }) { + final client = MockStreamChatClient(); + final state = MockClientState(); + final replayed = []; + final queriedPages = >[]; + + when(() => client.chatPersistenceClient).thenReturn(persistence); + client.persistenceEnabled = persistenceEnabled; + when(() => client.recoverStateOnReconnect).thenReturn(recoverStateOnReconnect); + when(() => client.state).thenReturn(state); + when(() => state.channels).thenReturn({ + for (final cid in activeCids) cid: _FakeChannel(cid, lastActiveAt: lastActiveAt[cid]), + }); + when(() => client.handleEvent(any())).thenAnswer((invocation) { + replayed.add(invocation.positionalArguments.first as Event); + }); + + when( + () => client.queryChannelsOnline( + filter: any(named: 'filter'), + paginationParams: any(named: 'paginationParams'), + waitForConnect: any(named: 'waitForConnect'), + ), + ).thenAnswer((invocation) async { + final filter = invocation.namedArguments[#filter] as Filter; + final page = (filter.value as List).cast(); + queriedPages.add(page); + + final outcome = onQueryPage?.call(page); + if (outcome is Exception) throw outcome; + + final answered = outcome is List ? outcome : page; + return [for (final cid in answered) _FakeChannel(cid)]; + }); + + return ( + manager: SyncManager(client: client, fetchMissedEvents: api.call), + client: client, + replayed: replayed, + queriedPages: queriedPages, + ); + } + + // Every test pins "now" to [t0] so the checkpoint-age rules are deterministic. + void testWithClock(String description, Future Function() body) { + test(description, () => withClock(Clock.fixed(t0), body)); + } + + testWithClock('sync caps the channel ids named in the request at 100', () async { + final api = _FakeSyncEndpoint(); + final harness = buildHarness( + api: api, + persistence: FakePersistenceClient(lastSyncAt: anHourAgo), + ); + + final cids = List.generate(300, (i) => 'messaging:$i'); + await harness.manager.sync(cids: cids); + + expect(api.calls.single.cids, cids.take(100)); + }); + + testWithClock('sync collapses duplicate channel ids before the cap applies', () async { + final api = _FakeSyncEndpoint(); + final harness = buildHarness( + api: api, + persistence: FakePersistenceClient(lastSyncAt: anHourAgo), + ); + + await harness.manager.sync(cids: ['messaging:a', 'messaging:a', 'messaging:b']); + + expect(api.calls.single.cids, ['messaging:a', 'messaging:b']); + }); + + testWithClock('sync asks for a window the server may refuse rather than pre-judging it', () async { + final api = _FakeSyncEndpoint(); + final persistence = FakePersistenceClient(lastSyncAt: t0.subtract(const Duration(days: 31))); + final harness = buildHarness(api: api, persistence: persistence); + + await harness.manager.sync(cids: ['messaging:a']); + + expect( + api.calls, + hasLength(1), + reason: 'the server owns the age limit; a refusal is handled, not predicted', + ); + }); + + testWithClock('sync does not request anything on a first sync', () async { + final api = _FakeSyncEndpoint(); + final harness = buildHarness(api: api, persistence: FakePersistenceClient()); + + await harness.manager.sync(cids: ['messaging:a']); + + expect(api.calls, isEmpty); + }); + + testWithClock('sync seeds the checkpoint to now on a first sync', () async { + final persistence = FakePersistenceClient(); + final harness = buildHarness(api: _FakeSyncEndpoint(), persistence: persistence); + + await harness.manager.sync(cids: ['messaging:a']); + + expect(await persistence.getLastSyncAt(), t0); + }); + + testWithClock('sync does nothing when there are no channels to catch up on', () async { + final api = _FakeSyncEndpoint(); + final persistence = FakePersistenceClient(); + final harness = buildHarness(api: api, persistence: persistence); + + await harness.manager.sync(); + + expect(api.calls, isEmpty); + expect(await persistence.getLastSyncAt(), isNull); + }); + + testWithClock('sync falls back to the persisted channel ids when none are named', () async { + final api = _FakeSyncEndpoint(); + final persistence = FakePersistenceClient(lastSyncAt: anHourAgo, channelCids: ['messaging:a']); + final harness = buildHarness(api: api, persistence: persistence); + + await harness.manager.sync(); + + expect(api.calls.single.cids, ['messaging:a']); + }); + + testWithClock('sync replays a window sitting exactly on the limit', () async { + final harness = buildHarness( + api: _FakeSyncEndpoint(events: eventsOf(250)), + persistence: FakePersistenceClient(lastSyncAt: anHourAgo), + ); + + await harness.manager.sync(cids: ['messaging:a']); + + expect(harness.replayed, hasLength(250)); + }); + + testWithClock('sync does not query channels for a window it replayed', () async { + final harness = buildHarness( + api: _FakeSyncEndpoint(events: eventsOf(250)), + persistence: FakePersistenceClient(lastSyncAt: anHourAgo), + ); + + await harness.manager.sync(cids: ['messaging:a']); + + expect(harness.queriedPages, isEmpty); + }); + + testWithClock('sync advances the checkpoint to the last replayed event', () async { + final events = eventsOf(250); + final persistence = FakePersistenceClient(lastSyncAt: anHourAgo); + final harness = buildHarness( + api: _FakeSyncEndpoint(events: events), + persistence: persistence, + ); + + await harness.manager.sync(cids: ['messaging:a']); + + expect(await persistence.getLastSyncAt(), events.last.createdAt); + }); + + testWithClock('sync does not replay a window one event over the limit', () async { + final harness = buildHarness( + api: _FakeSyncEndpoint(events: eventsOf(251)), + persistence: FakePersistenceClient(lastSyncAt: anHourAgo), + ); + + await harness.manager.sync(cids: ['messaging:a']); + + expect(harness.replayed, isEmpty); + }); + + testWithClock('sync queries the channels of a window it did not replay', () async { + final harness = buildHarness( + api: _FakeSyncEndpoint(events: eventsOf(251)), + persistence: FakePersistenceClient(lastSyncAt: anHourAgo), + ); + + await harness.manager.sync(cids: ['messaging:a']); + + expect(harness.queriedPages, [ + ['messaging:a'], + ]); + }); + + testWithClock('sync flushes the store for a window it did not replay', () async { + final persistence = FakePersistenceClient(lastSyncAt: anHourAgo); + final harness = buildHarness( + api: _FakeSyncEndpoint(events: eventsOf(251)), + persistence: persistence, + ); + + await harness.manager.sync(cids: ['messaging:a']); + + expect( + persistence.flushCallCount, + 1, + reason: 'what the store holds is missing every change the skipped events carried', + ); + }); + + testWithClock('sync advances past a window it did not replay', () async { + final events = eventsOf(251); + final persistence = FakePersistenceClient(lastSyncAt: anHourAgo); + final harness = buildHarness( + api: _FakeSyncEndpoint(events: events), + persistence: persistence, + ); + + await harness.manager.sync(cids: ['messaging:a']); + + expect(await persistence.getLastSyncAt(), events.last.createdAt); + }); + + testWithClock('sync takes the checkpoint from the newest event, whatever order they arrive in', () async { + final oldest = eventAt(anHourAgo.add(const Duration(minutes: 1))); + final newest = eventAt(anHourAgo.add(const Duration(minutes: 3))); + final persistence = FakePersistenceClient(lastSyncAt: anHourAgo); + final harness = buildHarness( + api: _FakeSyncEndpoint(events: [newest, oldest]), + persistence: persistence, + ); + + await harness.manager.sync(cids: ['messaging:a']); + + expect(await persistence.getLastSyncAt(), newest.createdAt); + }); + + testWithClock('sync does not let a persistence failure escape to the caller', () async { + final persistence = _ThrowingPersistenceClient(); + final harness = buildHarness(api: _FakeSyncEndpoint(), persistence: persistence); + + await expectLater(harness.manager.sync(cids: ['messaging:a']), completes); + }); + + testWithClock('sync advances the checkpoint to now when the window holds no events', () async { + final persistence = FakePersistenceClient(lastSyncAt: anHourAgo); + final harness = buildHarness(api: _FakeSyncEndpoint(), persistence: persistence); + + await harness.manager.sync(cids: ['messaging:a']); + + expect(await persistence.getLastSyncAt(), t0); + }); + + group('when a window cannot be replayed', () { + testWithClock('puts the checkpoint back if the refresh replacing it fails', () async { + final persistence = FakePersistenceClient(lastSyncAt: anHourAgo); + final harness = buildHarness( + api: _FakeSyncEndpoint(events: eventsOf(251)), + persistence: persistence, + onQueryPage: (_) => Exception('offline again'), + ); + + await harness.manager.sync(cids: ['messaging:a']); + + expect(persistence.flushCallCount, 1); + expect( + await persistence.getLastSyncAt(), + anHourAgo, + reason: 'the flush took the checkpoint with it, and the skipped events must stay recoverable', + ); + }); + + testWithClock('attempts every page even when one of them fails', () async { + final persistence = FakePersistenceClient(lastSyncAt: anHourAgo); + final harness = buildHarness( + api: _FakeSyncEndpoint(events: eventsOf(251)), + persistence: persistence, + onQueryPage: (page) => page.contains('messaging:0') ? Exception('offline again') : null, + ); + + await harness.manager.sync(cids: List.generate(90, (i) => 'messaging:$i')); + + expect(harness.queriedPages, hasLength(3), reason: 'a failing page must not skip the others'); + }); + + testWithClock('credits the pages that landed so recovery does not redo them', () async { + final cids = List.generate(90, (i) => 'messaging:$i'); + final lastPage = cids.sublist(60); + final persistence = FakePersistenceClient(lastSyncAt: anHourAgo); + final harness = buildHarness( + api: _FakeSyncEndpoint(events: eventsOf(251)), + persistence: persistence, + activeCids: cids, + onQueryPage: (page) => page.contains(lastPage.first) ? Exception('offline again') : null, + ); + + await harness.manager.recoverState(); + + expect( + await persistence.getLastSyncAt(), + anHourAgo, + reason: 'a channel left unrefreshed still has a window of events to catch up on', + ); + expect( + harness.queriedPages.skip(3), + [lastPage], + reason: 'the two pages that landed are credited, so only the failed one is retried', + ); + }); + + testWithClock('flushes, refreshes and advances when the server refuses it', () async { + final persistence = FakePersistenceClient(lastSyncAt: anHourAgo); + final harness = buildHarness( + api: _FakeSyncEndpoint(error: badRequest()), + persistence: persistence, + ); + + await harness.manager.sync(cids: ['messaging:a']); + + expect(persistence.flushCallCount, 1); + expect(harness.queriedPages, [ + ['messaging:a'], + ]); + expect(await persistence.getLastSyncAt(), t0); + }); + + testWithClock('still advances when the refresh after a refusal fails', () async { + final persistence = FakePersistenceClient(lastSyncAt: anHourAgo); + final harness = buildHarness( + api: _FakeSyncEndpoint(error: badRequest()), + persistence: persistence, + onQueryPage: (_) => Exception('offline again'), + ); + + await harness.manager.sync(cids: ['messaging:a']); + + expect(persistence.flushCallCount, 1); + expect( + await persistence.getLastSyncAt(), + t0, + reason: 'a refused window is refused again, so holding it would flush on every reconnect', + ); + }); + + testWithClock('keeps the checkpoint when the store will not drop', () async { + final persistence = _UnflushablePersistenceClient(lastSyncAt: anHourAgo); + final harness = buildHarness( + api: _FakeSyncEndpoint(events: eventsOf(251)), + persistence: persistence, + ); + + await harness.manager.sync(cids: ['messaging:a']); + + expect( + harness.queriedPages, + [ + ['messaging:a'], + ], + reason: 'the refresh still runs, so the channel is at least current in memory', + ); + expect( + await persistence.getLastSyncAt(), + anHourAgo, + reason: 'a store still holding the stale state must not be advanced past', + ); + }); + + testWithClock('does not let a failed flush escape to the caller', () async { + final harness = buildHarness( + api: _FakeSyncEndpoint(events: eventsOf(251)), + persistence: _ThrowingPersistenceClient(), + ); + + await expectLater( + harness.manager.sync(cids: ['messaging:a'], lastSyncAt: anHourAgo), + completes, + ); + }); + + testWithClock('leaves the checkpoint and the store alone on a non-400 failure', () async { + final persistence = FakePersistenceClient(lastSyncAt: anHourAgo); + final harness = buildHarness( + api: _FakeSyncEndpoint( + error: StreamChatNetworkError.raw(code: 0, message: 'boom', statusCode: 500), + ), + persistence: persistence, + ); + + await harness.manager.sync(cids: ['messaging:a']); + + expect(persistence.flushCallCount, 0); + expect(harness.queriedPages, isEmpty); + expect(await persistence.getLastSyncAt(), anHourAgo); + }); + }); + + testWithClock('recoverState asks about the most recently active channels first', () async { + final api = _FakeSyncEndpoint(); + final harness = buildHarness( + api: api, + persistence: FakePersistenceClient(lastSyncAt: anHourAgo), + // Held in memory in the order queries paged through them, oldest first. + activeCids: ['messaging:quiet', 'messaging:busy'], + lastActiveAt: { + 'messaging:quiet': anHourAgo.subtract(const Duration(days: 7)), + 'messaging:busy': anHourAgo, + }, + ); + + await harness.manager.recoverState(); + + expect( + api.calls.single.cids, + ['messaging:busy', 'messaging:quiet'], + reason: 'the cap drops whatever the request ends with, so recency has to decide the order', + ); + }); + + testWithClock('recoverState leaves channels with no loaded state at the end', () async { + final api = _FakeSyncEndpoint(); + final harness = buildHarness( + api: api, + persistence: FakePersistenceClient(lastSyncAt: anHourAgo), + activeCids: ['messaging:unloaded', 'messaging:quiet', 'messaging:busy'], + lastActiveAt: { + 'messaging:quiet': anHourAgo.subtract(const Duration(days: 7)), + 'messaging:busy': anHourAgo, + }, + ); + + await harness.manager.recoverState(); + + expect(api.calls.single.cids, ['messaging:busy', 'messaging:quiet', 'messaging:unloaded']); + }); + + testWithClock('recoverState does not query the channels the sync already refreshed', () async { + final harness = buildHarness( + api: _FakeSyncEndpoint(events: eventsOf(251)), + persistence: FakePersistenceClient(lastSyncAt: anHourAgo), + activeCids: ['messaging:a', 'messaging:b'], + ); + + await harness.manager.recoverState(); + + expect( + harness.queriedPages, + [ + ['messaging:a', 'messaging:b'], + ], + reason: 'the sync refreshed both, so recovery has nothing left to query', + ); + }); + + testWithClock('recoverState queries the channels a replayed window left stale', () async { + final harness = buildHarness( + api: _FakeSyncEndpoint(events: eventsOf(2)), + persistence: FakePersistenceClient(lastSyncAt: anHourAgo), + activeCids: ['messaging:a', 'messaging:b'], + ); + + await harness.manager.recoverState(); + + expect(harness.queriedPages, [ + ['messaging:a', 'messaging:b'], + ]); + }); + + testWithClock('recoverState still replays when recovery on reconnect is off', () async { + final api = _FakeSyncEndpoint(events: eventsOf(2)); + final harness = buildHarness( + api: api, + persistence: FakePersistenceClient(lastSyncAt: anHourAgo), + recoverStateOnReconnect: false, + ); + + await harness.manager.recoverState(); + + expect(api.calls, hasLength(1), reason: 'the replay answers to persistence, not to this flag'); + }); + + testWithClock('recoverState queries nothing when recovery on reconnect is off', () async { + final harness = buildHarness( + api: _FakeSyncEndpoint(events: eventsOf(2)), + persistence: FakePersistenceClient(lastSyncAt: anHourAgo), + recoverStateOnReconnect: false, + ); + + await harness.manager.recoverState(); + + expect(harness.queriedPages, isEmpty); + }); + + testWithClock('recoverState does not replay when persistence is disabled', () async { + final api = _FakeSyncEndpoint(events: eventsOf(2)); + final harness = buildHarness( + api: api, + persistence: FakePersistenceClient(lastSyncAt: anHourAgo), + persistenceEnabled: false, + ); + + await harness.manager.recoverState(); + + expect(api.calls, isEmpty); + }); + + testWithClock('recoverState still refreshes when persistence is disabled', () async { + final harness = buildHarness( + api: _FakeSyncEndpoint(events: eventsOf(2)), + persistence: FakePersistenceClient(lastSyncAt: anHourAgo), + persistenceEnabled: false, + ); + + await harness.manager.recoverState(); + + expect(harness.queriedPages, [ + ['messaging:a'], + ]); + }); + + testWithClock('recoverState does nothing when no channel is active', () async { + final api = _FakeSyncEndpoint(); + final harness = buildHarness( + api: api, + persistence: FakePersistenceClient(lastSyncAt: anHourAgo), + activeCids: const [], + ); + + await harness.manager.recoverState(); + + expect(api.calls, isEmpty); + expect(harness.queriedPages, isEmpty); + }); + + testWithClock('recoverState does not throw when the refresh fails', () async { + final harness = buildHarness( + api: _FakeSyncEndpoint(events: eventsOf(2)), + persistence: FakePersistenceClient(lastSyncAt: anHourAgo), + onQueryPage: (_) => Exception('offline again'), + ); + + await expectLater(harness.manager.recoverState(), completes); + }); + + testWithClock('sync does not throw when applying an event does', () async { + final harness = buildHarness( + api: _FakeSyncEndpoint(events: eventsOf(2)), + persistence: FakePersistenceClient(lastSyncAt: anHourAgo), + ); + when(() => harness.client.handleEvent(any())).thenThrow(Exception('a listener blew up')); + + await expectLater(harness.manager.sync(cids: ['messaging:a']), completes); + }); + + testWithClock('sync keeps lastSyncAt when a window is only partly applied', () async { + final persistence = FakePersistenceClient(lastSyncAt: anHourAgo); + final harness = buildHarness( + api: _FakeSyncEndpoint(events: eventsOf(2)), + persistence: persistence, + ); + when(() => harness.client.handleEvent(any())).thenThrow(Exception('a listener blew up')); + + await harness.manager.sync(cids: ['messaging:a']); + + expect(await persistence.getLastSyncAt(), anHourAgo); + }); + + testWithClock('recoverState pages the refresh rather than truncating it', () async { + final cids = List.generate(300, (i) => 'messaging:$i'); + final harness = buildHarness( + api: _FakeSyncEndpoint(events: eventsOf(2)), + persistence: FakePersistenceClient(lastSyncAt: anHourAgo), + activeCids: cids, + ); + + await harness.manager.recoverState(); + + expect(harness.queriedPages, hasLength(10)); + expect(harness.queriedPages.every((page) => page.length == 30), isTrue); + }); + + testWithClock('a channel the server omits does not hold lastSyncAt back', () async { + final events = eventsOf(251); + final persistence = FakePersistenceClient(lastSyncAt: anHourAgo); + final harness = buildHarness( + api: _FakeSyncEndpoint(events: events), + persistence: persistence, + // The server answers with only the first of the two asked for, as it + // would for a channel that has been deleted. + onQueryPage: (page) => [page.first], + ); + + await harness.manager.sync(cids: ['messaging:a', 'messaging:b']); + + expect(await persistence.getLastSyncAt(), events.last.createdAt); + }); +} diff --git a/packages/stream_chat/test/src/fakes.dart b/packages/stream_chat/test/src/fakes.dart index 3a919eb87..d14384a35 100644 --- a/packages/stream_chat/test/src/fakes.dart +++ b/packages/stream_chat/test/src/fakes.dart @@ -59,6 +59,7 @@ class FakePersistenceClient extends Fake implements ChatPersistenceClient { // Track method calls for testing int connectCallCount = 0; int disconnectCallCount = 0; + int flushCallCount = 0; @override bool get isConnected => _isConnected; @@ -84,6 +85,7 @@ class FakePersistenceClient extends Fake implements ChatPersistenceClient { @override Future flush() async { + flushCallCount++; _lastSyncAt = null; _channelCids = []; } @@ -98,6 +100,22 @@ class FakePersistenceClient extends Fake implements ChatPersistenceClient { @override Future> getChannelCids() async => _channelCids; + + @override + Future saveChannelQueries({ + required List cids, + Filter? filter, + SortOrder? sort, + String? predefinedFilter, + Filter? resolvedFilter, + SortOrder? resolvedSort, + Map? filterValues, + Map? sortValues, + bool clearQueryCache = false, + }) async {} + + @override + Future updateChannelStates(List channelStates) async {} } class FakeChatApi extends Fake implements StreamChatApi { diff --git a/packages/stream_chat/test/src/mocks.dart b/packages/stream_chat/test/src/mocks.dart index e248766ac..11912631d 100644 --- a/packages/stream_chat/test/src/mocks.dart +++ b/packages/stream_chat/test/src/mocks.dart @@ -99,8 +99,11 @@ class MockPersistenceClient extends Mock implements ChatPersistenceClient { } class MockStreamChatClient extends Mock implements StreamChatClient { + // A plain settable field for the same reason as [isLocalUnreadCountEnabled] + // below: stubbing it via `when()` corrupts mocktail's global stubbing state + // when this mock is lazily constructed inside another `when()`. @override - bool get persistenceEnabled => false; + bool persistenceEnabled = false; // A plain settable field (not a `when(...)` stub) so tests can flip it // with a direct assignment, e.g. `client.isLocalUnreadCountEnabled = true`. @@ -142,15 +145,20 @@ class MockStreamChatClient extends Mock implements StreamChatClient { } class MockStreamChatClientWithPersistence extends MockStreamChatClient { + MockStreamChatClientWithPersistence() { + // Sets the inherited field rather than overriding its getter, which would + // leave the inherited setter silently doing nothing. + persistenceEnabled = true; + } + ChatPersistenceClient? _persistenceClient; @override ChatPersistenceClient get chatPersistenceClient => _persistenceClient ??= MockPersistenceClient(); - - @override - bool get persistenceEnabled => true; } +class MockClientState extends Mock implements ClientState {} + class MockChannelConfig extends Mock implements ChannelConfig {} class MockRetryQueueChannel extends Mock implements Channel {