diff --git a/mobile/lib/features/channels/channel_directory.dart b/mobile/lib/features/channels/channel_directory.dart index a99822ede5..042ad4a6ce 100644 --- a/mobile/lib/features/channels/channel_directory.dart +++ b/mobile/lib/features/channels/channel_directory.dart @@ -485,3 +485,86 @@ extension _ObservedUnreadRecording on ChannelsNotifier { } } } + +/// Scope-owned live events received before the initial list is published. +class _BootstrapLiveEventBuffer { + _BootstrapLiveEventBuffer(this.fence); + + final _ChannelRefreshFence fence; + final Map events = {}; +} + +extension _ChannelBootstrapMerging on ChannelsNotifier { + void _cacheMemberSnapshots( + Iterable events, { + bool replaceAll = false, + }) { + final latestByChannelId = {}; + for (final event in events) { + final channelId = event.getTagValue('d'); + if (channelId == null) continue; + final current = latestByChannelId[channelId]; + if (current == null || event.createdAt > current.createdAt) { + latestByChannelId[channelId] = event; + } + } + + final snapshots = replaceAll + ? >{} + : Map>.of(_memberSnapshotsByChannelId); + snapshots.addAll({ + for (final entry in latestByChannelId.entries) + entry.key: List.unmodifiable([ + for (final member in membersFromEvent(entry.value)) + ChannelMember( + pubkey: member.pubkey, + role: member.role, + joinedAt: DateTime.fromMillisecondsSinceEpoch( + entry.value.createdAt * 1000, + isUtc: true, + ), + ), + ]), + }); + _memberSnapshotsByChannelId = Map.unmodifiable(snapshots); + } + + List _lastMessageFilters(List channels) => [ + for (final channel in channels) + NostrFilter( + kinds: EventKind.channelMessageEventKinds, + tags: { + '#h': [channel.id], + }, + limit: channel.isDm ? 1 : 20, + ), + ]; + + void _mergeLastMessageEvents( + Map lastMessageMap, + Iterable events, { + required Map channelById, + required String myPk, + required Set mutedChannelIds, + }) { + for (final event in events) { + final channelId = event.channelId; + if (channelId == null) continue; + final channel = channelById[channelId]; + if (channel == null) continue; + if (!channel.isDm && + !shouldNotifyForEvent( + event, + myPk, + mutedChannelIds: mutedChannelIds, + channelId: channelId, + )) { + continue; + } + final current = lastMessageMap[channelId]; + if (current == null || event.createdAt > current) { + lastMessageMap[channelId] = event.createdAt; + } + } + } +} diff --git a/mobile/lib/features/channels/channels_provider.dart b/mobile/lib/features/channels/channels_provider.dart index d3bb29a070..79097f9d00 100644 --- a/mobile/lib/features/channels/channels_provider.dart +++ b/mobile/lib/features/channels/channels_provider.dart @@ -24,6 +24,9 @@ part 'channels_provider_lifecycle.dart'; const _channelTypeOrder = {'stream': 0, 'forum': 1, 'dm': 2}; const _unreadCatchUpLimit = 1000; +const _latestMessageQueryBatchSize = 128; +const _latestMessageQueryDeadline = Duration(seconds: 8); +const _liveSubscriptionStartInterval = Duration(milliseconds: 125); const _participatedRootIdsPrefix = 'buzz-thread-participation.v1'; const _authoredRootIdsPrefix = 'buzz-thread-authored.v1'; @@ -46,6 +49,7 @@ class ChannelsNotifier extends AsyncNotifier> { int _subscriptionVersion = 0; String? _subscriptionRelayBaseUrl; Timer? _backstopTimer; + _BootstrapLiveEventBuffer? _bootstrapLiveEventBuffer; final Map _latestObservedByChannel = {}; final Map> _observedUnreadEventsByChannel = {}; @@ -128,6 +132,7 @@ class ChannelsNotifier extends AsyncNotifier> { _observedUnreadEventsByChannel.clear(); _backstopTimer?.cancel(); _backstopTimer = null; + _bootstrapLiveEventBuffer = null; }); if (sessionState.status != SessionStatus.connected) { @@ -291,11 +296,17 @@ class ChannelsNotifier extends AsyncNotifier> { } } - // Step 3: fetch the most recent message per channel to populate lastMessageAt. - // kind:39000 metadata doesn't carry message timestamps, so channels load with - // lastMessageAt: null. Without this, unread detection and badge computation - // see every channel as having no messages. Skipped on backstop refreshes since - // live subscriptions keep lastMessageAt current after the initial load. + final bootstrapBuffer = subscribeLive && !_hasLoaded + ? _BootstrapLiveEventBuffer(fence) + : null; + if (bootstrapBuffer != null) { + _bootstrapLiveEventBuffer = bootstrapBuffer; + fence.ensureCurrent(); + unawaited(_subscribeLive(channels, fence)); + } + + // Fetch a bounded latest-message snapshot while initial live subscriptions + // are admitted in the background. Batches share one total startup deadline. if (fetchLastMessage) { final activeChannels = [ for (final channel in channels) @@ -304,29 +315,38 @@ class ChannelsNotifier extends AsyncNotifier> { final channelById = { for (final channel in activeChannels) channel.id: channel, }; - final events = await _fenced( - fence, - _fetchLastMessageEvents(session, activeChannels), - ); final lastMessageMap = {}; final mutedChannelIds = _mutedChannelIds(); - for (final event in events) { - final channelId = event.channelId; - if (channelId == null) continue; - final channel = channelById[channelId]; - if (channel == null) continue; - if (!channel.isDm && - !shouldNotifyForEvent( - event, - myPk, - mutedChannelIds: mutedChannelIds, - channelId: channelId, - )) { - continue; - } - final current = lastMessageMap[channelId]; - if (current == null || event.createdAt > current) { - lastMessageMap[channelId] = event.createdAt; + final stopwatch = Stopwatch()..start(); + for ( + var start = 0; + start < activeChannels.length; + start += _latestMessageQueryBatchSize + ) { + final remaining = _latestMessageQueryDeadline - stopwatch.elapsed; + if (remaining <= Duration.zero) break; + final end = min( + start + _latestMessageQueryBatchSize, + activeChannels.length, + ); + try { + final events = await _fenced( + fence, + session.queryRelay( + _lastMessageFilters(activeChannels.sublist(start, end)), + timeout: remaining, + ), + ); + _mergeLastMessageEvents( + lastMessageMap, + events, + channelById: channelById, + myPk: myPk, + mutedChannelIds: mutedChannelIds, + ); + } catch (error) { + if (error is _StaleChannelRefresh) rethrow; + debugPrint('[ChannelsNotifier] last-message batch failed: $error'); } } @@ -378,7 +398,18 @@ class ChannelsNotifier extends AsyncNotifier> { } } - if (subscribeLive) { + if (bootstrapBuffer != null) { + if (!identical(_bootstrapLiveEventBuffer, bootstrapBuffer)) { + throw const _StaleChannelRefresh(); + } + for (final event in bootstrapBuffer.events.values) { + _mergeLiveEventIntoChannels(channels, event); + } + // Publish while the buffer is still authoritative. No asynchronous + // callback can interleave between this write and the handoff below. + state = AsyncData(channels); + _bootstrapLiveEventBuffer = null; + } else if (subscribeLive) { // Subscriptions are shared relay state, so a retired refresh must not // install them even though its channel list is already built. fence.ensureCurrent(); @@ -390,68 +421,6 @@ class ChannelsNotifier extends AsyncNotifier> { return channels; } - void _cacheMemberSnapshots( - Iterable events, { - bool replaceAll = false, - }) { - final latestByChannelId = {}; - for (final event in events) { - final channelId = event.getTagValue('d'); - if (channelId == null) continue; - final current = latestByChannelId[channelId]; - if (current == null || event.createdAt > current.createdAt) { - latestByChannelId[channelId] = event; - } - } - - final snapshots = replaceAll - ? >{} - : Map>.of(_memberSnapshotsByChannelId); - snapshots.addAll({ - for (final entry in latestByChannelId.entries) - entry.key: List.unmodifiable([ - for (final member in membersFromEvent(entry.value)) - ChannelMember( - pubkey: member.pubkey, - role: member.role, - joinedAt: DateTime.fromMillisecondsSinceEpoch( - entry.value.createdAt * 1000, - isUtc: true, - ), - ), - ]), - }); - _memberSnapshotsByChannelId = Map.unmodifiable(snapshots); - } - - /// Fetches each channel's independent latest-message window in one HTTP - /// bridge request. The relay preserves NIP-01 per-filter limits while - /// executing the filters with bounded concurrency, avoiding an unbounded - /// burst of websocket REQs on communities with many channels. - Future> _fetchLastMessageEvents( - RelaySessionNotifier session, - List channels, - ) async { - if (channels.isEmpty) return const []; - - final filters = [ - for (final channel in channels) - NostrFilter( - kinds: EventKind.channelMessageEventKinds, - tags: { - '#h': [channel.id], - }, - limit: channel.isDm ? 1 : 20, - ), - ]; - - return _fetchChannelHistoryBatch( - session, - filters, - operation: 'latest-message query', - ); - } - Future> _fetchChannelHistoryBatch( RelaySessionNotifier session, List filters, { @@ -604,11 +573,16 @@ class ChannelsNotifier extends AsyncNotifier> { entry.value(); } - for (final channelId in channelIds) { + final pendingChannelIds = channelIds.toList(); + for (var index = 0; index < pendingChannelIds.length; index++) { + final channelId = pendingChannelIds[index]; if (ref.read(relaySessionProvider).status != SessionStatus.connected) { return; } if (_unsubscribersByChannel.containsKey(channelId)) continue; + if (subscriptionVersion != _subscriptionVersion || !fence.isCurrent) { + return; + } try { final unsubscribe = await session.subscribe( NostrFilter( @@ -638,6 +612,9 @@ class ChannelsNotifier extends AsyncNotifier> { continue; } _unsubscribersByChannel[channelId] = unsubscribe; + if (index + 1 < pendingChannelIds.length) { + await Future.delayed(_liveSubscriptionStartInterval); + } } on _StaleChannelRefresh { rethrow; } catch (error) { @@ -778,50 +755,55 @@ class ChannelsNotifier extends AsyncNotifier> { } void _handleLiveEvent(NostrEvent event) { - final channelId = event.channelId; - if (channelId == null) return; - - final myPk = ref.read(myPubkeyProvider); - final mutedChannelIds = _mutedChannelIds(); + final bootstrapBuffer = _bootstrapLiveEventBuffer; + if (bootstrapBuffer != null) { + if (bootstrapBuffer.fence.isCurrent) { + bootstrapBuffer.events[event.id] = event; + } + return; + } + if (event.channelId == null) return; state = state.whenData((channels) { - final idx = channels.indexWhere((c) => c.id == channelId); - if (idx == -1) { - refresh(); - return channels; - } final updated = List.of(channels); - final channel = updated[idx]; - - if (myPk != null && event.pubkey.toLowerCase() == myPk.toLowerCase()) { - _recordSelfThreadInterest(event, myPk); - } - - if (myPk != null && - shouldNotifyForEvent( - event, - myPk, - participatedRootIds: _participatedRootIds, - followedRootIds: _followedRootIds(), - authoredRootIds: _authoredRootIds, - mutedChannelIds: mutedChannelIds, - channelId: channel.id, - )) { - _recordUnreadEvent(channel, event, myPk); - final eventTime = DateTime.fromMillisecondsSinceEpoch( - event.createdAt * 1000, - isUtc: true, - ); - if (channel.lastMessageAt == null || - eventTime.isAfter(channel.lastMessageAt!)) { - updated[idx] = channel.copyWith(lastMessageAt: eventTime); - } - } - + _mergeLiveEventIntoChannels(updated, event); return updated; }); } + void _mergeLiveEventIntoChannels(List channels, NostrEvent event) { + final channelId = event.channelId; + if (channelId == null) return; + final idx = channels.indexWhere((channel) => channel.id == channelId); + if (idx == -1) return; + final myPk = ref.read(myPubkeyProvider); + final channel = channels[idx]; + if (myPk != null && event.pubkey.toLowerCase() == myPk.toLowerCase()) { + _recordSelfThreadInterest(event, myPk); + } + if (myPk == null || + !shouldNotifyForEvent( + event, + myPk, + participatedRootIds: _participatedRootIds, + followedRootIds: _followedRootIds(), + authoredRootIds: _authoredRootIds, + mutedChannelIds: _mutedChannelIds(), + channelId: channel.id, + )) { + return; + } + _recordUnreadEvent(channel, event, myPk); + final eventTime = DateTime.fromMillisecondsSinceEpoch( + event.createdAt * 1000, + isUtc: true, + ); + if (channel.lastMessageAt == null || + eventTime.isAfter(channel.lastMessageAt!)) { + channels[idx] = channel.copyWith(lastMessageAt: eventTime); + } + } + Set _mutedChannelIds() => { for (final entry in ref.read(channelMutesProvider).store.channels.entries) if (entry.value.muted) entry.key, diff --git a/mobile/lib/shared/relay/relay_session.dart b/mobile/lib/shared/relay/relay_session.dart index b8d13ee34a..fe0eb3daf4 100644 --- a/mobile/lib/shared/relay/relay_session.dart +++ b/mobile/lib/shared/relay/relay_session.dart @@ -154,6 +154,7 @@ class RelaySessionNotifier extends Notifier { List filters, { Duration timeout = const Duration(seconds: 8), }) async { + if (_rateLimitGate.isActive) await _rateLimitGate.wait(); final config = ref.read(relayConfigProvider); final url = Uri.parse(config.baseUrl).resolve('/query').toString(); final bodyBytes = utf8.encode( diff --git a/mobile/test/features/channels/channels_provider_test.dart b/mobile/test/features/channels/channels_provider_test.dart index 04f7cd917b..b81b3d8c4c 100644 --- a/mobile/test/features/channels/channels_provider_test.dart +++ b/mobile/test/features/channels/channels_provider_test.dart @@ -1593,6 +1593,7 @@ void main() { addTearDown(container.dispose); await container.read(channelsProvider.future); + await _waitUntil(() => session.activeSubscriptionCount == 2); // One subscription per joined, non-archived channel. expect(session.subscribeFilters, hasLength(2)); @@ -1643,6 +1644,11 @@ void main() { addTearDown(container.dispose); await container.read(channelsProvider.future); + await _waitUntil(() => session.activeSubscriptionCount == 2); + // Active fake subscriptions become visible just before the notifier + // records their unsubscribe handles. Let that synchronous handoff settle + // before testing an unchanged refresh. + await Future.delayed(const Duration(milliseconds: 10)); final initialSubscribeCount = session.totalSubscribeCount; await container.read(channelsProvider.notifier).refresh(); @@ -1709,6 +1715,7 @@ void main() { addTearDown(container.dispose); await container.read(channelsProvider.future); + await _waitUntil(() => session.activeSubscriptionCount == 2); session.memberships = []; session.metadata = []; @@ -1900,6 +1907,58 @@ void main() { }, ); + test('initial list retains a live event overlapping its snapshot', () async { + final session = _FakeRelaySession( + memberships: [_membership(_channelA, myPk)], + metadata: [_meta(id: _channelA, name: 'general')], + recentMessages: [ + NostrEvent( + id: 'snapshot', + pubkey: 'alice', + createdAt: 10, + kind: EventKind.streamMessageV2, + tags: const [ + ['h', _channelA], + ], + content: 'snapshot', + sig: 'sig', + ), + ], + )..pauseNextLatestMessageQuery(); + final container = _buildContainer(session: session); + addTearDown(container.dispose); + + final channelsFuture = container.read(channelsProvider.future); + await session.nextLatestMessageQueryStarted; + await _waitUntil(() => session.activeSubscriptionCount == 1); + final live = NostrEvent( + id: 'live', + pubkey: 'alice', + createdAt: 20, + kind: EventKind.streamMessageV2, + tags: const [ + ['h', _channelA], + ], + content: 'live', + sig: 'sig', + ); + session.emit(live); + session.emit(live); + session.resumePausedLatestMessageQuery(); + + final channels = await channelsFuture; + expect( + channels.single.lastMessageAt, + DateTime.fromMillisecondsSinceEpoch(20 * 1000, isUtc: true), + ); + expect( + container + .read(channelsProvider.notifier) + .observedUnreadEventsByChannel[_channelA], + contains('live'), + ); + }); + test('ephemeral (TTL) channels appear in the list', () async { // Regression: previously the provider unconditionally dropped any channel // with a `ttl` tag, which made TTL channels invisible on iOS even when the @@ -2402,7 +2461,7 @@ Future _settle() async { Future _waitUntil(bool Function() predicate) async { for (var i = 0; i < 100; i++) { if (predicate()) return; - await Future.delayed(Duration.zero); + await Future.delayed(const Duration(milliseconds: 5)); } fail('Timed out waiting for asynchronous provider work'); } @@ -2439,6 +2498,8 @@ class _FakeRelaySession extends RelaySessionNotifier { final List huddleStarts; List recentMessages; int membershipFailures; + Completer? _pausedLatestMessageQuery; + Completer? _latestMessageQueryStarted; int directoryFailures = 0; bool failClaimedMemberCountQuery = false; bool failClaimedUnreadCatchUpQuery = false; @@ -2485,6 +2546,21 @@ class _FakeRelaySession extends RelaySessionNotifier { await started.future; } + Future get nextLatestMessageQueryStarted async { + final started = _latestMessageQueryStarted; + if (started == null) throw StateError('No latest-message query pending'); + await started.future; + } + + void pauseNextLatestMessageQuery() { + _pausedLatestMessageQuery = Completer(); + _latestMessageQueryStarted = Completer(); + } + + void resumePausedLatestMessageQuery() { + _pausedLatestMessageQuery!.complete(); + } + void pauseNextSubscribe() { if (_pausedSubscribe != null) { throw StateError('A subscription is already paused'); @@ -2793,6 +2869,16 @@ class _FakeRelaySession extends RelaySessionNotifier { return filter.until == null ? directorySnapshot : const []; } queryBatches.add(filters); + final isLatestMessage = + filters.isNotEmpty && filters.every((filter) => filter.since == null); + if (isLatestMessage && _pausedLatestMessageQuery != null) { + final messages = List.of(recentMessages); + _latestMessageQueryStarted!.complete(); + await _pausedLatestMessageQuery!.future; + _pausedLatestMessageQuery = null; + _latestMessageQueryStarted = null; + return _matchingMessages(filters, messages); + } // The unread catch-up is the only batch that carries `since` on every // filter; the latest-message batch leaves it null. Snapshot the messages at // request time so a parked response reflects the scope that asked for it. @@ -2814,7 +2900,14 @@ class _FakeRelaySession extends RelaySessionNotifier { } } } - return messageSnapshot.where((event) { + return _matchingMessages(filters, messageSnapshot); + } + + List _matchingMessages( + List filters, + List messages, + ) { + return messages.where((event) { return filters.any((filter) { if (!filter.kinds.contains(event.kind)) return false; for (final entry in filter.tags.entries) { diff --git a/mobile/test/shared/relay/relay_session_test.dart b/mobile/test/shared/relay/relay_session_test.dart index c6ae6e9bd8..1749d8c59b 100644 --- a/mobile/test/shared/relay/relay_session_test.dart +++ b/mobile/test/shared/relay/relay_session_test.dart @@ -331,7 +331,7 @@ void main() { expect(gate.isActive, isFalse); }); - test('queryRelay does not wait for an active rate-limit gate', () async { + test('queryRelay waits for an active rate-limit gate', () async { final gate = RelayRateLimitGate( now: () => DateTime(2026), timerFactory: _ManualTimer.new, @@ -356,10 +356,10 @@ void main() { final query = harness.session.queryRelay(const []); await Future.delayed(Duration.zero); - expect(requestCount, 1); + expect(requestCount, 0); + gate.reset(); expect(await query, isEmpty); - // Still armed: the read must neither wait on the gate nor clear it. - expect(gate.isActive, isTrue); + expect(requestCount, 1); }); test(