fix(mobile): refetch thread replies when the relay session reconnects

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<List<NostrEvent>, ThreadRepliesArgs>((
-      ref,
-      args,
-    ) async {
-      final session = ref.watch(relaySessionProvider.notifier);
+final threadRepliesProvider = FutureProvider.autoDispose
+    .family<List<NostrEvent>, 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 = <NostrEvent>[];
       _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<AsyncValue<List<NostrEvent>>, 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<AsyncValue<List<NostrEvent>>, ThreadRepliesArgs>((ref, args) {
       final relayReplies = ref.watch(threadRepliesProvider(args));
       final localReplies = ref.watch(threadLocalRepliesProvider(args));
       final authoritative = relayReplies.value;

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,

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<NostrEvent> replies = const [];
+  Completer<List<NostrEvent>>? nextQueryGate;
+
+  @override
+  SessionState build() => const SessionState(status: SessionStatus.connected);
+
+  void setStatus(SessionStatus status) {
+    state = SessionState(status: status);
+  }
+
+  @override
+  Future<List<NostrEvent>> queryRelay(
+    List<NostrFilter> 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<Object?>)
+  makeHarness(List<NostrEvent> 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<List<NostrEvent>>();
+    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);
+  });
+}