fix(mobile): live-update open threads instead of freezing on the history fetch

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<List<NostrEvent>, ThreadRepliesArgs>((
-      ref,
-      args,
-    ) async {
+final threadRepliesProvider = FutureProvider.autoDispose
+    .family<List<NostrEvent>, 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 = <NostrEvent>[];
       _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<List<NostrEvent>> {
+  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<void> _subscribed = Completer<void>();
+  Future<void> get subscribed => _subscribed.future;
+
+  void _markSubscribed() {
+    if (!_subscribed.isCompleted) _subscribed.complete();
+  }
+
+  @override
+  List<NostrEvent> 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<SessionStatus>(
+      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<void> _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<ThreadLiveRepliesNotifier, List<NostrEvent>, 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<List<NostrEvent>> {
   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<AsyncValue<List<NostrEvent>>, 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<AsyncValue<List<NostrEvent>>, 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)),
       );
     });
 

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,