diff --git a/mobile/lib/features/channels/thread_replies_provider.dart b/mobile/lib/features/channels/thread_replies_provider.dart index 81ab5a4b39d..3000e58221e 100644 --- a/mobile/lib/features/channels/thread_replies_provider.dart +++ b/mobile/lib/features/channels/thread_replies_provider.dart @@ -29,12 +29,22 @@ class _ThreadCursor { const _ThreadCursor({required this.createdAt, required this.eventId}); } -final threadRepliesProvider = - FutureProvider.family, ThreadRepliesArgs>(( - ref, - args, - ) async { - final session = ref.watch(relaySessionProvider.notifier); +final threadRepliesProvider = FutureProvider.autoDispose + .family, ThreadRepliesArgs>((ref, args) async { + // Refetch when the relay session reconnects: a reply that arrives while + // the connection is down is otherwise unreachable — the replayed live + // subscription only starts at the reconnect time, and the channel + // window backfill carries thread summaries, not reply bodies. Listening + // for the connected edge (rather than watching the status) refreshes + // exactly once per successful reconnect, does nothing on the disconnect + // edge, and keeps the previous replies rendered while the refresh runs. + ref.listen(relaySessionProvider, (previous, next) { + if (previous?.status != SessionStatus.connected && + next.status == SessionStatus.connected) { + ref.invalidateSelf(); + } + }); + final session = ref.read(relaySessionProvider.notifier); final replies = []; _ThreadCursor? cursor; for (var page = 0; page < 500; page++) { @@ -99,11 +109,13 @@ final threadLocalRepliesProvider = /// Relay-backed replies merged with signed local replies that are still /// waiting for acknowledgement. -final threadRepliesWithLocalProvider = - Provider.family>, ThreadRepliesArgs>(( - ref, - args, - ) { +/// +/// Auto-disposed together with [threadRepliesProvider] so a reopened thread +/// always starts from a fresh authoritative query; [threadLocalRepliesProvider] +/// stays alive so an optimistic just-sent reply survives route changes until +/// the relay confirms it. +final threadRepliesWithLocalProvider = Provider.autoDispose + .family>, ThreadRepliesArgs>((ref, args) { final relayReplies = ref.watch(threadRepliesProvider(args)); final localReplies = ref.watch(threadLocalRepliesProvider(args)); final authoritative = relayReplies.value; diff --git a/mobile/test/features/channels/channel_messages_provider_test.dart b/mobile/test/features/channels/channel_messages_provider_test.dart index 4f4efdce1f7..a0bd94cd5a2 100644 --- a/mobile/test/features/channels/channel_messages_provider_test.dart +++ b/mobile/test/features/channels/channel_messages_provider_test.dart @@ -384,7 +384,9 @@ void main() { await relaySession.subscribed; await _pumpEventQueue(); const args = ThreadRepliesArgs(channelId: _channelId, rootId: 'root'); - container.read(threadRepliesWithLocalProvider(args)); + // The thread providers auto-dispose without a listener; hold one for + // the test's duration, mirroring an open thread page. + container.listen(threadRepliesWithLocalProvider(args), (_, _) {}); await _pumpEventQueue(); final notifier = container.read( channelMessagesProvider(_channelId).notifier, @@ -463,7 +465,9 @@ void main() { await relaySession.subscribed; await _pumpEventQueue(); const args = ThreadRepliesArgs(channelId: _channelId, rootId: 'root'); - container.read(threadRepliesWithLocalProvider(args)); + // The thread providers auto-dispose without a listener; hold one for + // the test's duration, mirroring an open thread page. + container.listen(threadRepliesWithLocalProvider(args), (_, _) {}); await _pumpEventQueue(); final notifier = container.read( channelMessagesProvider(_channelId).notifier, diff --git a/mobile/test/features/channels/thread_replies_provider_test.dart b/mobile/test/features/channels/thread_replies_provider_test.dart new file mode 100644 index 00000000000..ef58bfd5a13 --- /dev/null +++ b/mobile/test/features/channels/thread_replies_provider_test.dart @@ -0,0 +1,160 @@ +import 'dart:async'; + +import 'package:buzz/features/channels/thread_replies_provider.dart'; +import 'package:buzz/shared/relay/relay.dart'; +import 'package:flutter_test/flutter_test.dart'; +import 'package:hooks_riverpod/hooks_riverpod.dart'; + +class _FakeRelaySession extends RelaySessionNotifier { + int queryCount = 0; + List replies = const []; + Completer>? nextQueryGate; + + @override + SessionState build() => const SessionState(status: SessionStatus.connected); + + void setStatus(SessionStatus status) { + state = SessionState(status: status); + } + + @override + Future> queryRelay( + List filters, { + Duration timeout = const Duration(seconds: 8), + }) async { + queryCount++; + final gate = nextQueryGate; + if (gate != null) { + nextQueryGate = null; + return gate.future; + } + return replies; + } +} + +NostrEvent _reply(String id, int createdAt) => NostrEvent( + id: id, + pubkey: 'bob', + createdAt: createdAt, + kind: EventKind.streamMessage, + tags: const [ + ['h', 'chan'], + ['e', 'root', '', 'reply'], + ], + content: 'reply $id', + sig: '', +); + +void main() { + const args = ThreadRepliesArgs(channelId: 'chan', rootId: 'root'); + + (ProviderContainer, _FakeRelaySession, ProviderSubscription) + makeHarness(List initialReplies) { + final fakeSession = _FakeRelaySession()..replies = initialReplies; + final container = ProviderContainer( + overrides: [relaySessionProvider.overrideWith(() => fakeSession)], + ); + // An auto-disposed provider needs a listener to stay alive, mirroring an + // open thread page. Creating it starts the first load, so the fake's + // replies must be in place first. + final subscription = container.listen( + threadRepliesProvider(args), + (_, _) {}, + ); + return (container, fakeSession, subscription); + } + + test('does not refetch on the disconnect edge', () async { + final (container, fakeSession, _) = makeHarness([_reply('r1', 1000)]); + addTearDown(container.dispose); + + await container.read(threadRepliesProvider(args).future); + final queriesAfterFirstLoad = fakeSession.queryCount; + + fakeSession.setStatus(SessionStatus.disconnected); + await container.pump(); + + expect(fakeSession.queryCount, queriesAfterFirstLoad); + }); + + test('refetches exactly once per reconnect edge', () async { + final (container, fakeSession, _) = makeHarness([_reply('r1', 1000)]); + addTearDown(container.dispose); + + final first = await container.read(threadRepliesProvider(args).future); + expect(first.map((event) => event.id), ['r1']); + final queriesAfterFirstLoad = fakeSession.queryCount; + + // A reply lands while the connection is down. + fakeSession.replies = [_reply('r1', 1000), _reply('r2', 2000)]; + fakeSession.setStatus(SessionStatus.disconnected); + await container.pump(); + fakeSession.setStatus(SessionStatus.connected); + await container.pump(); + + final second = await container.read(threadRepliesProvider(args).future); + expect(second.map((event) => event.id), ['r1', 'r2']); + expect(fakeSession.queryCount, queriesAfterFirstLoad + 1); + }); + + test( + 'does not refetch on session emissions that keep the same status', + () async { + final (container, fakeSession, _) = makeHarness([_reply('r1', 1000)]); + addTearDown(container.dispose); + + await container.read(threadRepliesProvider(args).future); + final queriesAfterFirstLoad = fakeSession.queryCount; + + // Same connected status, new state object (e.g. reconnectAttempt bump). + fakeSession.setStatus(SessionStatus.connected); + await container.pump(); + + expect(fakeSession.queryCount, queriesAfterFirstLoad); + }, + ); + + test('keeps previous replies available while a refresh is pending', () async { + final (container, fakeSession, _) = makeHarness([_reply('r1', 1000)]); + addTearDown(container.dispose); + + await container.read(threadRepliesProvider(args).future); + + // Hold the reconnect refresh open and verify the old data still reads. + final gate = Completer>(); + fakeSession.nextQueryGate = gate; + fakeSession.setStatus(SessionStatus.disconnected); + await container.pump(); + fakeSession.setStatus(SessionStatus.connected); + await container.pump(); + + final pending = container.read(threadRepliesProvider(args)); + expect(pending.isLoading, isTrue); + expect(pending.value?.map((event) => event.id), ['r1']); + + gate.complete([_reply('r1', 1000), _reply('r2', 2000)]); + final refreshed = await container.read(threadRepliesProvider(args).future); + expect(refreshed.map((event) => event.id), ['r1', 'r2']); + }); + + test('reopening a disposed thread performs a fresh load', () async { + final (container, fakeSession, subscription) = makeHarness([ + _reply('r1', 1000), + ]); + addTearDown(container.dispose); + + await container.read(threadRepliesProvider(args).future); + final queriesAfterFirstLoad = fakeSession.queryCount; + + // Close the page: the auto-disposed query is torn down… + subscription.close(); + await container.pump(); + + // …so reopening loads fresh instead of serving a stale cache. + fakeSession.replies = [_reply('r1', 1000), _reply('r2', 2000)]; + container.listen(threadRepliesProvider(args), (_, _) {}); + final reopened = await container.read(threadRepliesProvider(args).future); + expect(reopened.map((event) => event.id), ['r1', 'r2']); + expect(fakeSession.queryCount, queriesAfterFirstLoad + 1); + }); +}