From 046221d25731e96a2762ab94203bc847f1298f89 Mon Sep 17 00:00:00 2001 From: Naadir Ally Date: Mon, 3 Aug 2026 00:18:30 +0200 Subject: [PATCH] fix(mobile): bootstrap presence snapshots reliably Signed-off-by: Naadir Ally --- .../channels/channel_detail_page/app_bar.dart | 10 +- .../channels/channels_page/channel_tile.dart | 9 +- .../profile/presence_cache_provider.dart | 269 ++++++++++++- .../features/profile/user_profile_sheet.dart | 8 +- .../channels/channel_detail_page_test.dart | 84 +++++ .../profile/presence_cache_provider_test.dart | 356 +++++++++++++++++- .../profile/user_profile_sheet_test.dart | 93 +++++ 7 files changed, 782 insertions(+), 47 deletions(-) create mode 100644 mobile/test/features/profile/user_profile_sheet_test.dart diff --git a/mobile/lib/features/channels/channel_detail_page/app_bar.dart b/mobile/lib/features/channels/channel_detail_page/app_bar.dart index 406d68b4f56..b39720d1543 100644 --- a/mobile/lib/features/channels/channel_detail_page/app_bar.dart +++ b/mobile/lib/features/channels/channel_detail_page/app_bar.dart @@ -107,13 +107,12 @@ class _DmAppBarTitle extends ConsumerWidget { (channel.participants.isNotEmpty ? channel.participants.first[0].toUpperCase() : '?'); - final presence = otherPubkey != null - ? (presenceMap[otherPubkey] ?? 'offline') - : 'offline'; + final presence = otherPubkey != null ? presenceMap[otherPubkey] : null; final presenceLabel = switch (presence) { 'online' => 'Online', 'away' => 'Away', - _ => 'Offline', + 'offline' => 'Offline', + _ => 'Status unknown', }; return Row( @@ -146,7 +145,8 @@ class _DmAppBarTitle extends ConsumerWidget { color: switch (presence) { 'online' => context.appColors.success, 'away' => context.appColors.warning, - _ => context.colors.outline, + 'offline' => context.colors.outline, + _ => context.colors.outlineVariant, }, shape: BoxShape.circle, border: Border.all( diff --git a/mobile/lib/features/channels/channels_page/channel_tile.dart b/mobile/lib/features/channels/channels_page/channel_tile.dart index 551e3f8dd69..67cc200f133 100644 --- a/mobile/lib/features/channels/channels_page/channel_tile.dart +++ b/mobile/lib/features/channels/channels_page/channel_tile.dart @@ -386,9 +386,7 @@ class _DmAvatar extends ConsumerWidget { (channel.participants.isNotEmpty ? channel.participants.first[0].toUpperCase() : '?'); - final presence = otherPubkey != null - ? (presenceMap[otherPubkey] ?? 'offline') - : 'offline'; + final presence = otherPubkey != null ? presenceMap[otherPubkey] : null; return SizedBox( width: _kDmAvatarSize, @@ -430,11 +428,12 @@ class _DmAvatar extends ConsumerWidget { ); } - Color _presenceColor(BuildContext context, String presence) { + Color _presenceColor(BuildContext context, String? presence) { return switch (presence) { 'online' => context.appColors.success, 'away' => context.appColors.warning, - _ => context.colors.outline, + 'offline' => context.colors.outline, + _ => context.colors.outlineVariant, }; } } diff --git a/mobile/lib/features/profile/presence_cache_provider.dart b/mobile/lib/features/profile/presence_cache_provider.dart index f735c376c35..a33b06f5959 100644 --- a/mobile/lib/features/profile/presence_cache_provider.dart +++ b/mobile/lib/features/profile/presence_cache_provider.dart @@ -1,4 +1,5 @@ import 'dart:async'; +import 'dart:math'; import 'package:flutter/foundation.dart'; import 'package:hooks_riverpod/hooks_riverpod.dart'; @@ -8,25 +9,52 @@ 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, then seeds newly tracked pubkeys from the relay's +/// synthesized current-presence snapshots over `POST /query`. class PresenceCacheNotifier extends Notifier> { + static const _maxRetryDelayMs = 30000; + static const _subscriptionStabilityWindow = Duration(seconds: 30); + + /// Creates a presence cache. + /// + /// [subscriptionRetryBaseDelay] is configurable so retry behavior can be + /// exercised deterministically in tests. [snapshotRetryBaseDelay] controls + /// the equivalent retry path for one-shot snapshot failures. + PresenceCacheNotifier({ + Duration subscriptionRetryBaseDelay = const Duration(seconds: 1), + Duration snapshotRetryBaseDelay = const Duration(seconds: 1), + }) : _subscriptionRetryBaseDelay = subscriptionRetryBaseDelay, + _snapshotRetryBaseDelay = snapshotRetryBaseDelay; + + final Duration _subscriptionRetryBaseDelay; + final Duration _snapshotRetryBaseDelay; final Set _tracked = {}; + final Set _snapshotRequested = {}; + final Map _liveUpdateVersions = {}; void Function()? _presenceUnsub; + Future? _presenceSubscriptionReady; + Timer? _subscriptionRetryTimer; + Timer? _subscriptionStabilityTimer; + Timer? _snapshotRetryTimer; int _subscriptionVersion = 0; + int _subscriptionRetryAttempt = 0; + int _snapshotRetryAttempt = 0; + bool _disposed = false; @override Map build() { + _disposed = false; final sessionState = ref.watch(relaySessionProvider); ref.onDispose(() { - _presenceUnsub?.call(); - _presenceUnsub = null; + _disposed = true; + _stopPresenceSubscription(clearSnapshots: false); }); if (sessionState.status == SessionStatus.connected) { - _subscribePresenceUpdates(); + _startPresenceSubscription(resetBackoff: true, resetSnapshots: true); + } else { + _stopPresenceSubscription(clearSnapshots: true); } return {}; @@ -34,49 +62,256 @@ class PresenceCacheNotifier extends Notifier> { /// 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. + /// The live subscription is established before a one-shot snapshot query is + /// issued, closing the gap where an update could land between the seed and + /// subscription. Duplicate calls are coalesced; failed requests retry with + /// capped backoff. void track(List pubkeys) { - final normalized = pubkeys.map((pk) => pk.toLowerCase()).toList(); + final normalized = pubkeys + .map((pk) => pk.trim().toLowerCase()) + .where((pk) => pk.isNotEmpty); _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". + unawaited(_fetchPendingSnapshots()); + } + + void _startPresenceSubscription({ + bool resetBackoff = false, + bool resetSnapshots = false, + }) { + if (_disposed || + ref.read(relaySessionProvider).status != SessionStatus.connected) { + return; + } + _subscriptionRetryTimer?.cancel(); + _subscriptionRetryTimer = null; + _subscriptionStabilityTimer?.cancel(); + _subscriptionStabilityTimer = null; + if (resetBackoff) _subscriptionRetryAttempt = 0; + if (resetSnapshots) _snapshotRequested.clear(); + + final ready = _subscribePresenceUpdates(); + _presenceSubscriptionReady = ready; + unawaited( + ready.then((subscribed) { + if (_disposed || !identical(ready, _presenceSubscriptionReady)) return; + if (subscribed) { + _scheduleSubscriptionStabilityReset(ready); + unawaited(_fetchPendingSnapshots()); + } else { + _scheduleSubscriptionRetry(); + } + }), + ); + } + + void _stopPresenceSubscription({required bool clearSnapshots}) { + _subscriptionRetryTimer?.cancel(); + _subscriptionRetryTimer = null; + _subscriptionStabilityTimer?.cancel(); + _subscriptionStabilityTimer = null; + _snapshotRetryTimer?.cancel(); + _snapshotRetryTimer = null; + _presenceUnsub?.call(); + _presenceUnsub = null; + _presenceSubscriptionReady = null; + _subscriptionVersion++; + _subscriptionRetryAttempt = 0; + _snapshotRetryAttempt = 0; + if (clearSnapshots) _snapshotRequested.clear(); } /// Subscribe to kind:20001 presence events over WebSocket. - Future _subscribePresenceUpdates() async { + Future _subscribePresenceUpdates() async { _presenceUnsub?.call(); _presenceUnsub = null; _subscriptionVersion++; final version = _subscriptionVersion; - final session = ref.read(relaySessionProvider.notifier); try { + if (_disposed) return false; + final session = ref.read(relaySessionProvider.notifier); final unsub = await session.subscribe( const NostrFilter(kinds: [EventKind.presenceUpdate], limit: 0), _handlePresenceEvent, + onClosed: (message) => _handlePresenceClosed(version, message), ); // Guard: if build() re-fired while we were awaiting, discard this // subscription to avoid leaking it. - if (version != _subscriptionVersion) { + if (_disposed || version != _subscriptionVersion) { unsub(); - return; + return false; } _presenceUnsub = unsub; + return true; } catch (error) { debugPrint( '[PresenceCacheNotifier] presence subscription failed: $error', ); + return false; } } + void _handlePresenceClosed(int version, String message) { + if (_disposed || version != _subscriptionVersion) return; + _subscriptionStabilityTimer?.cancel(); + _subscriptionStabilityTimer = null; + _snapshotRetryTimer?.cancel(); + _snapshotRetryTimer = null; + _presenceUnsub = null; + _presenceSubscriptionReady = null; + _subscriptionVersion++; + _snapshotRetryAttempt = 0; + _snapshotRequested.clear(); + _clearTrackedStatuses(); + debugPrint( + '[PresenceCacheNotifier] presence subscription closed: $message', + ); + _scheduleSubscriptionRetry(); + } + + void _scheduleSubscriptionRetry() { + if (_disposed || + _subscriptionRetryTimer != null || + ref.read(relaySessionProvider).status != SessionStatus.connected) { + return; + } + + final exponent = min(_subscriptionRetryAttempt, 10); + final delayMs = min( + _subscriptionRetryBaseDelay.inMilliseconds * (1 << exponent), + _maxRetryDelayMs, + ); + _subscriptionRetryAttempt++; + _subscriptionRetryTimer = Timer(Duration(milliseconds: delayMs), () { + _subscriptionRetryTimer = null; + if (_disposed) return; + _startPresenceSubscription(); + }); + } + + void _scheduleSubscriptionStabilityReset(Future ready) { + _subscriptionStabilityTimer?.cancel(); + _subscriptionStabilityTimer = Timer(_subscriptionStabilityWindow, () { + _subscriptionStabilityTimer = null; + if (_disposed || !identical(ready, _presenceSubscriptionReady)) return; + _subscriptionRetryAttempt = 0; + }); + } + + void _scheduleSnapshotRetry() { + if (_disposed || + _snapshotRetryTimer != null || + ref.read(relaySessionProvider).status != SessionStatus.connected) { + return; + } + + final exponent = min(_snapshotRetryAttempt, 10); + final delayMs = min( + _snapshotRetryBaseDelay.inMilliseconds * (1 << exponent), + _maxRetryDelayMs, + ); + _snapshotRetryAttempt++; + _snapshotRetryTimer = Timer(Duration(milliseconds: delayMs), () { + _snapshotRetryTimer = null; + if (_disposed) return; + unawaited(_fetchPendingSnapshots()); + }); + } + void _handlePresenceEvent(NostrEvent event) { + if (_disposed) return; final pubkey = event.pubkey.toLowerCase(); if (!_tracked.contains(pubkey)) return; final status = event.content; if (status != 'online' && status != 'away' && status != 'offline') return; + + // Advance even for a no-op status so a snapshot request already in flight + // cannot overwrite a newer live event with stale relay state. + _liveUpdateVersions[pubkey] = (_liveUpdateVersions[pubkey] ?? 0) + 1; + _applyStatus(pubkey, status); + } + + Future _fetchPendingSnapshots() async { + if (_disposed) return; + final ready = _presenceSubscriptionReady; + if (ready == null || !await ready) return; + if (_disposed || !identical(ready, _presenceSubscriptionReady)) return; + final subscriptionVersion = _subscriptionVersion; + + final pubkeys = _tracked.difference(_snapshotRequested).toList()..sort(); + if (pubkeys.isEmpty) return; + + _snapshotRetryTimer?.cancel(); + _snapshotRetryTimer = null; + + // Claim before yielding to queryRelay so concurrent track() calls cannot + // request the same pubkey twice. + _snapshotRequested.addAll(pubkeys); + final liveVersions = { + for (final pubkey in pubkeys) pubkey: _liveUpdateVersions[pubkey] ?? 0, + }; + + try { + if (_disposed) return; + final session = ref.read(relaySessionProvider.notifier); + final events = await session.queryRelay([ + NostrFilter( + kinds: const [EventKind.presenceUpdate], + authors: pubkeys, + limit: pubkeys.length, + ), + ]); + if (_disposed || subscriptionVersion != _subscriptionVersion) return; + + final resolvedPubkeys = {}; + for (final event in events) { + if (event.kind != EventKind.presenceUpdate) continue; + final pubkey = (event.getTagValue('p') ?? event.pubkey) + .trim() + .toLowerCase(); + if (!liveVersions.containsKey(pubkey) || !_tracked.contains(pubkey)) { + continue; + } + if ((_liveUpdateVersions[pubkey] ?? 0) != liveVersions[pubkey]) { + continue; + } + final status = event.content; + if (status != 'online' && status != 'away' && status != 'offline') { + continue; + } + resolvedPubkeys.add(pubkey); + _applyStatus(pubkey, status); + } + + // Redis omits expired/absent presence. Once the query succeeds, that + // absence is the authoritative offline state unless a live event arrived + // while the request was in flight. + for (final pubkey in pubkeys) { + if (resolvedPubkeys.contains(pubkey)) continue; + if ((_liveUpdateVersions[pubkey] ?? 0) != liveVersions[pubkey]) { + continue; + } + _applyStatus(pubkey, 'offline'); + } + _snapshotRetryAttempt = 0; + } catch (error) { + if (!_disposed && subscriptionVersion == _subscriptionVersion) { + _snapshotRequested.removeAll(pubkeys); + debugPrint('[PresenceCacheNotifier] presence snapshot failed: $error'); + _scheduleSnapshotRetry(); + } + } + } + + void _clearTrackedStatuses() { + if (_disposed || !state.keys.any(_tracked.contains)) return; + final updated = Map.from(state) + ..removeWhere((pubkey, _) => _tracked.contains(pubkey)); + state = updated; + } + + void _applyStatus(String pubkey, String status) { + if (_disposed) return; if (state[pubkey] == status) return; final updated = Map.from(state); updated[pubkey] = status; diff --git a/mobile/lib/features/profile/user_profile_sheet.dart b/mobile/lib/features/profile/user_profile_sheet.dart index bce99dc8409..a0aa929f71f 100644 --- a/mobile/lib/features/profile/user_profile_sheet.dart +++ b/mobile/lib/features/profile/user_profile_sheet.dart @@ -40,7 +40,7 @@ class UserProfileSheet extends HookConsumerWidget { ref.watch(userCacheProvider.select((cache) => cache[pk])) ?? ref.read(userCacheProvider.notifier).get(pk); final presenceMap = ref.watch(presenceCacheProvider); - final presence = presenceMap[pk] ?? 'offline'; + final presence = presenceMap[pk]; final statusCache = ref.watch(userStatusCacheProvider); final userStatus = statusCache[pk]; @@ -76,12 +76,14 @@ class UserProfileSheet extends HookConsumerWidget { final presenceColor = switch (presence) { 'online' => context.appColors.success, 'away' => context.appColors.warning, - _ => context.colors.outline, + 'offline' => context.colors.outline, + _ => context.colors.outlineVariant, }; final presenceLabel = switch (presence) { 'online' => 'Online', 'away' => 'Away', - _ => 'Offline', + 'offline' => 'Offline', + _ => 'Status unknown', }; return SizedBox( diff --git a/mobile/test/features/channels/channel_detail_page_test.dart b/mobile/test/features/channels/channel_detail_page_test.dart index f95c6fefed1..34c6655c26f 100644 --- a/mobile/test/features/channels/channel_detail_page_test.dart +++ b/mobile/test/features/channels/channel_detail_page_test.dart @@ -23,6 +23,7 @@ import 'package:buzz/features/channels/timeline_message.dart'; import 'package:buzz/features/channels/channels_provider.dart'; import 'package:buzz/features/channels/read_state/read_state_provider.dart'; import 'package:buzz/features/channels/small_avatar.dart'; +import 'package:buzz/features/profile/presence_cache_provider.dart'; import 'package:buzz/features/profile/profile_provider.dart'; import 'package:buzz/features/profile/user_cache_provider.dart'; import 'package:buzz/features/profile/user_profile.dart'; @@ -159,6 +160,7 @@ Widget _buildTestable({ Map>> pendingThreadReplies = const {}, TextScaler textScaler = TextScaler.noScaling, RelaySessionNotifier? relaySessionNotifier, + Map? presence, }) { final resolvedChannel = channel ?? _testChannel; final fakeChannelsNotifier = @@ -175,6 +177,10 @@ Widget _buildTestable({ ).overrideWith(() => _FakeTypingNotifier(typing)), userCacheProvider.overrideWith(() => _FakeUserCacheNotifier(users)), profileProvider.overrideWith(() => _FakeProfileNotifier()), + if (presence != null) + presenceCacheProvider.overrideWith( + () => _FakePresenceCacheNotifier(presence), + ), channelsProvider.overrideWith(() => fakeChannelsNotifier), channelDetailsProvider(_channelId).overrideWith( (ref) async => ChannelDetails.fromChannel(resolvedChannel), @@ -408,6 +414,72 @@ void main() { }, ); + testWidgets('does not label missing DM presence as offline', ( + tester, + ) async { + final dmChannel = Channel( + id: _channelId, + name: 'DM', + channelType: 'dm', + visibility: 'private', + description: '', + createdBy: 'self', + createdAt: DateTime(2025), + memberCount: 2, + participants: const ['Self', 'Fable'], + participantPubkeys: const ['self', 'fable'], + isMember: true, + ); + + await tester.pumpWidget( + _buildTestable( + messages: const [], + channel: dmChannel, + users: const { + 'fable': UserProfile(pubkey: 'fable', displayName: 'Fable'), + }, + presence: const {}, + ), + ); + await tester.pumpAndSettle(); + + expect(find.text('Status unknown'), findsOneWidget); + expect(find.text('Offline'), findsNothing); + }); + + testWidgets('labels an explicit DM offline event as offline', ( + tester, + ) async { + final dmChannel = Channel( + id: _channelId, + name: 'DM', + channelType: 'dm', + visibility: 'private', + description: '', + createdBy: 'self', + createdAt: DateTime(2025), + memberCount: 2, + participants: const ['Self', 'Fable'], + participantPubkeys: const ['self', 'fable'], + isMember: true, + ); + + await tester.pumpWidget( + _buildTestable( + messages: const [], + channel: dmChannel, + users: const { + 'fable': UserProfile(pubkey: 'fable', displayName: 'Fable'), + }, + presence: const {'fable': 'offline'}, + ), + ); + await tester.pumpAndSettle(); + + expect(find.text('Offline'), findsOneWidget); + expect(find.text('Status unknown'), findsNothing); + }); + testWidgets('debounces same-slot reconnect skeletons before revealing', ( tester, ) async { @@ -2895,6 +2967,18 @@ class _FakeProfileNotifier extends ProfileNotifier { const UserProfile(pubkey: 'self', displayName: 'Self'); } +class _FakePresenceCacheNotifier extends PresenceCacheNotifier { + final Map _presence; + + _FakePresenceCacheNotifier(this._presence); + + @override + Map build() => _presence; + + @override + void track(List pubkeys) {} +} + class _FakeUserCacheNotifier extends UserCacheNotifier { final Map _users; _FakeUserCacheNotifier(this._users); diff --git a/mobile/test/features/profile/presence_cache_provider_test.dart b/mobile/test/features/profile/presence_cache_provider_test.dart index 9fc02d9575b..5f8210210af 100644 --- a/mobile/test/features/profile/presence_cache_provider_test.dart +++ b/mobile/test/features/profile/presence_cache_provider_test.dart @@ -1,17 +1,255 @@ +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 subscribe-first live + snapshot flow. void main() { + test('current online snapshot seeds a newly tracked pubkey', () async { + final relaySession = _RecordingRelaySessionNotifier( + queryResult: [_presenceSnapshot('relay-signer', 'alice', 'online')], + ); + final container = _buildContainer(relaySession: relaySession); + addTearDown(container.dispose); + + container.read(presenceCacheProvider); + await _pumpEventQueue(); + + container.read(presenceCacheProvider.notifier).track(['alice']); + await _pumpEventQueue(); + + expect(container.read(presenceCacheProvider)['alice'], 'online'); + expect(relaySession.operations, ['subscribe', 'query']); + }); + + test('waits for the live subscription before querying snapshots', () async { + final subscriptionReady = Completer(); + final relaySession = _RecordingRelaySessionNotifier( + subscribeGate: subscriptionReady, + ); + final container = _buildContainer(relaySession: relaySession); + addTearDown(container.dispose); + + container.read(presenceCacheProvider); + container.read(presenceCacheProvider.notifier).track(['alice']); + await _pumpEventQueue(); + + expect(relaySession.operations, ['subscribe']); + expect(relaySession.queries, isEmpty); + + subscriptionReady.complete(); + await _pumpEventQueue(); + + expect(relaySession.operations, ['subscribe', 'query']); + }); + + test('relay-signed snapshot uses p-tag subject, not event author', () async { + final relaySession = _RecordingRelaySessionNotifier( + queryResult: [_presenceSnapshot('relay-signer', 'alice', 'away')], + ); + final container = _buildContainer(relaySession: relaySession); + addTearDown(container.dispose); + + container.read(presenceCacheProvider); + await _pumpEventQueue(); + container.read(presenceCacheProvider.notifier).track(['alice']); + await _pumpEventQueue(); + + final cache = container.read(presenceCacheProvider); + expect(cache['alice'], 'away'); + expect(cache.containsKey('relay-signer'), isFalse); + }); + + test('live update wins when snapshot query completes later', () async { + final queryCompleter = Completer>(); + final relaySession = _RecordingRelaySessionNotifier( + queryHandler: (_) => queryCompleter.future, + ); + final container = _buildContainer(relaySession: relaySession); + addTearDown(container.dispose); + + container.read(presenceCacheProvider); + await _pumpEventQueue(); + container.read(presenceCacheProvider.notifier).track(['alice']); + await _pumpEventQueue(); + + relaySession.emit(_presence('alice', 'away')); + queryCompleter.complete([ + _presenceSnapshot('relay-signer', 'alice', 'online'), + ]); + await _pumpEventQueue(); + + expect(container.read(presenceCacheProvider)['alice'], 'away'); + }); + + test('tracking the same pubkey does not repeat the snapshot query', () async { + final relaySession = _RecordingRelaySessionNotifier(); + final container = _buildContainer(relaySession: relaySession); + addTearDown(container.dispose); + + container.read(presenceCacheProvider); + await _pumpEventQueue(); + + container.read(presenceCacheProvider.notifier).track(['alice']); + container.read(presenceCacheProvider.notifier).track(['ALICE', 'alice']); + await _pumpEventQueue(); + container.read(presenceCacheProvider.notifier).track(['alice']); + await _pumpEventQueue(); + + expect(relaySession.queries, hasLength(1)); + expect(relaySession.queries.single.single.authors, ['alice']); + }); + + test('successful empty snapshot resolves tracked pubkey offline', () async { + final relaySession = _RecordingRelaySessionNotifier(); + final container = _buildContainer(relaySession: relaySession); + addTearDown(container.dispose); + + container.read(presenceCacheProvider); + await _pumpEventQueue(); + container.read(presenceCacheProvider.notifier).track(['alice']); + await _pumpEventQueue(); + + expect(container.read(presenceCacheProvider)['alice'], 'offline'); + }); + + test('failed snapshot query retries automatically', () async { + var attempts = 0; + final relaySession = _RecordingRelaySessionNotifier( + queryHandler: (_) { + attempts++; + if (attempts == 1) { + return Future.error(Exception('temporary query failure')); + } + return Future.value([ + _presenceSnapshot('relay-signer', 'alice', 'online'), + ]); + }, + ); + final container = _buildContainer( + relaySession: relaySession, + snapshotRetryBaseDelay: Duration.zero, + ); + addTearDown(container.dispose); + + container.read(presenceCacheProvider); + await _pumpEventQueue(); + container.read(presenceCacheProvider.notifier).track(['alice']); + await _pumpEventQueue(); + + expect(attempts, 2); + expect(container.read(presenceCacheProvider)['alice'], 'online'); + }); + + test( + 'initial subscription failure retries while session stays connected', + () async { + final relaySession = _RecordingRelaySessionNotifier( + subscribeFailuresRemaining: 1, + ); + final container = _buildContainer( + relaySession: relaySession, + subscriptionRetryBaseDelay: Duration.zero, + ); + addTearDown(container.dispose); + + container.read(presenceCacheProvider); + container.read(presenceCacheProvider.notifier).track(['alice']); + await _pumpEventQueue(); + + expect( + relaySession.operations.where((operation) => operation == 'subscribe'), + hasLength(2), + ); + expect(relaySession.queries, hasLength(1)); + expect(container.read(presenceCacheProvider)['alice'], 'offline'); + }, + ); + + test( + 'late CLOSED resubscribes, resnapshots, and resumes live updates', + () async { + final relaySession = _RecordingRelaySessionNotifier( + queryResult: [_presenceSnapshot('relay-signer', 'alice', 'online')], + ); + final container = _buildContainer( + relaySession: relaySession, + subscriptionRetryBaseDelay: Duration.zero, + ); + addTearDown(container.dispose); + + container.read(presenceCacheProvider); + await _pumpEventQueue(); + container.read(presenceCacheProvider.notifier).track(['alice']); + await _pumpEventQueue(); + expect(container.read(presenceCacheProvider)['alice'], 'online'); + + relaySession.closeLatest('relay maintenance'); + expect( + container.read(presenceCacheProvider).containsKey('alice'), + isFalse, + ); + await _pumpEventQueue(); + + expect( + relaySession.operations.where((operation) => operation == 'subscribe'), + hasLength(2), + ); + expect(relaySession.queries, hasLength(2)); + + relaySession.emit(_presence('alice', 'away')); + expect(container.read(presenceCacheProvider)['alice'], 'away'); + }, + ); + + test( + 'dispose invalidates a delayed subscription without leaking it', + () async { + final subscriptionReady = Completer(); + final relaySession = _RecordingRelaySessionNotifier( + subscribeGate: subscriptionReady, + ); + final container = _buildContainer(relaySession: relaySession); + + container.read(presenceCacheProvider); + container.read(presenceCacheProvider.notifier).track(['alice']); + await _pumpEventQueue(); + container.dispose(); + + subscriptionReady.complete(); + await _pumpEventQueue(); + + expect(relaySession.unsubscribeCount, 1); + expect(relaySession.activeSubscriptionCount, 0); + expect(relaySession.queries, isEmpty); + }, + ); + + test('dispose invalidates an in-flight snapshot query', () async { + final queryCompleter = Completer>(); + final relaySession = _RecordingRelaySessionNotifier( + queryHandler: (_) => queryCompleter.future, + ); + final container = _buildContainer(relaySession: relaySession); + + container.read(presenceCacheProvider); + await _pumpEventQueue(); + container.read(presenceCacheProvider.notifier).track(['alice']); + await _pumpEventQueue(); + expect(relaySession.queries, hasLength(1)); + + container.dispose(); + queryCompleter.complete([ + _presenceSnapshot('relay-signer', 'alice', 'online'), + ]); + await _pumpEventQueue(); + + expect(relaySession.activeSubscriptionCount, 0); + }); + test('WS presence event updates cache for tracked pubkey', () async { final relaySession = _RecordingRelaySessionNotifier(); final container = _buildContainer(relaySession: relaySession); @@ -143,25 +381,72 @@ NostrEvent _presence(String pubkey, String status) => NostrEvent( sig: 'sig', ); +NostrEvent _presenceSnapshot( + String relayPubkey, + String subjectPubkey, + String status, +) => NostrEvent( + id: 'snapshot-$subjectPubkey-$status', + pubkey: relayPubkey, + createdAt: 1000, + kind: EventKind.presenceUpdate, + tags: [ + ['p', subjectPubkey], + ], + content: status, + sig: 'relay-sig', +); + Future _pumpEventQueue() async { - await Future.delayed(Duration.zero); - await Future.delayed(Duration.zero); + for (var i = 0; i < 5; i++) { + await Future.delayed(Duration.zero); + } } ProviderContainer _buildContainer({ required _RecordingRelaySessionNotifier relaySession, + Duration subscriptionRetryBaseDelay = const Duration(seconds: 1), + Duration snapshotRetryBaseDelay = const Duration(seconds: 1), }) { return ProviderContainer( overrides: [ appLifecycleProvider.overrideWith(() => _FakeAppLifecycleNotifier()), relaySessionProvider.overrideWith(() => relaySession), + presenceCacheProvider.overrideWith( + () => PresenceCacheNotifier( + subscriptionRetryBaseDelay: subscriptionRetryBaseDelay, + snapshotRetryBaseDelay: snapshotRetryBaseDelay, + ), + ), ], ); } class _RecordingRelaySessionNotifier extends RelaySessionNotifier { - final List filters = []; - final List _listeners = []; + _RecordingRelaySessionNotifier({ + List queryResult = const [], + Future> Function(List)? queryHandler, + Completer? subscribeGate, + int subscribeFailuresRemaining = 0, + }) : _queryResult = queryResult, + _queryHandler = queryHandler, + _subscribeGate = subscribeGate, + _subscribeFailuresRemaining = subscribeFailuresRemaining; + + final List _queryResult; + final Future> Function(List)? _queryHandler; + final Completer? _subscribeGate; + int _subscribeFailuresRemaining; + final List<_RecordedSubscription> _subscriptions = []; + final List> queries = []; + final List operations = []; + int unsubscribeCount = 0; + + List get filters => [ + for (final subscription in _subscriptions) subscription.filter, + ]; + + int get activeSubscriptionCount => _subscriptions.length; @override SessionState build() => const SessionState(status: SessionStatus.connected); @@ -172,20 +457,57 @@ class _RecordingRelaySessionNotifier extends RelaySessionNotifier { void Function(NostrEvent) onEvent, { void Function(String message)? onClosed, }) async { - filters.add(filter); - _listeners.add(onEvent); + operations.add('subscribe'); + await _subscribeGate?.future; + if (_subscribeFailuresRemaining > 0) { + _subscribeFailuresRemaining--; + throw Exception('temporary subscription failure'); + } + final subscription = _RecordedSubscription( + filter: filter, + onEvent: onEvent, + onClosed: onClosed, + ); + _subscriptions.add(subscription); return () { - filters.remove(filter); - _listeners.remove(onEvent); + if (!_subscriptions.remove(subscription)) return; + unsubscribeCount++; }; } + @override + Future> queryRelay( + List filters, { + Duration timeout = const Duration(seconds: 8), + }) { + operations.add('query'); + 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)) { - listener(event); + for (final subscription in List.of(_subscriptions)) { + subscription.onEvent(event); } } + + void closeLatest(String message) { + final subscription = _subscriptions.removeLast(); + subscription.onClosed?.call(message); + } +} + +class _RecordedSubscription { + final NostrFilter filter; + final void Function(NostrEvent) onEvent; + final void Function(String message)? onClosed; + + const _RecordedSubscription({ + required this.filter, + required this.onEvent, + required this.onClosed, + }); } class _FakeAppLifecycleNotifier extends AppLifecycleNotifier { diff --git a/mobile/test/features/profile/user_profile_sheet_test.dart b/mobile/test/features/profile/user_profile_sheet_test.dart new file mode 100644 index 00000000000..f0e70eda5a0 --- /dev/null +++ b/mobile/test/features/profile/user_profile_sheet_test.dart @@ -0,0 +1,93 @@ +import 'package:flutter/material.dart'; +import 'package:flutter_test/flutter_test.dart'; +import 'package:hooks_riverpod/hooks_riverpod.dart'; +import 'package:buzz/features/channels/channel_management_provider.dart'; +import 'package:buzz/features/profile/presence_cache_provider.dart'; +import 'package:buzz/features/profile/user_cache_provider.dart'; +import 'package:buzz/features/profile/user_profile.dart'; +import 'package:buzz/features/profile/user_profile_sheet.dart'; +import 'package:buzz/features/profile/user_status.dart'; +import 'package:buzz/features/profile/user_status_cache_provider.dart'; +import 'package:buzz/shared/relay/relay.dart'; +import 'package:buzz/shared/theme/theme.dart'; + +void main() { + testWidgets('does not label missing profile presence as offline', ( + tester, + ) async { + await tester.pumpWidget( + ProviderScope( + overrides: [ + currentPubkeyProvider.overrideWithValue('self'), + relaySessionProvider.overrideWith(() => _FakeRelaySessionNotifier()), + presenceCacheProvider.overrideWith( + () => _FakePresenceCacheNotifier(const {}), + ), + userCacheProvider.overrideWith( + () => _FakeUserCacheNotifier(const { + 'fable': UserProfile(pubkey: 'fable', displayName: 'Fable'), + }), + ), + userStatusCacheProvider.overrideWith( + () => _FakeUserStatusCacheNotifier(), + ), + ], + child: MaterialApp( + theme: AppTheme.light(), + home: const Scaffold(body: UserProfileSheet(pubkey: 'fable')), + ), + ), + ); + await tester.pumpAndSettle(); + + expect(find.text('Status unknown'), findsOneWidget); + expect(find.text('Offline'), findsNothing); + }); +} + +class _FakeRelaySessionNotifier extends RelaySessionNotifier { + @override + SessionState build() => + const SessionState(status: SessionStatus.disconnected); + + @override + Future> fetchHistory( + NostrFilter filter, { + Duration timeout = const Duration(seconds: 8), + }) async => const []; +} + +class _FakePresenceCacheNotifier extends PresenceCacheNotifier { + final Map _presence; + + _FakePresenceCacheNotifier(this._presence); + + @override + Map build() => _presence; + + @override + void track(List pubkeys) {} +} + +class _FakeUserCacheNotifier extends UserCacheNotifier { + final Map _users; + + _FakeUserCacheNotifier(this._users); + + @override + Map build() => _users; + + @override + UserProfile? get(String pubkey) => _users[pubkey.toLowerCase()]; + + @override + void preload(List pubkeys) {} +} + +class _FakeUserStatusCacheNotifier extends UserStatusCacheNotifier { + @override + Map build() => const {}; + + @override + void track(List pubkeys) {} +}