diff --git a/mobile/lib/features/channels/thread_replies_provider.dart b/mobile/lib/features/channels/thread_replies_provider.dart index 6904ba8c5e..3e0e9aa793 100644 --- a/mobile/lib/features/channels/thread_replies_provider.dart +++ b/mobile/lib/features/channels/thread_replies_provider.dart @@ -1,5 +1,6 @@ import 'dart:async'; +import 'package:flutter/foundation.dart'; import 'package:hooks_riverpod/hooks_riverpod.dart'; import '../../shared/relay/relay.dart'; @@ -29,12 +30,15 @@ class _ThreadCursor { const _ThreadCursor({required this.createdAt, required this.eventId}); } -final threadRepliesProvider = - FutureProvider.family, ThreadRepliesArgs>(( - ref, - args, - ) async { +final threadRepliesProvider = FutureProvider.autoDispose + .family, ThreadRepliesArgs>((ref, args) async { final session = ref.watch(relaySessionProvider.notifier); + // Establish the live subscription BEFORE snapshotting history. Otherwise + // an event published after `/query` returns but before the REQ registers + // is in neither result and is lost permanently — the same class of bug + // this provider exists to fix. `ChannelMessagesNotifier._init()` orders + // it the same way. + await ref.read(threadLiveRepliesProvider(args).notifier).subscribed; final replies = []; _ThreadCursor? cursor; for (var page = 0; page < 500; page++) { @@ -68,6 +72,160 @@ NostrFilter _threadRepliesFilter( ); } +/// Live relay replies for an open thread. +/// +/// [threadRepliesProvider] is a one-shot paginated query: it resolves once when +/// the thread opens and caches. Replies that arrive afterwards — including the +/// viewer's own, and agent replies — never entered that list, so an open thread +/// looked frozen while the activity feed (which subscribes) updated instantly. +/// +/// This subscribes to the same filter with `limit: 0` (live only, no history — +/// history is the paginated fetch's job) and accumulates arrivals. Merging +/// happens in [threadRepliesWithLocalProvider] via [_mergeReplies], which +/// de-duplicates by event id, so an event delivered both by the fetch and the +/// subscription appears once. +/// +/// Mirrors `ChannelsNotifier._subscribeLive`. +class ThreadLiveRepliesNotifier extends Notifier> { + final ThreadRepliesArgs args; + + ThreadLiveRepliesNotifier(this.args); + + void Function()? _unsubscribe; + bool _disposed = false; + int _subscribeGeneration = 0; + + /// Completes once the live REQ is registered, or has definitively failed. + /// [threadRepliesProvider] awaits this before issuing its history query, so + /// an event published between the history snapshot and the subscription + /// being registered cannot fall through the gap and be lost permanently. + /// `ChannelMessagesNotifier._init()` establishes the same ordering. + final Completer _subscribed = Completer(); + Future get subscribed => _subscribed.future; + + void _markSubscribed() { + if (!_subscribed.isCompleted) _subscribed.complete(); + } + + @override + List build() { + _disposed = false; + + ref.onDispose(() { + _disposed = true; + _clearSubscription(); + }); + + // Subscribe when connected. If already connected, initiate immediately. + // Listen for reconnection or late connection if not already subscribed. + ref.listen( + relaySessionProvider.select((session) => session.status), + (previous, next) { + if (next == SessionStatus.connected && _unsubscribe == null) { + unawaited(_subscribe()); + } + }, + ); + + if (ref.read(relaySessionProvider).status == SessionStatus.connected) { + unawaited(_subscribe()); + } + + return const []; + } + + Future _subscribe() async { + if (_disposed || _unsubscribe != null) return; + final generation = ++_subscribeGeneration; + final session = ref.read(relaySessionProvider.notifier); + try { + final unsubscribe = await session.subscribe( + _threadLiveRepliesFilter(args), + _handleLiveEvent, + onClosed: (message) => _handleClosed(generation, message), + ); + // Dispose can land while this await is suspended. On a fast thread + // open/close, onDispose runs while `_unsubscribe` is still null. + // Also check if a new subscription attempt began or if session disconnected. + if (_disposed || + generation != _subscribeGeneration || + ref.read(relaySessionProvider).status != SessionStatus.connected) { + unsubscribe(); + _markSubscribed(); + return; + } + _unsubscribe = unsubscribe; + } catch (error) { + if (generation == _subscribeGeneration) { + debugPrint( + '[ThreadLiveReplies] live subscription failed for ${args.rootId}: $error', + ); + } + } finally { + // Always release the history fetch, success or failure. A thread that + // could not subscribe must still load its backlog rather than hang. + _markSubscribed(); + } + } + + /// The relay sent CLOSED for this subscription. RelaySessionNotifier drops it + /// from `_liveSubscriptions`, so it will NOT be replayed on reconnect — but + /// this notifier would still hold a non-null `_unsubscribe` and report + /// healthy while silently receiving nothing again, which is the exact + /// failure this provider exists to fix. Clear local state so a later + /// reconnect can re-subscribe. The generation check stops a stale callback + /// from tearing down a newer subscription. + void _handleClosed(int generation, String message) { + if (_disposed || generation != _subscribeGeneration) return; + debugPrint( + '[ThreadLiveReplies] subscription CLOSED for ${args.rootId}: $message', + ); + _unsubscribe = null; + if (ref.read(relaySessionProvider).status == SessionStatus.connected) { + unawaited(_subscribe()); + } + } + + void _clearSubscription() { + _subscribeGeneration++; + _unsubscribe?.call(); + _unsubscribe = null; + } + + void _handleLiveEvent(NostrEvent event) { + if (_disposed) return; + state = _mergeReplies(state, [event]); + } +} + +/// AUTO-DISPOSE IS LOAD-BEARING. As a keep-alive family this leaks one live +/// relay subscription per thread opened for the lifetime of the app session — +/// `ref.onDispose` would only run when the container itself is torn down, so +/// the `_disposed` guard below would never fire in the case that matters. The +/// relay caps subscriptions (see buzz-relay handlers/req.rs), so the leak is +/// bounded only by eventual rejection. +/// +/// The consumer [threadRepliesWithLocalProvider] must be auto-dispose too: a +/// keep-alive consumer watching an auto-dispose provider keeps it alive anyway. +final threadLiveRepliesProvider = NotifierProvider.autoDispose + .family, ThreadRepliesArgs>( + ThreadLiveRepliesNotifier.new, + ); + +/// Live-only variant of [_threadRepliesFilter]: `limit: 0` asks the relay for +/// new events and no backlog. History is the paginated fetch's responsibility. +NostrFilter _threadLiveRepliesFilter(ThreadRepliesArgs args) { + return NostrFilter( + kinds: EventKind.channelTimelineContentKinds, + tags: { + '#e': [args.rootId], + '#h': [args.channelId], + }, + limit: 0, + extensions: {'depth_limit': 64}, + ); +} + class ThreadLocalRepliesNotifier extends Notifier> { final ThreadRepliesArgs args; @@ -99,13 +257,23 @@ final threadLocalRepliesProvider = /// Relay-backed replies merged with signed local replies that are still /// waiting for acknowledgement. -final threadRepliesWithLocalProvider = - Provider.family>, ThreadRepliesArgs>(( - ref, - args, - ) { +/// Auto-dispose: a keep-alive consumer would hold the auto-dispose providers +/// it watches alive indefinitely, defeating the subscription cleanup entirely. +final threadRepliesWithLocalProvider = Provider.autoDispose + .family>, ThreadRepliesArgs>((ref, args) { final relayReplies = ref.watch(threadRepliesProvider(args)); + final liveReplies = ref.watch(threadLiveRepliesProvider(args)); final localReplies = ref.watch(threadLocalRepliesProvider(args)); + // Ownership is released ONLY by the authoritative paginated fetch — a + // live echo does not count, even though it is displayed immediately. + // + // The live subscription can deliver an event the relay will not return + // on a later query (rejected downstream, dropped on a failover, CLOSED + // mid-flight). Confirming on the echo hands the row back to the relay, + // and if the refetch then fails there is nothing left holding the + // message on screen — the viewer's own reply disappears. Holding + // ownership until a query confirms costs one redundant entry that + // `_mergeReplies` de-duplicates by id anyway. final authoritative = relayReplies.value; if (authoritative != null && localReplies.isNotEmpty) { final authoritativeIds = authoritative.map((event) => event.id).toSet(); @@ -120,11 +288,14 @@ final threadRepliesWithLocalProvider = }); } } - if (localReplies.isEmpty) return relayReplies; + if (localReplies.isEmpty && liveReplies.isEmpty) return relayReplies; return relayReplies.when( - data: (events) => AsyncData(_mergeReplies(events, localReplies)), - loading: () => AsyncData(localReplies), - error: (error, stackTrace) => AsyncData(localReplies), + data: (events) => AsyncData( + _mergeReplies(_mergeReplies(events, liveReplies), localReplies), + ), + loading: () => AsyncData(_mergeReplies(liveReplies, localReplies)), + error: (error, stackTrace) => + AsyncData(_mergeReplies(liveReplies, localReplies)), ); }); diff --git a/mobile/test/features/channels/channel_messages_provider_test.dart b/mobile/test/features/channels/channel_messages_provider_test.dart index 6620c8c506..7493f75ba7 100644 --- a/mobile/test/features/channels/channel_messages_provider_test.dart +++ b/mobile/test/features/channels/channel_messages_provider_test.dart @@ -10,6 +10,71 @@ import 'package:buzz/features/channels/thread_replies_provider.dart'; import 'package:buzz/shared/relay/relay.dart'; void main() { + test( + 'surfaces live thread replies that arrive after the history fetch', + () async { + const rootId = 'thread-root'; + const args = ThreadRepliesArgs(channelId: _channelId, rootId: rootId); + final relaySession = _RecordingRelaySessionNotifier( + queryResults: [ + [ + _event( + id: 'history', + createdAt: 10, + extraTags: const [ + ['e', rootId], + ], + ), + ], + ], + ); + final container = _buildContainer(relaySession); + addTearDown(container.dispose); + + // These providers are auto-dispose: a bare `read` does not retain them, + // so the subscription would be torn down before the assertions run. Hold + // a listener for the life of the test, exactly as a mounted widget does. + final sub = container.listen( + threadRepliesWithLocalProvider(args), + (_, _) {}, + ); + addTearDown(sub.close); + + await relaySession.subscribed; + await _pumpEventQueue(); + + expect( + container + .read(threadRepliesWithLocalProvider(args)) + .value + ?.map((event) => event.id), + ['history'], + ); + + // The reply the open thread used to never see. + relaySession.emit( + _event( + id: 'live', + createdAt: 20, + extraTags: const [ + ['e', rootId], + ], + ), + ); + await _pumpEventQueue(); + + expect( + container + .read(threadRepliesWithLocalProvider(args)) + .value + ?.map((event) => event.id), + ['history', 'live'], + ); + // Live subscription must not re-request history. + expect(relaySession.liveFilters.last.limit, 0); + }, + ); + test( 'keeps live events that arrive while initial history is loading', () async { @@ -307,7 +372,13 @@ void main() { await relaySession.subscribed; await _pumpEventQueue(); const args = ThreadRepliesArgs(channelId: _channelId, rootId: 'root'); - container.read(threadRepliesWithLocalProvider(args)); + // Auto-dispose: hold a listener so the provider is not torn down while + // still awaiting its live subscription, as a mounted widget would. + final threadSub = container.listen( + threadRepliesWithLocalProvider(args), + (_, _) {}, + ); + addTearDown(threadSub.close); await _pumpEventQueue(); final notifier = container.read( channelMessagesProvider(_channelId).notifier, @@ -386,7 +457,13 @@ void main() { await relaySession.subscribed; await _pumpEventQueue(); const args = ThreadRepliesArgs(channelId: _channelId, rootId: 'root'); - container.read(threadRepliesWithLocalProvider(args)); + // Auto-dispose: hold a listener so the provider is not torn down while + // still awaiting its live subscription, as a mounted widget would. + final threadSub = container.listen( + threadRepliesWithLocalProvider(args), + (_, _) {}, + ); + addTearDown(threadSub.close); await _pumpEventQueue(); final notifier = container.read( channelMessagesProvider(_channelId).notifier,