diff --git a/mobile/lib/features/profile/presence_cache_provider.dart b/mobile/lib/features/profile/presence_cache_provider.dart index 8a56dc1226..1c0914c0af 100644 --- a/mobile/lib/features/profile/presence_cache_provider.dart +++ b/mobile/lib/features/profile/presence_cache_provider.dart @@ -27,9 +27,16 @@ class PresenceCacheNotifier extends Notifier> { final Map _timestamps = {}; final Map _snapshotAt = {}; final List _seenIds = []; + // Keys the live subscription asks for. Only tracked keys are subscribed: + // a community's presence heartbeats arrive many times a second. + final Set _subscribedKeys = {}; void Function()? _presenceUnsub; Timer? _poll; Timer? _queued; + Timer? _resubscribe; + int _nextSubscription = 0; + int _activeSubscription = 0; + bool _subscribeAttempted = false; int _generation = 0; int _epoch = 0; bool _querying = false; @@ -53,6 +60,9 @@ class PresenceCacheNotifier extends Notifier> { _timestamps.clear(); _snapshotAt.clear(); _seenIds.clear(); + _subscribedKeys.clear(); + _activeSubscription = 0; + _subscribeAttempted = false; _pending ..clear() ..addAll(_tracked); @@ -63,11 +73,13 @@ class PresenceCacheNotifier extends Notifier> { _poll?.cancel(); _queued?.cancel(); _queued = null; + _resubscribe?.cancel(); + _resubscribe = null; }); if (sessionState.status == SessionStatus.connected && lifecycle == AppLifecycleState.resumed) { _poll = Timer.periodic(const Duration(seconds: 60), (_) { - if (_presenceUnsub == null) { + if (_presenceUnsub == null || !_subscribedKeys.containsAll(_tracked)) { unawaited(_subscribe(generation)); } else if (!_querying) { _pending.addAll(_tracked); @@ -82,10 +94,32 @@ class PresenceCacheNotifier extends Notifier> { /// Request initial presence for exact keys. Repeated render calls coalesce. void track(List pubkeys) { + var added = false; for (final key in pubkeys.map((key) => key.trim().toLowerCase())) { - if (key.isNotEmpty && _tracked.add(key)) _pending.add(key); + if (key.isNotEmpty && _tracked.add(key)) { + _pending.add(key); + added = true; + } } - _schedule(_generation); + // While live (the poll runs only then), new keys are read once the + // subscription covers them, so no live event falls between the snapshot + // and the subscription. + if (added && (_poll?.isActive ?? false)) { + _queueResubscribe(_generation); + } else { + _schedule(_generation); + } + } + + void _queueResubscribe(int generation) { + _resubscribe ??= Timer(Duration.zero, () { + _resubscribe = null; + // Without a subscription, the next poll opens one with every key. + if (generation == _generation && + (_presenceUnsub != null || !_subscribeAttempted)) { + unawaited(_subscribe(generation)); + } + }); } void _schedule(int generation) { @@ -105,22 +139,36 @@ class PresenceCacheNotifier extends Notifier> { } Future _subscribe(int generation) async { - if (_opening || generation != _generation) return; + if (_opening || + generation != _generation || + _tracked.isEmpty || + (_presenceUnsub != null && _subscribedKeys.containsAll(_tracked))) { + return; + } _opening = true; - _retrying = false; + _subscribeAttempted = true; + final keys = _tracked.toList()..sort(); + final subscription = ++_nextSubscription; var closed = false; + // The subscription being replaced keeps delivering until this one opens. + bool current() => + generation == _generation && + !closed && + subscription == _activeSubscription; try { final unsub = await ref .read(relaySessionProvider.notifier) .subscribeWithStatus( - const NostrFilter(kinds: [EventKind.presenceUpdate], limit: 0), + NostrFilter( + kinds: const [EventKind.presenceUpdate], + authors: keys, + limit: 0, + ), (event) { - if (generation == _generation && !closed && !_retrying) { - _handlePresenceEvent(event); - } + if (current() && !_retrying) _handlePresenceEvent(event); }, onStatusChanged: (status) { - if (generation != _generation || closed) return; + if (!current()) return; _retrying = status == RelaySubscriptionStatus.retrying; _invalidateSnapshot(); if (!_retrying) _schedule(generation); @@ -128,8 +176,10 @@ class PresenceCacheNotifier extends Notifier> { onClosed: (_) { if (generation != _generation || closed) return; closed = true; + if (subscription != _activeSubscription) return; _presenceUnsub?.call(); _presenceUnsub = null; + _subscribedKeys.clear(); _invalidateSnapshot(); }, ); @@ -137,7 +187,16 @@ class PresenceCacheNotifier extends Notifier> { unsub(); return; } + final previous = _presenceUnsub; + _activeSubscription = subscription; + _retrying = false; _presenceUnsub = unsub; + _subscribedKeys + ..clear() + ..addAll(keys); + previous?.call(); + // Keys tracked while this one opened need a wider subscription. + if (!_subscribedKeys.containsAll(_tracked)) _queueResubscribe(generation); _schedule(generation); } catch (error) { if (generation == _generation) { diff --git a/mobile/test/features/profile/presence_cache_provider_test.dart b/mobile/test/features/profile/presence_cache_provider_test.dart index 219afd8f3c..150439991d 100644 --- a/mobile/test/features/profile/presence_cache_provider_test.dart +++ b/mobile/test/features/profile/presence_cache_provider_test.dart @@ -17,6 +17,7 @@ void main() { // Track alice, then emit her initial 'online' status. container.read(presenceCacheProvider.notifier).track(['alice']); + await _pumpEventQueue(); relaySession.emit(_presence('alice', 'online')); expect(container.read(presenceCacheProvider)['alice'], 'online'); @@ -35,6 +36,7 @@ void main() { // Track only alice. container.read(presenceCacheProvider.notifier).track(['alice']); + await _pumpEventQueue(); // Emit event for bob (untracked). relaySession.emit(_presence('bob', 'online')); @@ -52,6 +54,7 @@ void main() { await _pumpEventQueue(); container.read(presenceCacheProvider.notifier).track(['alice']); + await _pumpEventQueue(); relaySession.emit(_presence('alice', 'online')); expect(container.read(presenceCacheProvider)['alice'], 'online'); @@ -71,6 +74,7 @@ void main() { await _pumpEventQueue(); container.read(presenceCacheProvider.notifier).track(['alice']); + await _pumpEventQueue(); relaySession.emit(_presence('alice', 'online')); // Listen for state changes after initial setup. @@ -84,18 +88,30 @@ void main() { expect(stateChangeCount, 0); }); - test('subscribes to kind:20001 with limit 0', () async { + test('subscribes to kind:20001 for tracked pubkeys only', () async { final relaySession = _RecordingRelaySessionNotifier(); final container = _buildContainer(relaySession: relaySession); addTearDown(container.dispose); container.read(presenceCacheProvider); await _pumpEventQueue(); + // Nothing is tracked, so nothing is worth a community-wide stream. + expect(relaySession.filters, isEmpty); - // Should have subscribed with the correct filter. + container.read(presenceCacheProvider.notifier).track(['alice']); + await _pumpEventQueue(); expect(relaySession.filters, hasLength(1)); expect(relaySession.filters.single.kinds, [EventKind.presenceUpdate]); + expect(relaySession.filters.single.authors, ['alice']); expect(relaySession.filters.single.limit, 0); + + // A newly tracked key replaces the subscription with a wider one. + container.read(presenceCacheProvider.notifier).track(['bob', 'alice']); + await _pumpEventQueue(); + expect(relaySession.filters, hasLength(1)); + expect(relaySession.filters.single.authors, ['alice', 'bob']); + relaySession.emit(_presence('bob', 'online')); + expect(container.read(presenceCacheProvider)['bob'], 'online'); }); test('WS event uses pubkey variable, not literal string', () async { @@ -112,6 +128,7 @@ void main() { 'deadbeef', 'cafebabe', ]); + await _pumpEventQueue(); // Seed cafebabe -> offline, then set deadbeef online. relaySession.emit(_presence('cafebabe', 'offline'));