From a7319c16a74fd80f8cb0749c0041e955cbafe5ff Mon Sep 17 00:00:00 2001 From: syntaxafterdark <309209481+syntaxafterdark@users.noreply.github.com> Date: Wed, 5 Aug 2026 14:34:52 +0800 Subject: [PATCH] fix(mobile): reconcile presence snapshots after reconnect Signed-off-by: syntaxafterdark <309209481+syntaxafterdark@users.noreply.github.com> --- .../profile/presence_cache_provider.dart | 183 +++++++- .../profile/presence_cache_provider_test.dart | 412 +++++++++++++++++- 2 files changed, 570 insertions(+), 25 deletions(-) diff --git a/mobile/lib/features/profile/presence_cache_provider.dart b/mobile/lib/features/profile/presence_cache_provider.dart index f735c376c3..72740aebd0 100644 --- a/mobile/lib/features/profile/presence_cache_provider.dart +++ b/mobile/lib/features/profile/presence_cache_provider.dart @@ -8,41 +8,170 @@ 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 and queries the relay's current presence snapshot when +/// pubkeys are first tracked, the session reconnects, and periodically while +/// connected. class PresenceCacheNotifier extends Notifier> { + static const _snapshotStaleAfter = Duration(seconds: 30); + static const _snapshotBatchDelay = Duration(milliseconds: 50); + static const _defaultRefreshInterval = Duration(seconds: 60); + + /// Creates a presence cache with an optional refresh interval for tests. + PresenceCacheNotifier({Duration refreshInterval = _defaultRefreshInterval}) + : _refreshInterval = refreshInterval; + final Set _tracked = {}; + final Set _pendingSnapshot = {}; + final Map _inFlightSnapshotVersion = {}; + final Map _latestSnapshotCreatedAt = {}; + final Map _liveArrivalRevision = {}; + final Map _snapshotFetchedAt = {}; + final Duration _refreshInterval; + Timer? _snapshotTimer; + Timer? _refreshTimer; void Function()? _presenceUnsub; int _subscriptionVersion = 0; + int _snapshotVersion = 0; @override Map build() { final sessionState = ref.watch(relaySessionProvider); + _snapshotVersion++; + _latestSnapshotCreatedAt.clear(); + _liveArrivalRevision.clear(); + _snapshotTimer?.cancel(); + _snapshotTimer = null; + _refreshTimer?.cancel(); + _refreshTimer = null; + _pendingSnapshot.clear(); + _inFlightSnapshotVersion.clear(); ref.onDispose(() { + _snapshotTimer?.cancel(); + _snapshotTimer = null; + _refreshTimer?.cancel(); + _refreshTimer = null; _presenceUnsub?.call(); _presenceUnsub = null; + _snapshotVersion++; }); if (sessionState.status == SessionStatus.connected) { _subscribePresenceUpdates(); + _scheduleSnapshot(_tracked, force: true); + _ensureRefreshTimer(); } 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. + /// Track presence for [pubkeys] and fetch their current relay snapshot. void track(List pubkeys) { - final normalized = pubkeys.map((pk) => pk.toLowerCase()).toList(); + final normalized = pubkeys + .map((pk) => pk.toLowerCase()) + .where((pk) => pk.isNotEmpty) + .toSet(); _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 (ref.read(relaySessionProvider).status == SessionStatus.connected) { + _scheduleSnapshot(normalized); + _ensureRefreshTimer(); + } + } + + void _ensureRefreshTimer() { + if (_tracked.isEmpty || _refreshTimer != null) return; + _refreshTimer = Timer.periodic(_refreshInterval, (_) { + if (ref.read(relaySessionProvider).status == SessionStatus.connected) { + _scheduleSnapshot(_tracked, force: true); + } + }); + } + + void _scheduleSnapshot(Iterable pubkeys, {bool force = false}) { + final now = DateTime.now(); + for (final pubkey in pubkeys) { + if (_inFlightSnapshotVersion[pubkey] == _snapshotVersion) continue; + final fetchedAt = _snapshotFetchedAt[pubkey]; + if (force || + fetchedAt == null || + now.difference(fetchedAt) >= _snapshotStaleAfter) { + _pendingSnapshot.add(pubkey); + } + } + if (_pendingSnapshot.isEmpty) return; + _snapshotTimer ??= Timer(_snapshotBatchDelay, _flushPendingSnapshot); + } + + Future _flushPendingSnapshot() async { + _snapshotTimer = null; + final pubkeys = _pendingSnapshot.toList(); + _pendingSnapshot.clear(); + if (pubkeys.isEmpty) return; + final requested = pubkeys.toSet(); + final version = _snapshotVersion; + for (final pubkey in requested) { + _inFlightSnapshotVersion[pubkey] = version; + } + final session = ref.read(relaySessionProvider.notifier); + final liveRevisionBeforeQuery = { + for (final pubkey in pubkeys) pubkey: _liveArrivalRevision[pubkey] ?? 0, + }; + + try { + final events = await session.queryRelay([ + NostrFilter(kinds: const [EventKind.presenceUpdate], authors: pubkeys), + ]); + if (version != _snapshotVersion) return; + + final returned = {}; + final updated = Map.from(state); + var changed = false; + for (final event in events) { + final taggedPubkey = event.getTagValue('p'); + final pubkey = + (taggedPubkey == null || taggedPubkey.isEmpty + ? event.pubkey + : taggedPubkey) + .toLowerCase(); + if (!requested.contains(pubkey) || !_isPresenceStatus(event.content)) { + continue; + } + returned.add(pubkey); + changed = _mergeSnapshotEvent(updated, pubkey, event) || changed; + } + + final fetchedAt = DateTime.now(); + for (final pubkey in pubkeys) { + _snapshotFetchedAt[pubkey] = fetchedAt; + if (returned.contains(pubkey)) continue; + // Do not let an absence from a snapshot overwrite a live event that + // arrived while the HTTP request was in flight. + if ((_liveArrivalRevision[pubkey] ?? 0) != + liveRevisionBeforeQuery[pubkey]) { + continue; + } + if (updated[pubkey] == 'offline') continue; + updated[pubkey] = 'offline'; + changed = true; + } + if (changed) state = updated; + } catch (error) { + if (version == _snapshotVersion) { + // Bound rebuild-triggered retries. Reconnects and the periodic + // backstop still bypass this freshness window with force: true. + final failedAt = DateTime.now(); + for (final pubkey in pubkeys) { + _snapshotFetchedAt[pubkey] = failedAt; + } + } + debugPrint('[PresenceCacheNotifier] presence snapshot failed: $error'); + } finally { + for (final pubkey in requested) { + if (_inFlightSnapshotVersion[pubkey] == version) { + _inFlightSnapshotVersion.remove(pubkey); + } + } + } } /// Subscribe to kind:20001 presence events over WebSocket. @@ -74,14 +203,40 @@ class PresenceCacheNotifier extends Notifier> { void _handlePresenceEvent(NostrEvent event) { final pubkey = event.pubkey.toLowerCase(); - if (!_tracked.contains(pubkey)) return; + if (!_tracked.contains(pubkey) || !_isPresenceStatus(event.content)) { + return; + } + _liveArrivalRevision[pubkey] = (_liveArrivalRevision[pubkey] ?? 0) + 1; + _applyPresenceStatus(pubkey, event.content); + } + + bool _mergeSnapshotEvent( + Map updated, + String pubkey, + NostrEvent event, + ) { + if (!_tracked.contains(pubkey)) return false; final status = event.content; - if (status != 'online' && status != 'away' && status != 'offline') return; + if (!_isPresenceStatus(status)) return false; + final latestCreatedAt = _latestSnapshotCreatedAt[pubkey]; + if (latestCreatedAt != null && latestCreatedAt >= event.createdAt) { + return false; + } + _latestSnapshotCreatedAt[pubkey] = event.createdAt; + if (updated[pubkey] == status) return false; + updated[pubkey] = status; + return true; + } + + void _applyPresenceStatus(String pubkey, String status) { if (state[pubkey] == status) return; final updated = Map.from(state); updated[pubkey] = status; state = updated; } + + bool _isPresenceStatus(String status) => + status == 'online' || status == 'away' || status == 'offline'; } final presenceCacheProvider = diff --git a/mobile/test/features/profile/presence_cache_provider_test.dart b/mobile/test/features/profile/presence_cache_provider_test.dart index 9fc02d9575..375c57dd4d 100644 --- a/mobile/test/features/profile/presence_cache_provider_test.dart +++ b/mobile/test/features/profile/presence_cache_provider_test.dart @@ -1,16 +1,12 @@ +import 'dart:async'; + import 'package:flutter/widgets.dart'; import 'package:flutter_test/flutter_test.dart'; import 'package:hooks_riverpod/hooks_riverpod.dart'; 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. +/// Tests for [PresenceCacheNotifier]'s live updates and snapshot backstop. void main() { test('WS presence event updates cache for tracked pubkey', () async { final relaySession = _RecordingRelaySessionNotifier(); @@ -27,8 +23,352 @@ void main() { expect(container.read(presenceCacheProvider)['alice'], 'online'); // Simulate a WS presence event: alice goes away. - relaySession.emit(_presence('alice', 'away')); + relaySession.emit(_presence('alice', 'away', createdAt: 1001)); + expect(container.read(presenceCacheProvider)['alice'], 'away'); + }); + + test('track batches snapshot queries and maps relay p tags', () async { + final relaySession = _RecordingRelaySessionNotifier() + ..queueQueryResponse([ + _presence( + 'relay-pubkey', + 'online', + createdAt: 1001, + tags: const [ + ['p', 'Alice'], + ], + ), + _presence('bob', 'away', createdAt: 1002), + ]); + final container = _buildContainer(relaySession: relaySession); + addTearDown(container.dispose); + + container.read(presenceCacheProvider); + container.read(presenceCacheProvider.notifier) + ..track(['ALICE']) + ..track(['alice', 'BOB']); + await _pumpSnapshotBatch(); + + expect(relaySession.queryFilters, hasLength(1)); + final filter = relaySession.queryFilters.single.single; + expect(filter.kinds, [EventKind.presenceUpdate]); + expect(filter.authors, ['alice', 'bob']); + expect(container.read(presenceCacheProvider), { + 'alice': 'online', + 'bob': 'away', + }); + }); + + test('snapshot keeps the latest event for each subject', () async { + final relaySession = _RecordingRelaySessionNotifier() + ..queueQueryResponse([ + _presence( + 'relay-pubkey', + 'online', + createdAt: 1002, + tags: const [ + ['p', 'alice'], + ], + ), + _presence( + 'relay-pubkey', + 'away', + createdAt: 1001, + tags: const [ + ['p', 'alice'], + ], + ), + ]); + final container = _buildContainer(relaySession: relaySession); + addTearDown(container.dispose); + + container.read(presenceCacheProvider); + container.read(presenceCacheProvider.notifier).track(['alice']); + await _pumpSnapshotBatch(); + + expect(container.read(presenceCacheProvider)['alice'], 'online'); + }); + + test( + 'clock-skewed live event updates after a newer snapshot stamp', + () async { + final relaySession = _RecordingRelaySessionNotifier() + ..queueQueryResponse([ + _presence( + 'relay-pubkey', + 'online', + createdAt: 2000, + tags: const [ + ['p', 'alice'], + ], + ), + ]); + final container = _buildContainer(relaySession: relaySession); + addTearDown(container.dispose); + + container.read(presenceCacheProvider); + container.read(presenceCacheProvider.notifier).track(['alice']); + await _pumpSnapshotBatch(); + expect(container.read(presenceCacheProvider)['alice'], 'online'); + + relaySession.emit(_presence('alice', 'away', createdAt: 1000)); + + expect(container.read(presenceCacheProvider)['alice'], 'away'); + }, + ); + + test( + 'snapshot marks requested pubkeys absent from the response offline', + () async { + final relaySession = _RecordingRelaySessionNotifier() + ..queueQueryResponse([]); + final container = _buildContainer(relaySession: relaySession); + addTearDown(container.dispose); + + container.read(presenceCacheProvider); + container.read(presenceCacheProvider.notifier).track(['alice']); + relaySession.emit(_presence('alice', 'online')); + await _pumpSnapshotBatch(); + + expect(container.read(presenceCacheProvider)['alice'], 'offline'); + }, + ); + + test( + 'snapshot absence does not overwrite a newer in-flight live event', + () async { + final response = Completer>(); + final relaySession = _RecordingRelaySessionNotifier() + ..queueQueryFuture(response.future); + final container = _buildContainer(relaySession: relaySession); + addTearDown(container.dispose); + + container.read(presenceCacheProvider); + container.read(presenceCacheProvider.notifier).track(['alice']); + relaySession.emit(_presence('alice', 'away', createdAt: 2000)); + await _pumpSnapshotBatch(); + expect(relaySession.queryFilters, hasLength(1)); + + relaySession.emit(_presence('alice', 'online', createdAt: 1000)); + response.complete([]); + await _pumpEventQueue(); + + expect(container.read(presenceCacheProvider)['alice'], 'online'); + }, + ); + + test('reconnect refreshes all tracked pubkeys', () async { + final relaySession = _RecordingRelaySessionNotifier() + ..queueQueryResponse([_presence('alice', 'online')]); + final container = _buildContainer(relaySession: relaySession); + addTearDown(container.dispose); + + container.read(presenceCacheProvider); + container.read(presenceCacheProvider.notifier).track(['alice']); + await _pumpSnapshotBatch(); + expect(container.read(presenceCacheProvider)['alice'], 'online'); + + relaySession.setStatus(SessionStatus.disconnected); + container.read(presenceCacheProvider); + await _pumpEventQueue(); + relaySession.queueQueryResponse([]); + relaySession.setStatus(SessionStatus.connected); + container.read(presenceCacheProvider); + await _pumpSnapshotBatch(); + + expect(relaySession.queryFilters, hasLength(2)); + expect(container.read(presenceCacheProvider)['alice'], 'offline'); + }); + + test('periodic refresh reconciles presence on a stable connection', () async { + final relaySession = _RecordingRelaySessionNotifier() + ..queueQueryResponse([_presence('alice', 'online')]) + ..queueQueryResponse([]); + final container = _buildContainer( + relaySession: relaySession, + refreshInterval: const Duration(milliseconds: 100), + ); + addTearDown(container.dispose); + + container.read(presenceCacheProvider); + container.read(presenceCacheProvider.notifier).track(['alice']); + await _pumpSnapshotBatch(); + expect(container.read(presenceCacheProvider)['alice'], 'online'); + + await _pumpPeriodicRefresh(); + + expect(relaySession.queryFilters, hasLength(2)); + expect(container.read(presenceCacheProvider)['alice'], 'offline'); + }); + + test( + 'periodic refresh stops while disconnected and after disposal', + () async { + final relaySession = _RecordingRelaySessionNotifier() + ..queueQueryResponse([]); + final container = _buildContainer( + relaySession: relaySession, + refreshInterval: const Duration(milliseconds: 100), + ); + + container.read(presenceCacheProvider); + container.read(presenceCacheProvider.notifier).track(['alice']); + await _pumpSnapshotBatch(); + expect(relaySession.queryFilters, hasLength(1)); + + relaySession.setStatus(SessionStatus.disconnected); + container.read(presenceCacheProvider); + await _pumpPeriodicRefresh(); + expect(relaySession.queryFilters, hasLength(1)); + + relaySession.queueQueryResponse([]); + relaySession.setStatus(SessionStatus.connected); + container.read(presenceCacheProvider); + await _pumpSnapshotBatch(); + expect(relaySession.queryFilters, hasLength(2)); + + container.dispose(); + relaySession.queueQueryResponse([]); + await _pumpPeriodicRefresh(); + expect(relaySession.queryFilters, hasLength(2)); + }, + ); + + test('relay snapshot reconciles a subject-clock-ahead live event', () async { + final relaySession = _RecordingRelaySessionNotifier() + ..queueQueryResponse([_presence('alice', 'online', createdAt: 1000)]) + ..queueQueryResponse([_presence('alice', 'online', createdAt: 2000)]); + final container = _buildContainer( + relaySession: relaySession, + refreshInterval: const Duration(milliseconds: 100), + ); + addTearDown(container.dispose); + + container.read(presenceCacheProvider); + container.read(presenceCacheProvider.notifier).track(['alice']); + await _pumpSnapshotBatch(); + relaySession.emit(_presence('alice', 'away', createdAt: 3000)); expect(container.read(presenceCacheProvider)['alice'], 'away'); + + await _pumpPeriodicRefresh(); + + expect(relaySession.queryFilters, hasLength(2)); + expect(container.read(presenceCacheProvider)['alice'], 'online'); + }); + + test('failed snapshot is negatively cached against rebuild churn', () async { + final relaySession = _RecordingRelaySessionNotifier() + ..queueQueryError(StateError('snapshot failed')); + final container = _buildContainer(relaySession: relaySession); + addTearDown(container.dispose); + + container.read(presenceCacheProvider); + final notifier = container.read(presenceCacheProvider.notifier); + notifier.track(['alice']); + await _pumpSnapshotBatch(); + + notifier.track(['alice']); + await _pumpSnapshotBatch(); + + expect(relaySession.queryFilters, hasLength(1)); + }); + + test('snapshot publishes one state update for multiple subjects', () async { + final relaySession = _RecordingRelaySessionNotifier() + ..queueQueryResponse([ + _presence('alice', 'online'), + _presence('bob', 'away'), + ]); + final container = _buildContainer(relaySession: relaySession); + addTearDown(container.dispose); + + container.read(presenceCacheProvider); + var stateChangeCount = 0; + container.listen(presenceCacheProvider, (prev, next) => stateChangeCount++); + + container.read(presenceCacheProvider.notifier).track(['alice', 'bob']); + await _pumpSnapshotBatch(); + + expect(stateChangeCount, 1); + expect(container.read(presenceCacheProvider), { + 'alice': 'online', + 'bob': 'away', + }); + }); + + test( + 'stale pre-reconnect response cannot disturb the new snapshot', + () async { + final staleResponse = Completer>(); + final reconnectResponse = Completer>(); + final relaySession = _RecordingRelaySessionNotifier() + ..queueQueryFuture(staleResponse.future) + ..queueQueryFuture(reconnectResponse.future); + final container = _buildContainer(relaySession: relaySession); + addTearDown(container.dispose); + + container.read(presenceCacheProvider); + final notifier = container.read(presenceCacheProvider.notifier); + notifier.track(['alice']); + await _pumpSnapshotBatch(); + + relaySession.setStatus(SessionStatus.disconnected); + container.read(presenceCacheProvider); + await _pumpEventQueue(); + relaySession.setStatus(SessionStatus.connected); + container.read(presenceCacheProvider); + await _pumpSnapshotBatch(); + expect(relaySession.queryFilters, hasLength(2)); + + staleResponse.complete([_presence('alice', 'away')]); + await _pumpEventQueue(); + notifier.track(['alice']); + await _pumpSnapshotBatch(); + expect(relaySession.queryFilters, hasLength(2)); + + reconnectResponse.complete([_presence('alice', 'online')]); + await _pumpEventQueue(); + expect(container.read(presenceCacheProvider)['alice'], 'online'); + }, + ); + + test('track deduplicates in-flight and recently fetched snapshots', () async { + final response = Completer>(); + final relaySession = _RecordingRelaySessionNotifier() + ..queueQueryFuture(response.future); + final container = _buildContainer(relaySession: relaySession); + addTearDown(container.dispose); + + container.read(presenceCacheProvider); + final notifier = container.read(presenceCacheProvider.notifier); + notifier.track(['alice']); + await _pumpSnapshotBatch(); + notifier.track(['alice']); + await _pumpSnapshotBatch(); + expect(relaySession.queryFilters, hasLength(1)); + + response.complete([_presence('alice', 'online')]); + await _pumpEventQueue(); + notifier.track(['alice']); + await _pumpSnapshotBatch(); + + expect(relaySession.queryFilters, hasLength(1)); + }); + + test('disposing ignores an in-flight snapshot response', () async { + final response = Completer>(); + final relaySession = _RecordingRelaySessionNotifier() + ..queueQueryFuture(response.future); + final container = _buildContainer(relaySession: relaySession); + + container.read(presenceCacheProvider); + container.read(presenceCacheProvider.notifier).track(['alice']); + await _pumpSnapshotBatch(); + expect(relaySession.queryFilters, hasLength(1)); + + container.dispose(); + response.complete([_presence('alice', 'online')]); + await _pumpEventQueue(); }); test('WS presence event ignores untracked pubkeys', () async { @@ -133,12 +473,17 @@ void main() { }); } -NostrEvent _presence(String pubkey, String status) => NostrEvent( +NostrEvent _presence( + String pubkey, + String status, { + int createdAt = 1000, + List> tags = const [], +}) => NostrEvent( id: 'evt-$pubkey-$status', pubkey: pubkey, - createdAt: 1000, + createdAt: createdAt, kind: EventKind.presenceUpdate, - tags: const [], + tags: tags, content: status, sig: 'sig', ); @@ -148,19 +493,37 @@ Future _pumpEventQueue() async { await Future.delayed(Duration.zero); } +Future _pumpSnapshotBatch() async { + await Future.delayed(const Duration(milliseconds: 75)); + await _pumpEventQueue(); +} + +Future _pumpPeriodicRefresh() async { + await Future.delayed(const Duration(milliseconds: 110)); + await _pumpEventQueue(); +} + ProviderContainer _buildContainer({ required _RecordingRelaySessionNotifier relaySession, + Duration? refreshInterval, }) { return ProviderContainer( overrides: [ appLifecycleProvider.overrideWith(() => _FakeAppLifecycleNotifier()), relaySessionProvider.overrideWith(() => relaySession), + if (refreshInterval != null) + presenceCacheProvider.overrideWith( + () => PresenceCacheNotifier(refreshInterval: refreshInterval), + ), ], ); } class _RecordingRelaySessionNotifier extends RelaySessionNotifier { final List filters = []; + final List> queryFilters = []; + final List>> _queryResponses = []; + final List _queryErrors = []; final List _listeners = []; @override @@ -180,6 +543,33 @@ class _RecordingRelaySessionNotifier extends RelaySessionNotifier { }; } + @override + Future> queryRelay( + List filters, { + Duration timeout = const Duration(seconds: 8), + }) async { + queryFilters.add(filters); + if (_queryErrors.isNotEmpty) throw _queryErrors.removeAt(0); + if (_queryResponses.isEmpty) return []; + return _queryResponses.removeAt(0); + } + + void queueQueryResponse(List events) { + _queryResponses.add(Future.value(events)); + } + + void queueQueryFuture(Future> events) { + _queryResponses.add(events); + } + + void queueQueryError(Object error) { + _queryErrors.add(error); + } + + void setStatus(SessionStatus status) { + state = SessionState(status: status); + } + /// Emit an event synchronously to all live subscribers. void emit(NostrEvent event) { for (final listener in List.of(_listeners)) {