diff --git a/mobile/lib/features/profile/presence_cache_provider.dart b/mobile/lib/features/profile/presence_cache_provider.dart index f735c376c35..2ce1100eb54 100644 --- a/mobile/lib/features/profile/presence_cache_provider.dart +++ b/mobile/lib/features/profile/presence_cache_provider.dart @@ -8,11 +8,15 @@ import '../../shared/relay/relay.dart'; /// In-memory cache of other users' presence. /// /// Subscribes to kind:20001 presence events over the relay WebSocket for -/// real-time updates. There is no longer a REST backstop — agents that -/// publish presence purely over WS are fine, and TTL expiry will be handled -/// by the relay-side `presence:true` filter extension when that lands. +/// real-time updates. Hydrates newly tracked users from relay-generated +/// kind:40902 snapshots so the UI does not wait for the next heartbeat. class PresenceCacheNotifier extends Notifier> { + static const _batchDelay = Duration(milliseconds: 50); + final Set _tracked = {}; + final Set _pending = {}; + final Map _revisions = {}; + Timer? _batchTimer; void Function()? _presenceUnsub; int _subscriptionVersion = 0; @@ -21,28 +25,36 @@ class PresenceCacheNotifier extends Notifier> { final sessionState = ref.watch(relaySessionProvider); ref.onDispose(() { + _batchTimer?.cancel(); + _batchTimer = null; + _pending.clear(); _presenceUnsub?.call(); _presenceUnsub = null; }); if (sessionState.status == SessionStatus.connected) { _subscribePresenceUpdates(); + Future.microtask(_refreshAll); } return {}; } /// Track presence for [pubkeys]. - /// - /// Currently a no-op for the actual fetch — we rely on live kind:20001 - /// events. The tracked set is still used to filter incoming events so the - /// cache doesn't grow unbounded. void track(List pubkeys) { - final normalized = pubkeys.map((pk) => pk.toLowerCase()).toList(); + final normalized = pubkeys + .map((pk) => pk.toLowerCase()) + .where((pk) => pk.isNotEmpty) + .toList(); + final uncached = normalized + .where((pk) => !state.containsKey(pk) && !_pending.contains(pk)) + .toList(); + _tracked.addAll(normalized); - // TODO(presence): once the relay supports a `presence:true` filter - // extension, issue a one-shot fetch here for the latest known state per - // pubkey. Until then, presence is "online whenever they publish". + + if (uncached.isEmpty) return; + _pending.addAll(uncached); + _batchTimer ??= Timer(_batchDelay, _flushPending); } /// Subscribe to kind:20001 presence events over WebSocket. @@ -76,12 +88,84 @@ class PresenceCacheNotifier extends Notifier> { final pubkey = event.pubkey.toLowerCase(); if (!_tracked.contains(pubkey)) return; final status = event.content; - if (status != 'online' && status != 'away' && status != 'offline') return; + if (!_isValidStatus(status)) return; + _updatePresence(pubkey, status); + } + + Future _refreshAll() async { + if (_tracked.isEmpty) return; + await _fetchPresence(_tracked.toList()); + } + + Future _flushPending() async { + _batchTimer = null; + if (_pending.isEmpty) return; + + final pubkeys = _pending.toList(); + _pending.clear(); + await _fetchPresence(pubkeys); + } + + Future _fetchPresence(List pubkeys) async { + final requested = pubkeys.toSet().toList(); + if (requested.isEmpty) return; + + // Relay queries require NIP-98 authentication. A connected-looking dev or + // test session can still be anonymous, so avoid starting a request that is + // guaranteed to fail until an identity is available. + final nsec = ref.read(relayConfigProvider).nsec; + if (nsec == null || nsec.isEmpty) return; + + final startingRevisions = { + for (final pubkey in requested) pubkey: _revisions[pubkey] ?? 0, + }; + + try { + final session = ref.read(relaySessionProvider.notifier); + final events = await session.queryRelay([ + NostrFilter( + kinds: const [EventKind.presenceSnapshot], + authors: requested, + limit: requested.length, + ), + ]); + + final statuses = {}; + for (final event in events) { + final pubkey = event.getTagValue('p')?.toLowerCase(); + final status = event.content; + if (pubkey == null || + !startingRevisions.containsKey(pubkey) || + !_isValidStatus(status)) { + continue; + } + final existing = statuses[pubkey]; + if (existing == null || event.createdAt > existing.createdAt) { + statuses[pubkey] = (createdAt: event.createdAt, status: status); + } + } + + for (final pubkey in requested) { + // A live update received while this request was in flight is newer + // than the snapshot and must win. + if ((_revisions[pubkey] ?? 0) != startingRevisions[pubkey]) continue; + _updatePresence(pubkey, statuses[pubkey]?.status ?? 'offline'); + } + } catch (error) { + debugPrint('[PresenceCacheNotifier] presence snapshot failed: $error'); + } + } + + void _updatePresence(String pubkey, String status) { if (state[pubkey] == status) return; final updated = Map.from(state); updated[pubkey] = status; + _revisions[pubkey] = (_revisions[pubkey] ?? 0) + 1; state = updated; } + + bool _isValidStatus(String status) => + status == 'online' || status == 'away' || status == 'offline'; } final presenceCacheProvider = diff --git a/mobile/lib/shared/relay/nostr_models.dart b/mobile/lib/shared/relay/nostr_models.dart index 820fee4ed6a..0a93ac618bb 100644 --- a/mobile/lib/shared/relay/nostr_models.dart +++ b/mobile/lib/shared/relay/nostr_models.dart @@ -24,6 +24,7 @@ abstract final class EventKind { static const streamMessageV2 = 40002; static const channelThreadSummary = 39005; static const channelWindowBounds = 39006; + static const presenceSnapshot = 40902; static const streamMessageEdit = 40003; static const streamMessageDiff = 40008; static const systemMessage = 40099; diff --git a/mobile/test/features/profile/presence_cache_provider_test.dart b/mobile/test/features/profile/presence_cache_provider_test.dart index 9fc02d9575b..4a8205f3683 100644 --- a/mobile/test/features/profile/presence_cache_provider_test.dart +++ b/mobile/test/features/profile/presence_cache_provider_test.dart @@ -1,17 +1,92 @@ +import 'dart:async'; + import 'package:flutter/widgets.dart'; import 'package:flutter_test/flutter_test.dart'; import 'package:hooks_riverpod/hooks_riverpod.dart'; +import 'package:nostr/nostr.dart' as nostr; import 'package:buzz/features/profile/presence_cache_provider.dart'; import 'package:buzz/shared/relay/relay.dart'; /// Tests for [PresenceCacheNotifier] in the pure-Nostr world. /// -/// The cache is now purely WS-driven: the notifier subscribes to kind:20001 -/// (presence updates) over the relay session and only mutates state for -/// pubkeys that have been registered via [PresenceCacheNotifier.track]. -/// There is no longer a REST backstop — the previous test seeded state via -/// a `GET /api/presence` call which has been removed. +/// The cache hydrates tracked users from relay-generated snapshots, then keeps +/// them current through live kind:20001 presence updates. void main() { + test('hydrates tracked pubkeys from a batched presence snapshot', () async { + final relaySession = _RecordingRelaySessionNotifier() + ..queryResult = [ + _snapshot('relay', 'alice', 'online'), + _snapshot('relay', 'bob', 'away'), + ]; + final container = _buildContainer(relaySession: relaySession); + addTearDown(container.dispose); + + container.read(presenceCacheProvider); + await _pumpEventQueue(); + + container.read(presenceCacheProvider.notifier).track(['Alice']); + container.read(presenceCacheProvider.notifier).track(['BOB']); + await _waitForSnapshot(); + + expect(relaySession.queries, hasLength(1)); + final filter = relaySession.queries.single.single; + expect(filter.kinds, [EventKind.presenceSnapshot]); + expect(filter.authors, containsAll(['alice', 'bob'])); + expect(filter.limit, 2); + expect(container.read(presenceCacheProvider), { + 'alice': 'online', + 'bob': 'away', + }); + }); + + test('marks a tracked pubkey offline when no snapshot exists', () async { + final relaySession = _RecordingRelaySessionNotifier(); + final container = _buildContainer(relaySession: relaySession); + addTearDown(container.dispose); + + container.read(presenceCacheProvider); + await _pumpEventQueue(); + + container.read(presenceCacheProvider.notifier).track(['alice']); + await _waitForSnapshot(); + + expect(container.read(presenceCacheProvider)['alice'], 'offline'); + }); + + test('live update wins over an in-flight presence snapshot', () async { + final snapshotCompleter = Completer>(); + final relaySession = _RecordingRelaySessionNotifier() + ..queryHandler = (_) => snapshotCompleter.future; + final container = _buildContainer(relaySession: relaySession); + addTearDown(container.dispose); + + container.read(presenceCacheProvider); + await _pumpEventQueue(); + + container.read(presenceCacheProvider.notifier).track(['alice']); + await _waitForSnapshotStart(); + relaySession.emit(_presence('alice', 'online')); + snapshotCompleter.complete(const []); + await _pumpEventQueue(); + + expect(container.read(presenceCacheProvider)['alice'], 'online'); + }); + + test('snapshot failure leaves presence unknown for a live update', () async { + final relaySession = _RecordingRelaySessionNotifier() + ..queryHandler = (_) => Future.error(Exception('snapshot unavailable')); + final container = _buildContainer(relaySession: relaySession); + addTearDown(container.dispose); + + container.read(presenceCacheProvider); + await _pumpEventQueue(); + + container.read(presenceCacheProvider.notifier).track(['alice']); + await _waitForSnapshot(); + + expect(container.read(presenceCacheProvider).containsKey('alice'), isFalse); + }); + test('WS presence event updates cache for tracked pubkey', () async { final relaySession = _RecordingRelaySessionNotifier(); final container = _buildContainer(relaySession: relaySession); @@ -148,20 +223,41 @@ Future _pumpEventQueue() async { await Future.delayed(Duration.zero); } +Future _waitForSnapshotStart() async { + await Future.delayed(const Duration(milliseconds: 75)); +} + +Future _waitForSnapshot() async { + await _waitForSnapshotStart(); + await _pumpEventQueue(); +} + ProviderContainer _buildContainer({ required _RecordingRelaySessionNotifier relaySession, }) { return ProviderContainer( overrides: [ appLifecycleProvider.overrideWith(() => _FakeAppLifecycleNotifier()), + relayConfigProvider.overrideWith(_AuthenticatedRelayConfigNotifier.new), relaySessionProvider.overrideWith(() => relaySession), ], ); } +class _AuthenticatedRelayConfigNotifier extends RelayConfigNotifier { + final String _nsec = nostr.Keys.generate().nsec; + + @override + RelayConfig build() => + RelayConfig(baseUrl: 'https://relay.example', nsec: _nsec); +} + class _RecordingRelaySessionNotifier extends RelaySessionNotifier { final List filters = []; + final List> queries = []; final List _listeners = []; + List queryResult = const []; + Future> Function(List)? queryHandler; @override SessionState build() => const SessionState(status: SessionStatus.connected); @@ -180,6 +276,15 @@ class _RecordingRelaySessionNotifier extends RelaySessionNotifier { }; } + @override + Future> queryRelay( + List filters, { + Duration timeout = const Duration(seconds: 8), + }) { + queries.add(filters); + return queryHandler?.call(filters) ?? Future.value(queryResult); + } + /// Emit an event synchronously to all live subscribers. void emit(NostrEvent event) { for (final listener in List.of(_listeners)) { @@ -188,6 +293,19 @@ class _RecordingRelaySessionNotifier extends RelaySessionNotifier { } } +NostrEvent _snapshot(String relayPubkey, String subjectPubkey, String status) => + NostrEvent( + id: 'snapshot-$subjectPubkey-$status', + pubkey: relayPubkey, + createdAt: 1000, + kind: EventKind.presenceUpdate, + tags: [ + ['p', subjectPubkey], + ], + content: status, + sig: 'sig', + ); + class _FakeAppLifecycleNotifier extends AppLifecycleNotifier { @override AppLifecycleState build() => AppLifecycleState.resumed;