perf(mobile): bound channel startup work

Signed-off-by: Carl <4ddaeec1f70e26699465b4f4d4a55a825abb6045d769a21d44551b15c1e81f93@buzz.block.builderlab.xyz>

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<String, NostrEvent> events = {};
+}
+
+extension _ChannelBootstrapMerging on ChannelsNotifier {
+  void _cacheMemberSnapshots(
+    Iterable<NostrEvent> events, {
+    bool replaceAll = false,
+  }) {
+    final latestByChannelId = <String, NostrEvent>{};
+    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
+        ? <String, List<ChannelMember>>{}
+        : Map<String, List<ChannelMember>>.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<NostrFilter> _lastMessageFilters(List<Channel> channels) => [
+    for (final channel in channels)
+      NostrFilter(
+        kinds: EventKind.channelMessageEventKinds,
+        tags: {
+          '#h': [channel.id],
+        },
+        limit: channel.isDm ? 1 : 20,
+      ),
+  ];
+
+  void _mergeLastMessageEvents(
+    Map<String, int> lastMessageMap,
+    Iterable<NostrEvent> events, {
+    required Map<String, Channel> channelById,
+    required String myPk,
+    required Set<String> 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;
+      }
+    }
+  }
+}

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<List<Channel>> {
   int _subscriptionVersion = 0;
   String? _subscriptionRelayBaseUrl;
   Timer? _backstopTimer;
+  _BootstrapLiveEventBuffer? _bootstrapLiveEventBuffer;
   final Map<String, int> _latestObservedByChannel = {};
   final Map<String, Map<String, ObservedUnreadEvent>>
   _observedUnreadEventsByChannel = {};
@@ -128,6 +132,7 @@ class ChannelsNotifier extends AsyncNotifier<List<Channel>> {
       _observedUnreadEventsByChannel.clear();
       _backstopTimer?.cancel();
       _backstopTimer = null;
+      _bootstrapLiveEventBuffer = null;
     });
 
     if (sessionState.status != SessionStatus.connected) {
@@ -291,11 +296,17 @@ class ChannelsNotifier extends AsyncNotifier<List<Channel>> {
       }
     }
 
-    // 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<List<Channel>> {
       final channelById = {
         for (final channel in activeChannels) channel.id: channel,
       };
-      final events = await _fenced(
-        fence,
-        _fetchLastMessageEvents(session, activeChannels),
-      );
       final lastMessageMap = <String, int>{};
       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<List<Channel>> {
       }
     }
 
-    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<List<Channel>> {
     return channels;
   }
 
-  void _cacheMemberSnapshots(
-    Iterable<NostrEvent> events, {
-    bool replaceAll = false,
-  }) {
-    final latestByChannelId = <String, NostrEvent>{};
-    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
-        ? <String, List<ChannelMember>>{}
-        : Map<String, List<ChannelMember>>.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<List<NostrEvent>> _fetchLastMessageEvents(
-    RelaySessionNotifier session,
-    List<Channel> 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<List<NostrEvent>> _fetchChannelHistoryBatch(
     RelaySessionNotifier session,
     List<NostrFilter> filters, {
@@ -604,11 +573,16 @@ class ChannelsNotifier extends AsyncNotifier<List<Channel>> {
       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<List<Channel>> {
           continue;
         }
         _unsubscribersByChannel[channelId] = unsubscribe;
+        if (index + 1 < pendingChannelIds.length) {
+          await Future<void>.delayed(_liveSubscriptionStartInterval);
+        }
       } on _StaleChannelRefresh {
         rethrow;
       } catch (error) {
@@ -778,50 +755,55 @@ class ChannelsNotifier extends AsyncNotifier<List<Channel>> {
   }
 
   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<Channel>.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<Channel> 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<String> _mutedChannelIds() => {
     for (final entry in ref.read(channelMutesProvider).store.channels.entries)
       if (entry.value.muted) entry.key,

mobile/lib/shared/relay/relay_session.dart

@@ -154,6 +154,7 @@ class RelaySessionNotifier extends Notifier<SessionState> {
     List<NostrFilter> 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(

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<void>.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<void> _settle() async {
 Future<void> _waitUntil(bool Function() predicate) async {
   for (var i = 0; i < 100; i++) {
     if (predicate()) return;
-    await Future<void>.delayed(Duration.zero);
+    await Future<void>.delayed(const Duration(milliseconds: 5));
   }
   fail('Timed out waiting for asynchronous provider work');
 }
@@ -2439,6 +2498,8 @@ class _FakeRelaySession extends RelaySessionNotifier {
   final List<NostrEvent> huddleStarts;
   List<NostrEvent> recentMessages;
   int membershipFailures;
+  Completer<void>? _pausedLatestMessageQuery;
+  Completer<void>? _latestMessageQueryStarted;
   int directoryFailures = 0;
   bool failClaimedMemberCountQuery = false;
   bool failClaimedUnreadCatchUpQuery = false;
@@ -2485,6 +2546,21 @@ class _FakeRelaySession extends RelaySessionNotifier {
     await started.future;
   }
 
+  Future<void> get nextLatestMessageQueryStarted async {
+    final started = _latestMessageQueryStarted;
+    if (started == null) throw StateError('No latest-message query pending');
+    await started.future;
+  }
+
+  void pauseNextLatestMessageQuery() {
+    _pausedLatestMessageQuery = Completer<void>();
+    _latestMessageQueryStarted = Completer<void>();
+  }
+
+  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<NostrEvent>.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<NostrEvent> _matchingMessages(
+    List<NostrFilter> filters,
+    List<NostrEvent> messages,
+  ) {
+    return messages.where((event) {
       return filters.any((filter) {
         if (!filter.kinds.contains(event.kind)) return false;
         for (final entry in filter.tags.entries) {

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<void>.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(