From 3ff2f290bac41811fe2db0b141961148740ebce5 Mon Sep 17 00:00:00 2001 From: Logan Johnson Date: Thu, 10 Sep 2026 23:47:09 -0400 Subject: [PATCH] fix(mobile): confirm ambiguous presence through fresh snapshots Signed-off-by: Logan Johnson --- .../channels/channel_detail_page/app_bar.dart | 39 +- .../channels/channels_page/channel_tile.dart | 33 +- .../profile/presence_cache_provider.dart | 238 ++++++++--- .../features/profile/user_profile_sheet.dart | 12 +- .../channels/channel_detail_page_test.dart | 109 ++++- .../features/channels/channels_page_test.dart | 62 +++ .../profile/presence_cache_provider_test.dart | 17 +- .../profile/presence_snapshot_test.dart | 389 ++++++++++++++++++ 8 files changed, 802 insertions(+), 97 deletions(-) create mode 100644 mobile/test/features/profile/presence_snapshot_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 2914184729e..badf85338d3 100644 --- a/mobile/lib/features/channels/channel_detail_page/app_bar.dart +++ b/mobile/lib/features/channels/channel_detail_page/app_bar.dart @@ -217,9 +217,7 @@ class _DmAppBarTitle extends ConsumerWidget { ); final presence = ref.watch( presenceCacheProvider.select( - (presenceMap) => otherPubkey == null - ? 'offline' - : (presenceMap[otherPubkey] ?? 'offline'), + (presenceMap) => otherPubkey == null ? null : presenceMap[otherPubkey], ), ); @@ -249,7 +247,8 @@ class _DmAppBarTitle extends ConsumerWidget { final presenceLabel = switch (presence) { 'online' => 'Online', 'away' => 'Away', - _ => 'Offline', + 'offline' => 'Offline', + _ => 'Unknown', }; return Row( @@ -278,22 +277,24 @@ class _DmAppBarTitle extends ConsumerWidget { ), ), ), - badge: Center( - child: FractionallySizedBox( - widthFactor: _dmPresenceDotRatio, - heightFactor: _dmPresenceDotRatio, - child: DecoratedBox( - decoration: BoxDecoration( - color: switch (presence) { - 'online' => context.appColors.success, - 'away' => context.appColors.warning, - _ => context.colors.outline, - }, - shape: BoxShape.circle, + badge: presence == null + ? null + : Center( + child: FractionallySizedBox( + widthFactor: _dmPresenceDotRatio, + heightFactor: _dmPresenceDotRatio, + child: DecoratedBox( + decoration: BoxDecoration( + color: switch (presence) { + 'online' => context.appColors.success, + 'away' => context.appColors.warning, + _ => context.colors.outline, + }, + shape: BoxShape.circle, + ), + ), + ), ), - ), - ), - ), ), const SizedBox(width: Grid.xxs), Expanded( diff --git a/mobile/lib/features/channels/channels_page/channel_tile.dart b/mobile/lib/features/channels/channels_page/channel_tile.dart index e060e9e9ac4..c6235b18879 100644 --- a/mobile/lib/features/channels/channels_page/channel_tile.dart +++ b/mobile/lib/features/channels/channels_page/channel_tile.dart @@ -176,9 +176,7 @@ class _DmAvatar extends ConsumerWidget { ); final presence = ref.watch( presenceCacheProvider.select( - (presenceMap) => otherPubkey == null - ? 'offline' - : (presenceMap[otherPubkey] ?? 'offline'), + (presenceMap) => otherPubkey == null ? null : presenceMap[otherPubkey], ), ); @@ -219,28 +217,29 @@ class _DmAvatar extends ConsumerWidget { ), isAgent: profile?.isAgent == true, ), - Positioned( - right: -1, - bottom: -1, - child: Container( - width: 8, - height: 8, - decoration: BoxDecoration( - color: _presenceColor(context, presence), - shape: BoxShape.circle, - border: Border.all( - color: context.theme.scaffoldBackgroundColor, - width: 1.5, + if (presence != null) + Positioned( + right: -1, + bottom: -1, + child: Container( + width: 8, + height: 8, + decoration: BoxDecoration( + color: _presenceColor(context, presence), + shape: BoxShape.circle, + border: Border.all( + color: context.theme.scaffoldBackgroundColor, + width: 1.5, + ), ), ), ), - ), ], ), ); } - Color _presenceColor(BuildContext context, String presence) { + Color _presenceColor(BuildContext context, String? presence) { return switch (presence) { 'online' => context.appColors.success, 'away' => context.appColors.warning, diff --git a/mobile/lib/features/profile/presence_cache_provider.dart b/mobile/lib/features/profile/presence_cache_provider.dart index f735c376c35..eaf97b78a39 100644 --- a/mobile/lib/features/profile/presence_cache_provider.dart +++ b/mobile/lib/features/profile/presence_cache_provider.dart @@ -1,87 +1,233 @@ import 'dart:async'; import 'package:flutter/foundation.dart'; +import 'package:flutter/widgets.dart'; import 'package:hooks_riverpod/hooks_riverpod.dart'; 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. +/// Community-scoped presence: absent means unknown. Polls recover expired leases. +/// Live timestamps order live events only. At/below the relay snapshot second, +/// persistence-before-fanout makes a fresh snapshot confirmation necessary. class PresenceCacheNotifier extends Notifier> { + /// Bound deduplication memory; redelivery must not re-arm confirmation. + static const _seenIdLimit = 256; + final Set _tracked = {}; + final Set _pending = {}; + final Map _revisions = {}; + final Map _timestamps = {}; + final Map _snapshotAt = {}; + final List _seenIds = []; void Function()? _presenceUnsub; - int _subscriptionVersion = 0; + Timer? _poll; + Timer? _queued; + int _generation = 0; + int _epoch = 0; + bool _querying = false; + bool _opening = false; + bool _retrying = false; + (String, String?)? _scope; @override Map build() { final sessionState = ref.watch(relaySessionProvider); - + final lifecycle = ref.watch(appLifecycleProvider); + final scope = ref.watch( + relayConfigProvider.select((config) => (config.baseUrl, config.nsec)), + ); + if (_scope != scope) _tracked.clear(); + _scope = scope; + final generation = ++_generation; + _epoch++; + _querying = _opening = _retrying = false; + _revisions.clear(); + _timestamps.clear(); + _snapshotAt.clear(); + _seenIds.clear(); + _pending + ..clear() + ..addAll(_tracked); ref.onDispose(() { + _generation++; _presenceUnsub?.call(); _presenceUnsub = null; + _poll?.cancel(); + _queued?.cancel(); + _queued = null; }); - - if (sessionState.status == SessionStatus.connected) { - _subscribePresenceUpdates(); + if (sessionState.status == SessionStatus.connected && + lifecycle == AppLifecycleState.resumed) { + _poll = Timer.periodic(const Duration(seconds: 60), (_) { + if (_presenceUnsub == null) { + unawaited(_subscribe(generation)); + } else if (!_querying) { + _pending.addAll(_tracked); + _schedule(generation); + } + }); + // A transport may report ready synchronously; initialize state first. + Future.microtask(() => _subscribe(generation)); } - 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. + /// Request initial presence for exact keys. Repeated render calls coalesce. void track(List pubkeys) { - final normalized = pubkeys.map((pk) => pk.toLowerCase()).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". + for (final key in pubkeys.map((key) => key.trim().toLowerCase())) { + if (key.isNotEmpty && _tracked.add(key)) _pending.add(key); + } + _schedule(_generation); } - /// Subscribe to kind:20001 presence events over WebSocket. - Future _subscribePresenceUpdates() async { - _presenceUnsub?.call(); - _presenceUnsub = null; - _subscriptionVersion++; - final version = _subscriptionVersion; + void _schedule(int generation) { + _queued ??= Timer(Duration.zero, () { + _queued = null; + if (generation == _generation && _presenceUnsub != null && !_retrying) { + unawaited(_refresh(generation)); + } + }); + } - final session = ref.read(relaySessionProvider.notifier); + void _invalidateSnapshot() { + _epoch++; + _querying = false; + _pending.addAll(_tracked); + if (state.isNotEmpty) state = {}; + } + + Future _subscribe(int generation) async { + if (_opening || generation != _generation) return; + _opening = true; + _retrying = false; + var closed = false; try { - final unsub = await session.subscribe( - const NostrFilter(kinds: [EventKind.presenceUpdate], limit: 0), - _handlePresenceEvent, - ); - // Guard: if build() re-fired while we were awaiting, discard this - // subscription to avoid leaking it. - if (version != _subscriptionVersion) { + final unsub = await ref + .read(relaySessionProvider.notifier) + .subscribeWithStatus( + const NostrFilter(kinds: [EventKind.presenceUpdate], limit: 0), + (event) { + if (generation == _generation && !closed && !_retrying) { + _handlePresenceEvent(event); + } + }, + onStatusChanged: (status) { + if (generation != _generation || closed) return; + _retrying = status == RelaySubscriptionStatus.retrying; + _invalidateSnapshot(); + if (!_retrying) _schedule(generation); + }, + onClosed: (_) { + if (generation != _generation || closed) return; + closed = true; + _presenceUnsub?.call(); + _presenceUnsub = null; + _invalidateSnapshot(); + }, + ); + if (generation != _generation || closed) { unsub(); return; } _presenceUnsub = unsub; + _schedule(generation); } catch (error) { - debugPrint( - '[PresenceCacheNotifier] presence subscription failed: $error', - ); + if (generation == _generation) { + _invalidateSnapshot(); + debugPrint('[PresenceCacheNotifier] subscription failed: $error'); + } + } finally { + if (generation == _generation) _opening = false; } } void _handlePresenceEvent(NostrEvent event) { final pubkey = event.pubkey.toLowerCase(); - if (!_tracked.contains(pubkey)) return; - final status = event.content; - if (status != 'online' && status != 'away' && status != 'offline') return; - if (state[pubkey] == status) return; - final updated = Map.from(state); - updated[pubkey] = status; - state = updated; + final status = event.content.trim(); + if (event.kind != EventKind.presenceUpdate || + !_tracked.contains(pubkey) || + !_validStatus(status) || + event.createdAt < (_timestamps[pubkey] ?? 0) || + _seenRecently(event.id)) { + return; + } + // Accepted evidence fences in-flight queries, even if it needs confirmation. + _revisions[pubkey] = (_revisions[pubkey] ?? 0) + 1; + // Equality cannot order events within the relay's observation second. + if (event.createdAt <= (_snapshotAt[pubkey] ?? 0)) { + _pending.add(pubkey); + _schedule(_generation); + return; + } + _timestamps[pubkey] = event.createdAt; + if (state[pubkey] != status) state = {...state, pubkey: status}; + } + + Future _refresh(int generation) async { + if (_querying || _pending.isEmpty) return; + _querying = true; + final epoch = _epoch; + final session = ref.read(relaySessionProvider.notifier); + while (_pending.isNotEmpty) { + final keys = _pending.take(100).toList(); + _pending.removeAll(keys); + final revisions = {for (final key in keys) key: _revisions[key] ?? 0}; + List? events; + try { + events = await session.queryRelay([ + NostrFilter( + kinds: [EventKind.presenceUpdate], + authors: keys, + limit: keys.length, + ), + ]); + } catch (error) { + debugPrint('[PresenceCacheNotifier] snapshot failed: $error'); + } + if (generation != _generation || epoch != _epoch) return; + final latest = {}; + var observedAt = 0; + for (final event in events ?? []) { + final key = (event.getTagValue('p') ?? event.pubkey).toLowerCase(); + if (!revisions.containsKey(key) || + event.kind != EventKind.presenceUpdate || + !_validStatus(event.content.trim())) { + continue; + } + if (!latest.containsKey(key) || + event.createdAt > latest[key]!.createdAt) { + latest[key] = event; + } + if (event.createdAt > observedAt) observedAt = event.createdAt; + } + final updated = {...state}; + for (final key in keys) { + if ((_revisions[key] ?? 0) != revisions[key]) continue; + if (events == null) { + updated.remove(key); // failure is unknown; next poll retries + } else { + // Shared synthesis clock fences the batch; empty answers retain it. + if (observedAt > (_snapshotAt[key] ?? 0)) { + _snapshotAt[key] = observedAt; + } + updated[key] = latest[key]?.content.trim() ?? 'offline'; + } + } + if (!mapEquals(state, updated)) state = updated; + } + _querying = false; } + + bool _seenRecently(String id) { + if (id.isEmpty) return false; + if (_seenIds.contains(id)) return true; + _seenIds.add(id); + if (_seenIds.length > _seenIdLimit) _seenIds.removeAt(0); + return false; + } + + bool _validStatus(String status) => + status == 'online' || status == 'away' || status == 'offline'; } final presenceCacheProvider = diff --git a/mobile/lib/features/profile/user_profile_sheet.dart b/mobile/lib/features/profile/user_profile_sheet.dart index 97af99fa00a..607f9878e28 100644 --- a/mobile/lib/features/profile/user_profile_sheet.dart +++ b/mobile/lib/features/profile/user_profile_sheet.dart @@ -54,7 +54,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]; @@ -338,13 +338,13 @@ void _showProfileCopyToast(BuildContext context) { class _ProfilePresenceChip extends StatelessWidget { const _ProfilePresenceChip({required this.presence}); - final String presence; + final String? presence; @override Widget build(BuildContext context) { final effectivePresence = switch (presence) { - 'online' || 'away' => presence, - _ => 'offline', + 'online' || 'away' || 'offline' => presence, + _ => null, }; final backgroundColor = switch (effectivePresence) { 'online' => context.appColors.success, @@ -354,11 +354,13 @@ class _ProfilePresenceChip extends StatelessWidget { final label = switch (effectivePresence) { 'online' => 'Online', 'away' => 'Away', - _ => 'Offline', + 'offline' => 'Offline', + _ => 'Unknown', }; return Semantics( label: 'Presence: $label', + excludeSemantics: true, child: SizedBox( height: Grid.xl, child: Center( diff --git a/mobile/test/features/channels/channel_detail_page_test.dart b/mobile/test/features/channels/channel_detail_page_test.dart index 23f547dbad0..32e287e2a4e 100644 --- a/mobile/test/features/channels/channel_detail_page_test.dart +++ b/mobile/test/features/channels/channel_detail_page_test.dart @@ -1,4 +1,7 @@ import 'dart:async'; + +import '../profile/presence_snapshot_test.dart' + show PresenceTestRelay, presenceEvent; import 'dart:collection'; import 'dart:convert'; @@ -518,6 +521,101 @@ void main() { }, ); + for (final profile in [false, true]) { + testWidgets('presence observation failure is unknown: profile=$profile', ( + tester, + ) async { + final relay = PresenceTestRelay(); + final semantics = tester.ensureSemantics(); + final dm = Channel( + id: _channelId, + name: 'DM', + channelType: 'dm', + visibility: 'private', + description: '', + createdBy: 'self', + createdAt: DateTime(2025), + memberCount: 2, + participants: const ['Self', 'Alice'], + participantPubkeys: const ['self', 'alice'], + isMember: true, + ); + await tester.pumpWidget( + _buildTestable( + messages: const [], + channel: dm, + relaySessionNotifier: relay, + home: profile ? const UserProfileSheet(pubkey: 'alice') : null, + users: const { + 'alice': UserProfile(pubkey: 'alice', displayName: 'Alice'), + }, + ), + ); + await tester.pumpAndSettle(); + void check(String label) { + if (profile) { + expect(find.text(label), findsOneWidget); + expect(find.bySemanticsLabel('Presence: $label'), findsOneWidget); + expect(find.bySemanticsLabel(label), findsNothing); + } else { + expect( + tester + .widget( + find.byKey(const ValueKey('dm-header-presence')), + ) + .data, + label, + ); + expect( + tester + .widget( + find.byKey(const ValueKey('dm-header-avatar')), + ) + .badge, + label == 'Unknown' ? isNull : isNotNull, + ); + } + if (label != 'Offline') expect(find.text('Offline'), findsNothing); + } + + check('Unknown'); + expect(relay.queries.single.authors, ['alice']); + relay.results.removeAt(0).complete([ + presenceEvent('relay', 'online', subject: 'alice', timestamp: 20), + ]); + await tester.pumpAndSettle(); + check('Online'); + await tester.pump(const Duration(seconds: 60)); + final stale = relay.results.removeAt(0); + relay.emit(presenceEvent('alice', 'offline', timestamp: 10)); + await tester.pump(); + check('Online'); + stale.complete([]); + await tester.pumpAndSettle(); + check('Online'); + relay.results.removeAt(0).complete([ + presenceEvent('relay', 'away', subject: 'alice', timestamp: 20), + ]); + await tester.pumpAndSettle(); + check('Away'); + await tester.pump(const Duration(seconds: 60)); + relay.results.removeAt(0).completeError(Exception('unavailable')); + await tester.pumpAndSettle(); + check('Unknown'); + await tester.pump(const Duration(seconds: 60)); + relay.results.removeAt(0).complete([]); + await tester.pumpAndSettle(); + check('Offline'); + relay.emit(presenceEvent('alice', 'online', timestamp: 21)); + await tester.pumpAndSettle(); + check('Online'); + relay.emit(presenceEvent('alice', 'offline', timestamp: 22)); + await tester.pumpAndSettle(); + check('Offline'); + semantics.dispose(); + }); + } + testWidgets('uses the shared 32px masked presence avatar in DM headers', ( tester, ) async { @@ -538,6 +636,7 @@ void main() { await tester.pumpWidget( _buildTestable( messages: const [], + relaySessionNotifier: PresenceTestRelay()..emptySnapshots = true, channel: dmChannel, users: const { 'alice': UserProfile(pubkey: 'alice', displayName: 'Alice'), @@ -677,6 +776,7 @@ void main() { await tester.pumpWidget( _buildTestable( messages: const [], + relaySessionNotifier: PresenceTestRelay()..emptySnapshots = true, channel: dmChannel, loadChannelBotPubkeys: () async => const {'bot'}, ), @@ -14453,6 +14553,10 @@ class _IdentityUpdateRelaySession extends RelaySessionNotifier { } }; } + if (!filter.kinds.contains(39002)) { + onStatusChanged(RelaySubscriptionStatus.ready); + return () {}; + } _membershipListener = onEvent; _membershipStatusListener = onStatusChanged; onStatusChanged(RelaySubscriptionStatus.ready); @@ -14655,8 +14759,9 @@ class _SynchronousReadStateNotifier extends ReadStateNotifier { class _FakeProfileNotifier extends ProfileNotifier { @override - Future build() async => - const UserProfile(pubkey: 'self', displayName: 'Self'); + // Fixed fixture identity must be available before DM presence is tracked. + Future build() => + SynchronousFuture(const UserProfile(pubkey: 'self', displayName: 'Self')); } class _FakeChannelStarsNotifier extends ChannelStarsNotifier { diff --git a/mobile/test/features/channels/channels_page_test.dart b/mobile/test/features/channels/channels_page_test.dart index 6293ae373e9..dcf8dbc8631 100644 --- a/mobile/test/features/channels/channels_page_test.dart +++ b/mobile/test/features/channels/channels_page_test.dart @@ -1,4 +1,7 @@ import 'dart:async'; + +import '../profile/presence_snapshot_test.dart' + show PresenceTestRelay, presenceEvent; import 'dart:math'; import 'package:flutter/foundation.dart'; @@ -134,6 +137,57 @@ void main() { ), ]; + testWidgets('DM tile presence observation failure hides the dot', ( + tester, + ) async { + final relay = PresenceTestRelay(); + await tester.pumpWidget( + buildTestable( + overrides: [ + channelsProvider.overrideWith(() => _FakeNotifier(testChannels)), + relaySessionProvider.overrideWith(() => relay), + ], + ), + ); + await tester.pumpAndSettle(); + final dot = find.descendant( + of: _dmTileFor('Alice'), + matching: find.byWidgetPredicate( + (widget) => + widget is Container && + widget.constraints?.maxWidth == 8 && + widget.constraints?.maxHeight == 8, + ), + ); + expect(dot, findsNothing); + relay.results.removeAt(0).complete([ + presenceEvent('relay', 'online', subject: 'alice', timestamp: 20), + ]); + await tester.pumpAndSettle(); + expect(dot, findsOneWidget); + final onlineColor = + (tester.widget(dot).decoration as BoxDecoration).color; + await tester.pump(const Duration(seconds: 60)); + relay.results.removeAt(0).completeError(Exception('unavailable')); + await tester.pumpAndSettle(); + expect(dot, findsNothing); + await tester.pump(const Duration(seconds: 60)); + relay.results.removeAt(0).complete([]); + await tester.pumpAndSettle(); + expect(dot, findsOneWidget); + expect( + (tester.widget(dot).decoration as BoxDecoration).color, + isNot(onlineColor), + ); + relay.emit(presenceEvent('alice', 'online', timestamp: 21)); + await tester.pumpAndSettle(); + expect( + (tester.widget(dot).decoration as BoxDecoration).color, + onlineColor, + ); + await tester.pumpWidget(const SizedBox()); + }); + testWidgets('shows grouped channel list when data loads', (tester) async { // Valid fixture keys whose npub encodings were verified against the // NIP-19 codec independently of the code under test. @@ -2655,6 +2709,14 @@ class _ReconnectingRelaySession extends RelaySessionNotifier { Duration timeout = const Duration(seconds: 8), }) async => []; + @override + Future subscribeWithStatus( + NostrFilter filter, + void Function(NostrEvent) onEvent, { + void Function(String message)? onClosed, + required void Function(RelaySubscriptionStatus) onStatusChanged, + }) => subscribe(filter, onEvent, onClosed: onClosed); + @override Future subscribe( NostrFilter filter, diff --git a/mobile/test/features/profile/presence_cache_provider_test.dart b/mobile/test/features/profile/presence_cache_provider_test.dart index 9fc02d9575b..219afd8f3c4 100644 --- a/mobile/test/features/profile/presence_cache_provider_test.dart +++ b/mobile/test/features/profile/presence_cache_provider_test.dart @@ -4,13 +4,7 @@ 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. +/// Live-stream behavior alongside the authenticated snapshot backstop. void main() { test('WS presence event updates cache for tracked pubkey', () async { final relaySession = _RecordingRelaySessionNotifier(); @@ -167,10 +161,11 @@ class _RecordingRelaySessionNotifier extends RelaySessionNotifier { SessionState build() => const SessionState(status: SessionStatus.connected); @override - Future subscribe( + Future subscribeWithStatus( NostrFilter filter, void Function(NostrEvent) onEvent, { void Function(String message)? onClosed, + required void Function(RelaySubscriptionStatus) onStatusChanged, }) async { filters.add(filter); _listeners.add(onEvent); @@ -180,6 +175,12 @@ class _RecordingRelaySessionNotifier extends RelaySessionNotifier { }; } + @override + Future> queryRelay( + List filters, { + Duration timeout = const Duration(seconds: 8), + }) async => []; + /// Emit an event synchronously to all live subscribers. void emit(NostrEvent event) { for (final listener in List.of(_listeners)) { diff --git a/mobile/test/features/profile/presence_snapshot_test.dart b/mobile/test/features/profile/presence_snapshot_test.dart new file mode 100644 index 00000000000..5ba40ef3ef1 --- /dev/null +++ b/mobile/test/features/profile/presence_snapshot_test.dart @@ -0,0 +1,389 @@ +import 'dart:async'; + +import 'package:buzz/features/profile/presence_cache_provider.dart'; +import 'package:buzz/shared/relay/relay.dart'; +import 'package:fake_async/fake_async.dart'; +import 'package:flutter/widgets.dart'; +import 'package:flutter_test/flutter_test.dart'; +import 'package:hooks_riverpod/hooks_riverpod.dart'; + +void main() { + void scenario( + void Function(FakeAsync, ProviderContainer, PresenceTestRelay) run, + ) { + fakeAsync((time) { + final relay = PresenceTestRelay(); + final container = ProviderContainer( + overrides: [ + relaySessionProvider.overrideWith(() => relay), + appLifecycleProvider.overrideWith(_Lifecycle.new), + ], + ); + container.read(presenceCacheProvider.notifier).track([' ALICE ', 'bob']); + time.elapse(Duration.zero); + try { + run(time, container, relay); + } finally { + container.dispose(); + time.flushMicrotasks(); + } + }); + } + + test('snapshot covers exact subjects; latest p-tag record wins', () { + scenario((time, container, relay) { + expect(relay.queries.single.authors, ['alice', 'bob']); + expect(relay.queries.single.limit, 2); + expect(relay.queries.single.kinds, [EventKind.presenceUpdate]); + relay.results.removeAt(0).complete([ + presenceEvent('relay', 'away', subject: 'alice', timestamp: 20), + presenceEvent('relay', ' online ', subject: 'alice'), + presenceEvent('stranger', 'online'), + ]); + time.flushMicrotasks(); + expect(container.read(presenceCacheProvider), { + 'alice': 'away', + 'bob': 'offline', + }); + container.read(presenceCacheProvider.notifier).track(['alice']); + time.elapse(Duration.zero); + expect(relay.queries, hasLength(1)); + }); + }); + + test('unchanged live heartbeat fences snapshot and older live events', () { + scenario((time, container, relay) { + relay.emit(presenceEvent('alice', 'online')); + relay.results.removeAt(0).complete([]); + time.flushMicrotasks(); + time.elapse(const Duration(seconds: 60)); + relay.emit(presenceEvent('alice', 'online', timestamp: 20)); + relay.emit(presenceEvent('alice', 'offline', timestamp: 1)); + relay.results.removeAt(0).complete([]); + time.flushMicrotasks(); + expect(container.read(presenceCacheProvider)['alice'], 'online'); + // No new heartbeat: a successful empty refresh observes relay TTL expiry. + time.elapse(const Duration(seconds: 60)); + relay.results.removeAt(0).complete([]); + time.flushMicrotasks(); + expect(container.read(presenceCacheProvider)['alice'], 'offline'); + }); + }); + + test('failed initial and refresh reads are unknown and retry on poll', () { + scenario((time, container, relay) { + relay.results.removeAt(0).completeError(Exception('unavailable')); + time.flushMicrotasks(); + expect(container.read(presenceCacheProvider), isEmpty); + time.elapse(const Duration(seconds: 60)); + relay.results.removeAt(0).complete([presenceEvent('alice', 'online')]); + time.flushMicrotasks(); + expect(container.read(presenceCacheProvider)['alice'], 'online'); + time.elapse(const Duration(seconds: 60)); + relay.results.removeAt(0).completeError(Exception('unavailable')); + time.flushMicrotasks(); + expect(container.read(presenceCacheProvider), isEmpty); + }); + }); + + for (final timestamp in [10, 20]) { + for (final settled in ['online', 'offline']) { + final transition = settled == 'online' ? 'offline' : 'online'; + test('ambiguous $timestamp/$settled fences and freshly confirms', () { + scenario((time, container, relay) { + final online = presenceEvent( + 'relay', + 'online', + subject: 'alice', + timestamp: 20, + ); + relay.results.removeAt(0).complete([online]); + time.flushMicrotasks(); + if (settled == 'offline') { + time.elapse(const Duration(seconds: 60)); + relay.results.removeAt(0).complete([]); + time.flushMicrotasks(); + } + final snapshot = settled == 'online' ? [online] : []; + expect(container.read(presenceCacheProvider)['alice'], settled); + time.elapse(const Duration(seconds: 60)); + final stale = relay.results.removeAt(0); + final delayed = presenceEvent( + 'alice', + transition, + timestamp: timestamp, + ); + relay.emit(delayed); + expect(container.read(presenceCacheProvider)['alice'], settled); + stale.complete([]); + time.flushMicrotasks(); + expect(container.read(presenceCacheProvider)['alice'], settled); + relay.results.removeAt(0).complete(snapshot); + time.flushMicrotasks(); + expect(container.read(presenceCacheProvider)['alice'], settled); + relay.emit(delayed); + time.elapse(Duration.zero); + expect(relay.queries, hasLength(settled == 'online' ? 3 : 4)); + relay.emit( + presenceEvent( + 'alice', + transition, + timestamp: timestamp, + subject: 'new', + ), + ); + time.elapse(Duration.zero); + relay.results + .removeAt(0) + .complete(transition == 'online' ? [online] : []); + time.flushMicrotasks(); + expect(container.read(presenceCacheProvider)['alice'], transition); + relay.emit(presenceEvent('alice', settled, timestamp: 21)); + expect(container.read(presenceCacheProvider)['alice'], settled); + }); + }); + } + } + + test('post-ready retry invalidates in-flight results and resnapshots', () { + scenario((time, container, relay) { + final stale = relay.results.removeAt(0); + relay.status(RelaySubscriptionStatus.retrying); + stale.complete([presenceEvent('alice', 'online')]); + time.flushMicrotasks(); + expect(container.read(presenceCacheProvider), isEmpty); + relay.status(RelaySubscriptionStatus.ready); + time.elapse(Duration.zero); + expect(relay.queries, hasLength(2)); + relay.results.removeAt(0).complete([presenceEvent('alice', 'away')]); + time.flushMicrotasks(); + expect(container.read(presenceCacheProvider)['alice'], 'away'); + }); + }); + + test('terminal close fences late events and recovers at bounded cadence', () { + scenario((time, container, relay) { + final emitOld = relay.emit; + final stale = relay.results.removeAt(0); + relay.close('restricted'); + emitOld(presenceEvent('alice', 'online')); + stale.complete([presenceEvent('alice', 'online')]); + time.flushMicrotasks(); + expect(container.read(presenceCacheProvider), isEmpty); + time.elapse(const Duration(seconds: 59)); + expect(relay.subscriptions, 1); + time.elapse(const Duration(seconds: 1)); + expect(relay.subscriptions, 2); + relay.results.removeAt(0).complete([]); + time.flushMicrotasks(); + relay.emit(presenceEvent('alice', 'away')); + expect(container.read(presenceCacheProvider)['alice'], 'away'); + }); + }); + + test('disconnect and disposal fence pending queries and unsubscribe', () { + scenario((time, container, relay) { + final stale = relay.results.removeAt(0); + relay.state = const SessionState(status: SessionStatus.disconnected); + expect(container.read(presenceCacheProvider), isEmpty); + stale.complete([presenceEvent('alice', 'online')]); + time.flushMicrotasks(); + expect(container.read(presenceCacheProvider), isEmpty); + relay.state = const SessionState(status: SessionStatus.connected); + container.read(presenceCacheProvider); + time.elapse(Duration.zero); + final disposed = relay.results.removeAt(0); + container.dispose(); + disposed.complete([presenceEvent('alice', 'online')]); + time.flushMicrotasks(); + expect(relay.unsubscribes, 2); + }); + }); + + test('community/account switch clears tracking and fences old snapshot', () { + scenario((time, container, relay) { + final stale = relay.results.removeAt(0); + container + .read(relayConfigProvider.notifier) + .update(baseUrl: 'https://other.example', nsec: 'different-account'); + expect(container.read(presenceCacheProvider), isEmpty); + time.elapse(Duration.zero); + stale.complete([presenceEvent('alice', 'online')]); + time.flushMicrotasks(); + expect(container.read(presenceCacheProvider), isEmpty); + expect(relay.queries, hasLength(1)); + container.read(presenceCacheProvider.notifier).track(['carol']); + time.elapse(Duration.zero); + expect(relay.queries.last.authors, ['carol']); + }); + }); + + test('background stops polling; foreground obtains a fresh snapshot', () { + scenario((time, container, relay) { + relay.results.removeAt(0).complete([presenceEvent('alice', 'online')]); + time.flushMicrotasks(); + container.read(appLifecycleProvider.notifier).state = + AppLifecycleState.paused; + expect(container.read(presenceCacheProvider), isEmpty); + time.elapse(const Duration(minutes: 2)); + expect(relay.queries, hasLength(1)); + container.read(appLifecycleProvider.notifier).state = + AppLifecycleState.resumed; + container.read(presenceCacheProvider); + time.elapse(Duration.zero); + expect(relay.queries, hasLength(2)); + }); + }); + + test( + 'batches beyond default limit without declaring unqueried keys offline', + () { + scenario((time, container, relay) { + container.read(presenceCacheProvider.notifier).track([ + for (var i = 0; i < 205; i++) 'agent-$i', + ]); + relay.results.removeAt(0).complete([]); + time.flushMicrotasks(); + expect(relay.queries.last.authors, hasLength(100)); + expect(container.read(presenceCacheProvider)['agent-204'], isNull); + for (var i = 0; i < 3; i++) { + relay.results.removeAt(0).complete([]); + time.flushMicrotasks(); + } + expect(relay.queries.map((filter) => filter.limit), [2, 100, 100, 5]); + expect(container.read(presenceCacheProvider), hasLength(207)); + }); + }, + ); + + test('initial subscription failure retries without snapshotting a gap', () { + fakeAsync((time) { + final relay = PresenceTestRelay()..failSubscribe = true; + final container = ProviderContainer( + overrides: [ + relaySessionProvider.overrideWith(() => relay), + appLifecycleProvider.overrideWith(_Lifecycle.new), + ], + ); + container.read(presenceCacheProvider.notifier).track(['alice']); + time.elapse(Duration.zero); + expect(relay.queries, isEmpty); + expect(container.read(presenceCacheProvider), isEmpty); + relay.failSubscribe = false; + time.elapse(const Duration(seconds: 60)); + expect(relay.subscriptions, 2); + expect(relay.queries.single.authors, ['alice']); + container.dispose(); + }); + }); + + test( + 'subscribe first; dispose before ready closes the late subscription', + () { + fakeAsync((time) { + final relay = PresenceTestRelay()..ready = Completer(); + final container = ProviderContainer( + overrides: [ + relaySessionProvider.overrideWith(() => relay), + appLifecycleProvider.overrideWith(_Lifecycle.new), + ], + ); + container.read(presenceCacheProvider.notifier).track(['alice']); + time.elapse(Duration.zero); + expect(relay.queries, isEmpty); + container.dispose(); + relay.ready!.complete(); + time.flushMicrotasks(); + expect(relay.unsubscribes, 1); + expect(relay.queries, isEmpty); + }); + }, + ); +} + +NostrEvent presenceEvent( + String author, + String status, { + String? subject, + int timestamp = 10, +}) => NostrEvent( + id: '$author-$status-$timestamp-$subject', + pubkey: author, + createdAt: timestamp, + kind: EventKind.presenceUpdate, + tags: [ + if (subject != null) ['p', subject], + ], + content: status, + sig: 'sig', +); + +class _Lifecycle extends AppLifecycleNotifier { + @override + AppLifecycleState build() => AppLifecycleState.resumed; +} + +class PresenceTestRelay extends RelaySessionNotifier { + final queries = []; + final results = >>[]; + late void Function(NostrEvent) emit; + late void Function(RelaySubscriptionStatus) status; + late void Function(String) close; + Completer? ready; + bool failSubscribe = false; + bool emptySnapshots = false; + int subscriptions = 0; + int unsubscribes = 0; + + @override + SessionState build() => const SessionState(status: SessionStatus.connected); + + @override + Future subscribeWithStatus( + NostrFilter filter, + void Function(NostrEvent) onEvent, { + void Function(String)? onClosed, + required void Function(RelaySubscriptionStatus) onStatusChanged, + }) async { + if (!filter.kinds.contains(EventKind.presenceUpdate)) { + onStatusChanged(RelaySubscriptionStatus.ready); + return () {}; + } + subscriptions++; + if (failSubscribe) throw StateError('unavailable'); + emit = onEvent; + status = onStatusChanged; + close = onClosed!; + onStatusChanged(RelaySubscriptionStatus.ready); + if (ready != null) await ready!.future; + return () => unsubscribes++; + } + + @override + Future> fetchHistory( + NostrFilter filter, { + Duration timeout = const Duration(seconds: 8), + }) async => []; + + @override + Future subscribe( + NostrFilter filter, + void Function(NostrEvent) onEvent, { + void Function(String)? onClosed, + }) async => () {}; + + @override + Future> queryRelay( + List filters, { + Duration timeout = const Duration(seconds: 8), + }) { + if (!filters.single.kinds.contains(EventKind.presenceUpdate)) { + return Future.value([]); + } + queries.addAll(filters); + if (emptySnapshots) return Future.value([]); + final result = Completer>(); + results.add(result); + return result.future; + } +}