perf(mobile): pace startup subscriptions

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/channel_sync.dart

@@ -0,0 +1,62 @@
+import 'dart:async';
+import 'dart:math';
+
+const channelQueryBatchSize = 100;
+const liveSubscriptionMaxConcurrent = 4;
+const liveSubscriptionStartInterval = Duration(milliseconds: 125);
+
+typedef TaskDelay = Future<void> Function(Duration duration);
+
+List<List<T>> chunkChannelQueryItems<T>(List<T> items) => [
+  for (var start = 0; start < items.length; start += channelQueryBatchSize)
+    items.sublist(start, min(start + channelQueryBatchSize, items.length)),
+];
+
+/// Runs tasks through one globally serialized admission chain.
+///
+/// The relay bills each REQ against a 50-per-five-second budget. A 125 ms
+/// interval admits at most 40 requests in a half-open five-second window, while
+/// the worker ceiling limits outstanding readiness waits when the relay stalls.
+Future<void> runPacedTasks(
+  List<Future<void> Function()> tasks, {
+  required int maxConcurrent,
+  required Duration startInterval,
+  required bool Function() isCancelled,
+  TaskDelay delay = defaultTaskDelay,
+  void Function(Object error)? onError,
+}) async {
+  if (maxConcurrent < 1) throw ArgumentError.value(maxConcurrent);
+  var nextTask = 0;
+  var firstStart = true;
+  Future<void> startPermit = Future.value();
+
+  Future<bool> acquireStartPermit() async {
+    if (firstStart) {
+      firstStart = false;
+    } else {
+      startPermit = startPermit.then((_) => delay(startInterval));
+      await startPermit;
+    }
+    return !isCancelled();
+  }
+
+  Future<void> worker() async {
+    while (true) {
+      final index = nextTask++;
+      if (index >= tasks.length) return;
+      if (!await acquireStartPermit()) return;
+      try {
+        await tasks[index]();
+      } catch (error) {
+        onError?.call(error);
+      }
+    }
+  }
+
+  await Future.wait([
+    for (var i = 0; i < min(maxConcurrent, tasks.length); i++) worker(),
+  ]);
+}
+
+Future<void> defaultTaskDelay(Duration duration) =>
+    Future<void>.delayed(duration);