fix(mobile): hydrate and refresh relay presence snapshots (#7382)

<!-- Draft PR body for #7382 · fix(mobile): hydrate and refresh relay
presence snapshots · https://github.com/block/buzz/pull/7382 -->

🤖

## Summary

Mobile showed every person in a DM as **Offline** until they happened to
publish a presence update while the app was connected. Presence was
never loaded when the app opened, never refreshed, and a missing or
failed read looked identical to actually being offline. This PR makes
presence load and stay current: on connect or foreground, the app
fetches the relay's presence snapshot for everyone visible, keeps the
live subscription open while it does, and re-checks every 60 seconds
while the app is in the foreground. A missing or failed read shows
**Unknown** (no dot), not Offline.

- Snapshot reads are authenticated like the rest of the session and
batched by exact key (at most 100 per batch), instead of relying on live
events only.
- The live subscription opens before the snapshot, so updates published
during the read are not dropped. Per-key revisions fence stale **query
completions**: a snapshot that finishes after newer live evidence for
that key does not overwrite it.
- Refreshing every 60 foreground seconds observes relay lease expiry and
retry failures; missing or failed evidence simply stays absent — it is
never reported as Offline.
- Disconnects, backgrounding, and account/community changes reset the
cache cleanly. The app's existing subscription/recovery machinery is
reused rather than duplicated.

Mirrors Desktop's presence hydration
(`desktop/src-tauri/src/commands/profile.rs::get_presence`,
`desktop/src/features/presence/hooks.ts`).

### Presence contract: best effort between polls

Presence is a best-effort estimate that the next foreground poll
corrects. This PR does not give strict causal order between snapshots
and live events:

- Snapshots carry relay time. Live kind:20001 events carry the signer's
time. These clocks are not comparable.
- A live event at or below a snapshot's relay second is not applied
directly. The client confirms it with a fresh snapshot. This works
because the relay persists presence before fan-out.
- An empty (all-Offline) snapshot carries no relay second. If the relay
stored an Online event, then its fan-out was delayed past lease expiry
and an empty snapshot, the delayed event can show Online/Away until the
next 60-second poll. Base `main` has the same stale-delivery exposure
with no correction at all.

Strict ordering needs relay-issued ordering metadata on both paths, or a
fresh confirmation for every live event. Both are protocol or latency
tradeoffs and are out of scope here.

### Related issue

- Fixes: N/A. Closest open duplicates: #4417 (larger presence bootstrap
+ UI patch), #5617. Related: #6200, #7295.
- First slice of the mobile remote-agent parity port. This PR includes
the Unknown rendering at all three consumers (DM header, channel tile,
profile sheet). #7383 and #7526 stack on it and add test coverage only.

### Testing

- Rebased on `main` `95b018c87` (2026-09-30). At stack tip `7c8a93081`:
`just mobile-check` clean; full `just mobile-test` **2,593 passed**.
Earlier receipts below apply to the pre-rebase head.

- Regressions cover cold-open hydration, live/snapshot races, lease
expiry, failed reads and retries, disconnect/background/scope changes,
and >100-key batching.
- At the branch head: `just mobile-check` and the full mobile test suite
pass independently, and every broader CI gate has successful evidence
after isolated recovery runs (no monolithic local `just ci` run) —
receipts in the [exact-head evidence
comment](https://github.com/block/buzz/pull/7382#issuecomment-5573112882).
- Verification is automated-test level; no native device or simulator
run is claimed. No layout change in this slice.

### Screenshots

No layout change in this slice — the renders below show the same
existing profile sheet, with the presence status now following hydrated
relay data. Flutter production-widget test renders — not native-device
screenshots or acceptance captures.

| Scenario | Before | After |
|---|---|---|
| Existing profile opened cold, with a signed relay snapshot saying
Online | ![Before: the profile stays Offline — the snapshot is never
fetched](https://github.com/user-attachments/assets/4b4899c8-364c-4621-8d59-949189860eb3)
| ![After: the profile shows Online once the snapshot
hydrates](https://github.com/user-attachments/assets/7b1a9ecd-a7a8-4e04-b007-b86565a39cfc)
|

<details>
<summary>Capture provenance</summary>

Rendered by the Flutter widget engine in a `flutter test` run
(production widgets, production theme; no device or simulator). Before:
this PR's declared base `3c7f288c60d67df78577b237e27c3dfc8831aaa1`.
After: its head `a8e8028bfb9bb67de2db03ed448839e2f3ab2b0d`.

</details>

Signed-off-by: Logan Johnson <loganj@squareup.com>
This commit is contained in:
Logan Johnson
2026-09-30 20:10:24 -04:00
committed by GitHub
parent f001ebbb6e
commit 965fe1dffa
8 changed files with 807 additions and 96 deletions

View File

@@ -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(

View File

@@ -181,9 +181,7 @@ class _DmAvatar extends ConsumerWidget {
);
final presence = ref.watch(
presenceCacheProvider.select(
(presenceMap) => otherPubkey == null
? 'offline'
: (presenceMap[otherPubkey] ?? 'offline'),
(presenceMap) => otherPubkey == null ? null : presenceMap[otherPubkey],
),
);
@@ -224,28 +222,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,

View File

@@ -1,87 +1,239 @@
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.
/// Community-scoped presence: absent means unknown. Polls recover expired leases.
///
/// 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.
/// Presence is a best-effort estimate between foreground polls. Per-key
/// revisions fence stale query completions. Snapshot/live ordering is not
/// causal: snapshots carry relay time and live events carry signer time. A
/// live event at/below the relay snapshot second is confirmed by a fresh
/// snapshot (persistence precedes fan-out). An empty snapshot has no relay
/// second, so a delayed older live event can briefly supersede its Offline;
/// the next 60-second poll corrects it.
class PresenceCacheNotifier extends Notifier<Map<String, String>> {
/// Bound deduplication memory; redelivery must not re-arm confirmation.
static const _seenIdLimit = 256;
final Set<String> _tracked = {};
final Set<String> _pending = {};
final Map<String, int> _revisions = {};
final Map<String, int> _timestamps = {};
final Map<String, int> _snapshotAt = {};
final List<String> _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<String, String> 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<String> 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<void> _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<void> _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<String, String>.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<void> _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<NostrEvent>? 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 = <String, NostrEvent>{};
var observedAt = 0;
for (final event in events ?? <NostrEvent>[]) {
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 =

View File

@@ -70,7 +70,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];
@@ -356,13 +356,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,
@@ -372,11 +372,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(

View File

@@ -1,4 +1,7 @@
import 'dart:async';
import '../profile/presence_snapshot_test.dart'
show PresenceTestRelay, presenceEvent;
import 'dart:collection';
import 'dart:convert';
@@ -643,6 +646,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<Text>(
find.byKey(const ValueKey('dm-header-presence')),
)
.data,
label,
);
expect(
tester
.widget<MaskedAvatarBadge>(
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 {
@@ -663,6 +761,7 @@ void main() {
await tester.pumpWidget(
_buildTestable(
messages: const [],
relaySessionNotifier: PresenceTestRelay()..emptySnapshots = true,
channel: dmChannel,
users: const {
'alice': UserProfile(pubkey: 'alice', displayName: 'Alice'),
@@ -802,6 +901,7 @@ void main() {
await tester.pumpWidget(
_buildTestable(
messages: const [],
relaySessionNotifier: PresenceTestRelay()..emptySnapshots = true,
channel: dmChannel,
loadChannelBotPubkeys: () async => const {'bot'},
),
@@ -15123,6 +15223,10 @@ class _IdentityUpdateRelaySession extends RelaySessionNotifier {
}
};
}
if (!filter.kinds.contains(39002)) {
onStatusChanged(RelaySubscriptionStatus.ready);
return () {};
}
_membershipListener = onEvent;
_membershipStatusListener = onStatusChanged;
onStatusChanged(RelaySubscriptionStatus.ready);
@@ -15325,8 +15429,9 @@ class _SynchronousReadStateNotifier extends ReadStateNotifier {
class _FakeProfileNotifier extends ProfileNotifier {
@override
Future<UserProfile?> build() async =>
const UserProfile(pubkey: 'self', displayName: 'Self');
// Fixed fixture identity must be available before DM presence is tracked.
Future<UserProfile?> build() =>
SynchronousFuture(const UserProfile(pubkey: 'self', displayName: 'Self'));
}
class _FakeChannelStarsNotifier extends ChannelStarsNotifier {

View File

@@ -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<Container>(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<Container>(dot).decoration as BoxDecoration).color,
isNot(onlineColor),
);
relay.emit(presenceEvent('alice', 'online', timestamp: 21));
await tester.pumpAndSettle();
expect(
(tester.widget<Container>(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<void Function()> subscribeWithStatus(
NostrFilter filter,
void Function(NostrEvent) onEvent, {
void Function(String message)? onClosed,
required void Function(RelaySubscriptionStatus) onStatusChanged,
}) => subscribe(filter, onEvent, onClosed: onClosed);
@override
Future<void Function()> subscribe(
NostrFilter filter,

View File

@@ -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<void Function()> subscribe(
Future<void Function()> 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<List<NostrEvent>> queryRelay(
List<NostrFilter> 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)) {

View File

@@ -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] : <NostrEvent>[];
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<void>();
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 = <NostrFilter>[];
final results = <Completer<List<NostrEvent>>>[];
late void Function(NostrEvent) emit;
late void Function(RelaySubscriptionStatus) status;
late void Function(String) close;
Completer<void>? ready;
bool failSubscribe = false;
bool emptySnapshots = false;
int subscriptions = 0;
int unsubscribes = 0;
@override
SessionState build() => const SessionState(status: SessionStatus.connected);
@override
Future<void Function()> 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<List<NostrEvent>> fetchHistory(
NostrFilter filter, {
Duration timeout = const Duration(seconds: 8),
}) async => [];
@override
Future<void Function()> subscribe(
NostrFilter filter,
void Function(NostrEvent) onEvent, {
void Function(String)? onClosed,
}) async => () {};
@override
Future<List<NostrEvent>> queryRelay(
List<NostrFilter> 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<List<NostrEvent>>();
results.add(result);
return result.future;
}
}