Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
loganj wants to merge 1 commit into
base: main
Choose a base branch
from
Open

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
39 changes: 20 additions & 19 deletions mobile/lib/features/channels/channel_detail_page/app_bar.dart
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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],
),
);

Expand Down Expand Up @@ -249,7 +247,8 @@ class _DmAppBarTitle extends ConsumerWidget {
final presenceLabel = switch (presence) {
'online' => 'Online',
'away' => 'Away',
_ => 'Offline',
'offline' => 'Offline',
_ => 'Unknown',
};

return Row(
Expand Down Expand Up @@ -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(
Expand Down
33 changes: 16 additions & 17 deletions mobile/lib/features/channels/channels_page/channel_tile.dart
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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],
),
);

Expand Down Expand Up @@ -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,
Expand Down
238 changes: 192 additions & 46 deletions mobile/lib/features/profile/presence_cache_provider.dart
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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<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 =
Expand Down
Loading
Loading