diff --git a/mobile/lib/features/channels/channel_management_provider.dart b/mobile/lib/features/channels/channel_management_provider.dart index aceaec21f7..a88e69be91 100644 --- a/mobile/lib/features/channels/channel_management_provider.dart +++ b/mobile/lib/features/channels/channel_management_provider.dart @@ -114,6 +114,7 @@ String _channelMemberSnapshotKey({ class _ChannelMembersSnapshotCache { final _membersByKey = >{}; + final _latestReadByKey = {}; List? read({ required String relayBaseUrl, @@ -126,19 +127,35 @@ class _ChannelMembersSnapshotCache { channelId: channelId, )]; + /// Starts a roster read for this scope and returns its generation. A later + /// call supersedes it, so its [write] is then ignored. + int beginRead({ + required String relayBaseUrl, + required String? pubkey, + required String channelId, + }) { + final key = _channelMemberSnapshotKey( + relayBaseUrl: relayBaseUrl, + pubkey: pubkey, + channelId: channelId, + ); + return _latestReadByKey[key] = (_latestReadByKey[key] ?? 0) + 1; + } + void write({ required String relayBaseUrl, required String? pubkey, required String channelId, + required int generation, required List members, }) { - _membersByKey[_channelMemberSnapshotKey( + final key = _channelMemberSnapshotKey( relayBaseUrl: relayBaseUrl, pubkey: pubkey, channelId: channelId, - )] = List.unmodifiable( - members, ); + if (_latestReadByKey[key] != generation) return; + _membersByKey[key] = List.unmodifiable(members); } } @@ -384,6 +401,14 @@ final channelMembersProvider = FutureProvider.autoDispose final relayBaseUrl = ref.watch(relayConfigProvider).baseUrl; final pubkey = ref.watch(myPubkeyProvider)?.toLowerCase(); final snapshotCache = ref.read(_channelMembersSnapshotCacheProvider); + // A read superseded by a later one (a roster update arriving mid-read) + // must not overwrite the snapshot the newer read already wrote. A read + // torn down with nothing newer still writes: it has the latest roster. + final generation = snapshotCache.beginRead( + relayBaseUrl: relayBaseUrl, + pubkey: pubkey, + channelId: channelId, + ); // Re-fetch only after reconnect completes. During the disconnected // interval this provider has no session dependency, so its current value // remains visible to every consumer rather than becoming AsyncData([]). @@ -409,6 +434,7 @@ final channelMembersProvider = FutureProvider.autoDispose relayBaseUrl: relayBaseUrl, pubkey: pubkey, channelId: channelId, + generation: generation, members: channelListMembers, ); return channelListMembers; @@ -424,6 +450,7 @@ final channelMembersProvider = FutureProvider.autoDispose relayBaseUrl: relayBaseUrl, pubkey: pubkey, channelId: channelId, + generation: generation, members: const [], ); return const []; @@ -446,6 +473,7 @@ final channelMembersProvider = FutureProvider.autoDispose relayBaseUrl: relayBaseUrl, pubkey: pubkey, channelId: channelId, + generation: generation, members: members, ); return members; diff --git a/mobile/test/features/channels/channel_management_provider_test.dart b/mobile/test/features/channels/channel_management_provider_test.dart index 1a89ca45d0..3e968a2381 100644 --- a/mobile/test/features/channels/channel_management_provider_test.dart +++ b/mobile/test/features/channels/channel_management_provider_test.dart @@ -1,3 +1,5 @@ +import 'dart:async'; + import 'package:flutter_test/flutter_test.dart'; import 'package:buzz/features/channels/channel.dart'; import 'package:hooks_riverpod/hooks_riverpod.dart'; @@ -812,6 +814,91 @@ void main() { }, ); + for (final (label, olderAnswer) in [ + ('a roster', [_heldRoster('bot', 2)]), + ('no roster event', []), + ('an empty roster', [_emptyHeldRoster(2)]), + ]) { + test('a superseded roster read answering with $label does not replace ' + 'the offline roster', () async { + final session = _HeldRosterRelaySession(); + final container = ProviderContainer( + retry: (_, _) => null, + overrides: [relaySessionProvider.overrideWith(() => session)], + ); + addTearDown(container.dispose); + final subscription = container.listen( + channelMembersProvider(_channelId), + (_, _) {}, + ); + addTearDown(subscription.close); + Future role() async => (await container.read( + channelMembersProvider(_channelId).future, + )).single.role; + + expect(await role(), 'bot'); + await container.pump(); + + // Two roster updates arrive; the first read answers after the second. + final older = Completer>(); + session.nextRead = older.future; + session.emit(_heldRoster('bot', 2)); + await container.pump(); + final newer = Completer>(); + session.nextRead = newer.future; + session.emit(_heldRoster('member', 3)); + await container.pump(); + newer.complete([_heldRoster('member', 3)]); + expect(await role(), 'member'); + older.complete(olderAnswer); + await container.pump(); + + session.disconnect(); + await container.pump(); + expect(await role(), 'member'); + }); + } + + test( + 'a roster read torn down with no newer read still updates the offline roster', + () async { + final session = _HeldRosterRelaySession(); + final container = ProviderContainer( + retry: (_, _) => null, + overrides: [relaySessionProvider.overrideWith(() => session)], + ); + addTearDown(container.dispose); + var subscription = container.listen( + channelMembersProvider(_channelId), + (_, _) {}, + ); + Future role() async => (await container.read( + channelMembersProvider(_channelId).future, + )).single.role; + expect(await role(), 'bot'); + await container.pump(); + + // A roster update arrives; the user leaves before its read answers. + final fresh = Completer>(); + session.nextRead = fresh.future; + session.emit(_heldRoster('member', 2)); + await container.pump(); + subscription.close(); + await container.pump(); + fresh.complete([_heldRoster('member', 2)]); + await container.pump(); + + session.disconnect(); + await container.pump(); + subscription = container.listen( + channelMembersProvider(_channelId), + (_, _) {}, + ); + addTearDown(subscription.close); + expect(await role(), 'member'); + }, + ); + test('keeps the member snapshot available during reconnect', () { final cachedMembers = [ ChannelMember( @@ -949,3 +1036,63 @@ class _ConnectionAwareRelaySession extends RelaySessionNotifier { void Function(String message)? onClosed, }) async => () {}; } + +NostrEvent _heldRoster(String role, int version) => NostrEvent( + id: 'members-$version', + pubkey: 'owner', + createdAt: 1800000000 + version, + kind: 39002, + tags: [ + ['d', _channelId], + ['p', _memberPubkey, '', role], + ], + content: '', + sig: '', +); + +NostrEvent _emptyHeldRoster(int version) => NostrEvent( + id: 'empty-members-$version', + pubkey: 'owner', + createdAt: 1800000000 + version, + kind: 39002, + tags: [ + ['d', _channelId], + ], + content: '', + sig: '', +); + +/// Answers each roster read with [nextRead] and delivers live roster updates +/// through [emit]. +class _HeldRosterRelaySession extends RelaySessionNotifier { + Future> nextRead = Future.value([_heldRoster('bot', 1)]); + void Function(NostrEvent)? _listener; + + @override + SessionState build() => const SessionState(status: SessionStatus.connected); + + @override + Future> fetchHistory( + NostrFilter filter, { + Duration timeout = const Duration(seconds: 8), + }) => nextRead; + + @override + Future subscribeWithStatus( + NostrFilter filter, + void Function(NostrEvent) onEvent, { + void Function(String message)? onClosed, + required void Function(RelaySubscriptionStatus status) onStatusChanged, + }) async { + _listener = onEvent; + onStatusChanged(RelaySubscriptionStatus.ready); + return () { + if (identical(_listener, onEvent)) _listener = null; + }; + } + + void emit(NostrEvent event) => _listener?.call(event); + + void disconnect() => + state = const SessionState(status: SessionStatus.disconnected); +}