diff --git a/mobile/lib/features/channels/channels_provider.dart b/mobile/lib/features/channels/channels_provider.dart index 094f37b3f..e44464585 100644 --- a/mobile/lib/features/channels/channels_provider.dart +++ b/mobile/lib/features/channels/channels_provider.dart @@ -20,6 +20,8 @@ import 'unread_badge/should_notify_for_event.dart'; const _channelTypeOrder = {'stream': 0, 'forum': 1, 'dm': 2}; const _unreadCatchUpLimit = 1000; +const _unreadCatchUpDelay = Duration(seconds: 2); +const _unreadCatchUpInterRequestDelay = Duration(milliseconds: 250); const _participatedRootIdsPrefix = 'buzz-thread-participation.v1'; const _authoredRootIdsPrefix = 'buzz-thread-authored.v1'; @@ -36,13 +38,15 @@ const _authoredRootIdsPrefix = 'buzz-thread-authored.v1'; class ChannelsNotifier extends AsyncNotifier> { static const _backstopInterval = Duration(seconds: 60); - final Map _unsubscribersByChannel = {}; + void Function()? _liveUnsubscribe; Future _liveSubscriptionQueue = Future.value(); List _desiredLiveChannels = const []; Set _desiredLiveChannelIds = const {}; + Set _subscribedLiveChannelIds = const {}; int _subscriptionVersion = 0; String? _subscriptionRelayBaseUrl; Timer? _backstopTimer; + Timer? _unreadCatchUpTimer; final Map _latestObservedByChannel = {}; final Map> _observedUnreadEventsByChannel = {}; @@ -110,6 +114,8 @@ class ChannelsNotifier extends AsyncNotifier> { _observedUnreadEventsByChannel.clear(); _backstopTimer?.cancel(); _backstopTimer = null; + _unreadCatchUpTimer?.cancel(); + _unreadCatchUpTimer = null; }); if (sessionState.status != SessionStatus.connected) { @@ -279,53 +285,46 @@ class ChannelsNotifier extends AsyncNotifier> { // see every channel as having no messages. Skipped on backstop refreshes since // live subscriptions keep lastMessageAt current after the initial load. if (fetchLastMessage) { - final lastMessageResults = await Future.wait( - channels.map((channel) async { - if (!channel.isMember || channel.isArchived) return null; - try { - if (channel.isDm) { - final events = await session.fetchHistory( - NostrFilter( - kinds: EventKind.channelMessageEventKinds, - tags: { - '#h': [channel.id], - }, - limit: 1, - ), - ); - if (events.isEmpty) return null; - return MapEntry(channel.id, events.first.createdAt); - } - final events = await session.fetchHistory( - NostrFilter( - kinds: EventKind.channelMessageEventKinds, - tags: { - '#h': [channel.id], - }, - limit: 20, - ), - ); - for (final event in events) { - if (shouldNotifyForEvent( + final activeChannels = [ + for (final channel in channels) + if (channel.isMember && !channel.isArchived) channel, + ]; + final activeChannelsById = { + for (final channel in activeChannels) channel.id: channel, + }; + final lastMessageMap = {}; + try { + final events = await session.queryRelay([ + for (final channel in activeChannels) + NostrFilter( + kinds: EventKind.channelMessageEventKinds, + tags: { + '#h': [channel.id], + }, + limit: channel.isDm ? 1 : 20, + ), + ]); + for (final event in events) { + final channelId = event.channelId; + if (channelId == null) continue; + final channel = activeChannelsById[channelId]; + if (channel == null) continue; + if (!channel.isDm && + !shouldNotifyForEvent( event, myPk, mutedChannelIds: _mutedChannelIds(), - channelId: channel.id, + channelId: channelId, )) { - return MapEntry(channel.id, event.createdAt); - } - } - return null; - } catch (_) { - return null; + continue; } - }), - ); - - final lastMessageMap = {}; - for (final entry - in lastMessageResults.whereType>()) { - lastMessageMap[entry.key] = entry.value; + final current = lastMessageMap[channelId]; + if (current == null || event.createdAt > current) { + lastMessageMap[channelId] = event.createdAt; + } + } + } catch (error) { + debugPrint('[ChannelsNotifier] last-message fetch failed: $error'); } for (var i = 0; i < channels.length; i++) { @@ -524,76 +523,54 @@ class ChannelsNotifier extends AsyncNotifier> { return; } - if (_subscriptionRelayBaseUrl != relayBaseUrl) { - for (final unsubscribe in _unsubscribersByChannel.values) { - unsubscribe(); - } - _unsubscribersByChannel.clear(); + if (ref.read(relayConfigProvider).baseUrl != relayBaseUrl) return; + final channelIds = _desiredLiveChannelIds; + if (_subscriptionRelayBaseUrl != relayBaseUrl || + !_sameStringSet(_subscribedLiveChannelIds, channelIds)) { + _liveUnsubscribe?.call(); + _liveUnsubscribe = null; + _subscribedLiveChannelIds = const {}; _subscriptionRelayBaseUrl = relayBaseUrl; + + if (channelIds.isNotEmpty) { + final session = ref.read(relaySessionProvider.notifier); + try { + final unsubscribe = await session.subscribe( + NostrFilter( + kinds: EventKind.channelEventKinds, + tags: {'#h': channelIds.toList()}, + limit: 0, + ), + _handleLiveEvent, + ); + if (subscriptionVersion != _subscriptionVersion || + ref.read(relaySessionProvider).status != + SessionStatus.connected || + ref.read(relayConfigProvider).baseUrl != relayBaseUrl) { + unsubscribe(); + return; + } + _liveUnsubscribe = unsubscribe; + _subscribedLiveChannelIds = Set.unmodifiable(channelIds); + } catch (error) { + debugPrint('[ChannelsNotifier] live subscription failed: $error'); + } + } } - if (ref.read(relayConfigProvider).baseUrl != relayBaseUrl) { + + if (subscriptionVersion != _subscriptionVersion || + ref.read(relaySessionProvider).status != SessionStatus.connected) { return; } - final session = ref.read(relaySessionProvider.notifier); - final channelIds = _desiredLiveChannelIds; - for (final entry in _unsubscribersByChannel.entries.toList()) { - if (channelIds.contains(entry.key)) continue; - _unsubscribersByChannel.remove(entry.key); - entry.value(); - } - - for (final channelId in channelIds) { - if (ref.read(relaySessionProvider).status != SessionStatus.connected) { + _unreadCatchUpTimer?.cancel(); + _unreadCatchUpTimer = Timer(_unreadCatchUpDelay, () { + if (subscriptionVersion != _subscriptionVersion || + ref.read(relaySessionProvider).status != SessionStatus.connected) { return; } - if (_unsubscribersByChannel.containsKey(channelId)) continue; - try { - final unsubscribe = await session.subscribe( - NostrFilter( - kinds: EventKind.channelEventKinds, - tags: { - '#h': [channelId], - }, - limit: 0, - ), - _handleLiveEvent, - ); - if (ref.read(relaySessionProvider).status != SessionStatus.connected || - !_desiredLiveChannelIds.contains(channelId) || - ref.read(relayConfigProvider).baseUrl != relayBaseUrl || - _subscriptionRelayBaseUrl != relayBaseUrl) { - unsubscribe(); - return; - } - final replaced = _unsubscribersByChannel[channelId]; - if (replaced != null) { - unsubscribe(); - continue; - } - _unsubscribersByChannel[channelId] = unsubscribe; - } catch (error) { - debugPrint( - '[ChannelsNotifier] live subscription failed for $channelId: $error', - ); - } - } - - if (ref.read(relaySessionProvider).status != SessionStatus.connected) { - return; - } - - if (subscriptionVersion != _subscriptionVersion) { - final desiredChannelIds = _desiredLiveChannelIds; - for (final entry in _unsubscribersByChannel.entries.toList()) { - if (desiredChannelIds.contains(entry.key)) continue; - _unsubscribersByChannel.remove(entry.key); - entry.value(); - } - return; - } - - unawaited(_catchUpUnreadEvents(channels)); + unawaited(_catchUpUnreadEvents(channels)); + }); _backstopTimer?.cancel(); _backstopTimer = Timer.periodic( @@ -615,13 +592,14 @@ class ChannelsNotifier extends AsyncNotifier> { debugPrint('[ChannelsNotifier] unread catch-up skipped: $error'); return; } - final futures = >[]; + final tasks = Function()>[]; for (final channel in channels) { if (!channel.isMember || channel.isArchived) continue; final readAt = readState.effectiveTimestamp(channel.id); - futures.add( - _catchUpUnreadEventsForChannel( + if (readAt == null) continue; + tasks.add( + () => _catchUpUnreadEventsForChannel( session, channel, myPk, @@ -631,9 +609,12 @@ class ChannelsNotifier extends AsyncNotifier> { ); } - const batchSize = 5; - for (var i = 0; i < futures.length; i += batchSize) { - await Future.wait(futures.sublist(i, min(i + batchSize, futures.length))); + for (final task in tasks) { + if (ref.read(relaySessionProvider).status != SessionStatus.connected) { + return; + } + await task(); + await Future.delayed(_unreadCatchUpInterRequestDelay); } state = state.whenData((channels) => List.of(channels)); @@ -861,13 +842,14 @@ class ChannelsNotifier extends AsyncNotifier> { _subscriptionVersion++; _desiredLiveChannels = const []; _desiredLiveChannelIds = const {}; - for (final unsubscribe in _unsubscribersByChannel.values) { - unsubscribe(); - } - _unsubscribersByChannel.clear(); + _liveUnsubscribe?.call(); + _liveUnsubscribe = null; + _subscribedLiveChannelIds = const {}; _subscriptionRelayBaseUrl = null; _backstopTimer?.cancel(); _backstopTimer = null; + _unreadCatchUpTimer?.cancel(); + _unreadCatchUpTimer = null; } } @@ -875,6 +857,9 @@ final channelsProvider = AsyncNotifierProvider>( ChannelsNotifier.new, ); +bool _sameStringSet(Set left, Set right) => + left.length == right.length && left.containsAll(right); + String? _observedUnreadRootId(NostrEvent event) => _isBroadcastReply(event) ? null : event.threadReference.rootId; diff --git a/mobile/test/features/channels/channels_provider_test.dart b/mobile/test/features/channels/channels_provider_test.dart index 116a97785..83f9a7863 100644 --- a/mobile/test/features/channels/channels_provider_test.dart +++ b/mobile/test/features/channels/channels_provider_test.dart @@ -57,38 +57,32 @@ void main() { }, ); - test( - 'subscribes per-channel with #h tags (only joined, non-archived)', - () async { - final session = _FakeRelaySession( - memberships: [ - _membership(_channelA, myPk), - _membership(_channelB, myPk), - _membership(_channelD, myPk), - ], - metadata: [ - _meta(id: _channelA, name: 'general'), - _meta(id: _channelB, name: 'random'), - // channelD metadata missing -> won't appear in channel list - ], - ); - final container = _buildContainer(session: session); - addTearDown(container.dispose); + test('subscribes once for all joined, non-archived channel ids', () async { + final session = _FakeRelaySession( + memberships: [ + _membership(_channelA, myPk), + _membership(_channelB, myPk), + _membership(_channelD, myPk), + ], + metadata: [ + _meta(id: _channelA, name: 'general'), + _meta(id: _channelB, name: 'random'), + // channelD metadata missing -> won't appear in channel list + ], + ); + final container = _buildContainer(session: session); + addTearDown(container.dispose); - await container.read(channelsProvider.future); + await container.read(channelsProvider.future); - // One subscription per joined, non-archived channel. - expect(session.subscribeFilters, hasLength(2)); - expect( - session.subscribeFilters.map((f) => f.tags['#h']?.single).toSet(), - {_channelA, _channelB}, - ); - for (final filter in session.subscribeFilters) { - expect(filter.kinds, EventKind.channelEventKinds); - expect(filter.limit, 0); - } - }, - ); + expect(session.subscribeFilters, hasLength(1)); + expect(session.subscribeFilters.single.tags['#h']?.toSet(), { + _channelA, + _channelB, + }); + expect(session.subscribeFilters.single.kinds, EventKind.channelEventKinds); + expect(session.subscribeFilters.single.limit, 0); + }); test('retains channel-list member snapshots for immediate reuse', () async { final joinedAt = DateTime.fromMillisecondsSinceEpoch(1000, isUtc: true); @@ -132,7 +126,7 @@ void main() { expect(session.totalSubscribeCount, initialSubscribeCount); expect(session.unsubscribeCount, 0); - expect(session.subscribeFilters, hasLength(2)); + expect(session.subscribeFilters, hasLength(1)); }, ); @@ -164,14 +158,12 @@ void main() { await container.read(channelsProvider.notifier).refresh(); - expect(session.totalSubscribeCount, 3); + expect(session.totalSubscribeCount, 2); expect(session.unsubscribeCount, 1); - expect( - session.subscribeFilters - .map((filter) => filter.tags['#h']!.single) - .toSet(), - {_channelB, _channelD}, - ); + expect(session.subscribeFilters.single.tags['#h']!.toSet(), { + _channelB, + _channelD, + }); }, ); @@ -199,7 +191,7 @@ void main() { expect(session.activeChannels, isEmpty); expect(session.activeSubscriptionCount, 0); - expect(session.unsubscribeCount, 2); + expect(session.unsubscribeCount, 1); }, ); @@ -239,7 +231,7 @@ void main() { await Future.wait([firstRefresh, secondRefresh]); expect(session.activeChannels, {_channelA, _channelB, _channelD}); - expect(session.activeSubscriptionCount, 3); + expect(session.activeSubscriptionCount, 1); }, ); @@ -686,7 +678,7 @@ class _FakeRelaySession extends RelaySessionNotifier { int totalSubscribeCount = 0; Set get activeChannels => { - for (final (filter, _) in _subscriptions.values) ?filter.tags['#h']?.single, + for (final (filter, _) in _subscriptions.values) ...?filter.tags['#h'], }; int get activeSubscriptionCount => _subscriptions.length; @@ -747,6 +739,18 @@ class _FakeRelaySession extends RelaySessionNotifier { return const []; } + @override + Future> queryRelay( + List filters, { + Duration timeout = const Duration(seconds: 8), + }) async { + final events = []; + for (final filter in filters) { + events.addAll(await fetchHistory(filter, timeout: timeout)); + } + return events; + } + @override Future subscribe( NostrFilter filter,