perf: single dispute-chat REQ and constant-time session resolution

lib/features/disputes/notifiers/dispute_chat_notifier.dart

@@ -17,6 +17,7 @@ import 'package:mostro_mobile/shared/mixins/media_cache_mixin.dart';
 import 'package:mostro_mobile/shared/utils/chat_keys.dart';
 import 'package:mostro_mobile/shared/utils/nostr_utils.dart';
 import 'package:mostro_mobile/shared/providers/mostro_service_provider.dart';
+import 'package:mostro_mobile/features/subscriptions/subscription_manager_provider.dart';
 import 'package:mostro_mobile/shared/providers/nostr_service_provider.dart';
 import 'package:mostro_mobile/shared/providers/session_notifier_provider.dart';
 import 'package:sembast/sembast.dart';
@@ -110,6 +111,11 @@ class DisputeChatNotifier extends StateNotifier<DisputeChatState> with MediaCach
   ChatKeys? _chatKeys;
   String? _chatKeysSource;
 
+  /// Whether the one-shot catch-up re-issue of the shared dispute-chat REQ
+  /// has already been asked for. [_subscribe] can run more than once (it
+  /// retries once the admin shared key lands), and one replay is enough.
+  bool _requestedBackfill = false;
+
   DisputeChatNotifier(this.disputeId, this.ref) : super(const DisputeChatState());
 
   /// Derive (and cache) the K_conv/K_sign pair from the admin shared key.
@@ -160,32 +166,25 @@ class DisputeChatNotifier extends StateNotifier<DisputeChatState> with MediaCach
       _subscription = null;
     }
 
-    // Subscribe to kind 14 chat events authored by K_sign. The spec requires
-    // filtering by authors, not #p, to prevent third-party flooding.
-    final chatKeys = _getChatKeys(session);
-    final nostrService = ref.read(nostrServiceProvider);
-
-    // Persisted per-conversation cursor (spec MUST); default lookback for
-    // conversations with no accepted events yet
-    final cursorSince =
-        await ref.read(disputeChatCursorStoreProvider).sinceFor(disputeId);
-    if (!mounted) return;
-    final since = cursorSince ??
-        DateTime.now().subtract(NostrEventExtensions.chatDefaultLookback);
-
-    final request = NostrRequest(
-      filters: [
-        NostrFilter(
-          kinds: [14],
-          authors: [chatKeys.sign.public],
-          since: since,
-          limit: NostrEventExtensions.chatDefaultLimit,
-        ),
-      ],
-    );
-
-    _subscription = nostrService.subscribeToEvents(request).listen(_onChatEvent);
-    logger.i('Subscribed to kind 14 chat via K_sign for dispute: $disputeId');
+    // Consume the app-wide dispute-chat stream: SubscriptionManager already
+    // maintains the kind-14 REQ (K_sign authors + shared persisted cursor)
+    // for every dispute session, and a private REQ here duplicated it on
+    // every relay. Events for other disputes are dropped by the K_sign
+    // pre-filter in _onChatEvent.
+    final subscriptionManager = ref.read(subscriptionManagerProvider);
+    _subscription = subscriptionManager.disputeChat.listen(_onChatEvent);
+    logger.i('Consuming shared dispute chat stream for dispute: $disputeId');
+
+    // The shared stream is a broadcast controller, so it dropped everything
+    // delivered while this lazily built notifier did not exist — including
+    // the backlog replayed when the REQ was first issued. Ask for one
+    // re-issue so the relay replays from the persisted cursor now that a
+    // listener is attached; without it an admin message that arrived before
+    // the Disputes tab was opened stays invisible until a background cycle.
+    if (!_requestedBackfill) {
+      _requestedBackfill = true;
+      subscriptionManager.refreshDisputeChatSubscription();
+    }
   }
 
   /// Listen for session changes and subscribe when admin shared key is ready
@@ -638,6 +637,16 @@ class DisputeChatNotifier extends StateNotifier<DisputeChatState> with MediaCach
     try {
       final sessions = ref.read(sessionNotifierProvider);
 
+      // Direct lookup: disputeId is persisted on the session when the
+      // dispute starts. Constant time, no side effects — the scan below
+      // instantiated an OrderNotifier per candidate on every event.
+      for (final session in sessions) {
+        if (session.disputeId == disputeId) {
+          return session;
+        }
+      }
+
+      // Fallback for sessions persisted before disputeId existed.
       for (final session in sessions) {
         if (session.orderId != null) {
           try {

lib/features/subscriptions/subscription_manager.dart

@@ -4,6 +4,7 @@ import 'package:dart_nostr/dart_nostr.dart';
 import 'package:flutter/foundation.dart';
 import 'package:flutter_riverpod/flutter_riverpod.dart';
 import 'package:mostro_mobile/services/logger_service.dart';
+import 'package:mostro_mobile/services/nostr_service.dart';
 import 'package:mostro_mobile/core/models/relay_list_event.dart';
 import 'package:mostro_mobile/data/models/nostr_event.dart';
 import 'package:mostro_mobile/data/models/session.dart';
@@ -471,7 +472,25 @@ class SubscriptionManager {
       request: request,
       streamSubscription: streamSubscription,
       onCancel: () {
-        ref.read(nostrServiceProvider).unsubscribe(request.subscriptionId!);
+        // dart_nostr assigns the id when it serializes the REQ onto a socket,
+        // so a request that never reached a relay (none connected) has none
+        // and there is nothing to CLOSE.
+        final subscriptionId = request.subscriptionId;
+        if (subscriptionId == null) return;
+
+        // Tolerate teardown ordering: reading the provider throws once the
+        // container is being disposed, and the socket (and its REQs) dies
+        // with it anyway. Only the read is guarded — a failing unsubscribe()
+        // means the relay CLOSE was skipped and the REQ lingers, which is
+        // exactly the waste this path exists to avoid, so it must surface.
+        final NostrService nostrService;
+        try {
+          nostrService = ref.read(nostrServiceProvider);
+        } catch (e) {
+          logger.w('Skipping relay CLOSE during teardown: $e');
+          return;
+        }
+        nostrService.unsubscribe(subscriptionId);
       },
     );
 
@@ -506,6 +525,25 @@ class SubscriptionManager {
     );
   }
 
+  /// Re-issues the dispute-chat REQ so the relay replays from the persisted
+  /// cursor.
+  ///
+  /// [disputeChat] is a broadcast stream, and a broadcast stream drops events
+  /// while nothing is listening. `DisputeChatNotifier` is built lazily — only
+  /// when the Disputes tab renders — so every envelope delivered before that,
+  /// including the backlog replayed when the REQ is first issued, would be
+  /// lost to the listener that shows up afterwards. Clearing the applied
+  /// filter key forces the next update past the identity skip. Still a single
+  /// REQ: it is re-issued once, when a consumer attaches.
+  void refreshDisputeChatSubscription() {
+    if (_isSuspended) return;
+    final sessions = ref.read(sessionNotifierProvider);
+    if (sessions.isEmpty) return;
+    logger.i('Re-issuing dispute chat REQ for a newly attached listener');
+    unsubscribeByType(SubscriptionType.disputeChat);
+    unawaited(_updateSubscription(SubscriptionType.disputeChat, sessions));
+  }
+
   void unsubscribeByType(SubscriptionType type) {
     _appliedFilterKeys.remove(type);
     final subscription = _subscriptions[type];

lib/services/lifecycle_manager.dart

@@ -119,7 +119,10 @@ class LifecycleManager extends WidgetsBindingObserver {
 
       // Reload dispute chats: the background service persists admin messages
       // to disk while the app sleeps, but an already-initialized notifier
-      // never re-reads storage nor re-opens its relay subscription on its own
+      // never re-reads storage on its own. It no longer owns a relay
+      // subscription — it consumes SubscriptionManager.disputeChat — so the
+      // rebuild also re-attaches its listener and asks for the catch-up
+      // re-issue.
       logger.i("Reloading dispute chats");
       ref.invalidate(disputeChatNotifierProvider);
 

test/features/disputes/dispute_chat_single_req_test.dart

@@ -0,0 +1,172 @@
+import 'dart:async';
+
+import 'package:dart_nostr/dart_nostr.dart';
+import 'package:flutter_riverpod/flutter_riverpod.dart';
+import 'package:flutter_test/flutter_test.dart';
+import 'package:mockito/mockito.dart';
+import 'package:mostro_mobile/data/models/nostr_event.dart';
+import 'package:mostro_mobile/data/models/session.dart';
+import 'package:mostro_mobile/data/repositories/event_storage.dart';
+import 'package:mostro_mobile/features/disputes/notifiers/dispute_chat_notifier.dart';
+import 'package:mostro_mobile/features/order/providers/order_notifier_provider.dart';
+import 'package:mostro_mobile/features/subscriptions/subscription_manager_provider.dart';
+import 'package:mostro_mobile/shared/notifiers/session_notifier.dart';
+import 'package:mostro_mobile/shared/providers/mostro_service_provider.dart';
+import 'package:mostro_mobile/shared/providers/session_notifier_provider.dart';
+import 'package:mostro_mobile/shared/providers/storage_providers.dart';
+import 'package:mostro_mobile/shared/utils/chat_keys.dart';
+import 'package:sembast/sembast_memory.dart';
+import 'package:shared_preferences/shared_preferences.dart';
+import 'package:shared_preferences_platform_interface/in_memory_shared_preferences_async.dart';
+import 'package:shared_preferences_platform_interface/shared_preferences_async_platform_interface.dart';
+
+import '../../mocks.mocks.dart';
+
+/// The dispute chat notifier used to open its OWN kind-14 REQ — a duplicate
+/// of the one `SubscriptionManager` already maintains for
+/// `SubscriptionType.disputeChat` (whose stream nobody consumed) — and
+/// resolved its session by scanning every session and instantiating an
+/// `OrderNotifier` (DB sync + subscriptions) per candidate, on EVERY
+/// incoming event. It now consumes the manager's stream and resolves the
+/// session through the persisted `session.disputeId`.
+void main() {
+  TestWidgetsFlutterBinding.ensureInitialized();
+
+  const disputeId = 'dispute-single-req';
+  const orderId = 'order-single-req';
+  final tradeKey = NostrKeyPairs(
+    private:
+        '0000000000000000000000000000000000000000000000000000000000000005',
+  );
+  final adminKey = NostrKeyPairs(
+    private:
+        '0000000000000000000000000000000000000000000000000000000000000006',
+  );
+
+  late Session session;
+  late StreamController<NostrEvent> disputeStream;
+  late MockSubscriptionManagerSpy manager;
+  late ProviderContainer container;
+
+  setUp(() async {
+    SharedPreferencesAsyncPlatform.instance =
+        InMemorySharedPreferencesAsync.empty();
+
+    session = Session(
+      masterKey: tradeKey,
+      tradeKey: tradeKey,
+      keyIndex: 1,
+      fullPrivacy: false,
+      startTime: DateTime.now(),
+      orderId: orderId,
+    )
+      ..disputeId = disputeId
+      ..setAdminPeer(adminKey.public);
+
+    disputeStream = StreamController<NostrEvent>.broadcast();
+    manager = MockSubscriptionManagerSpy();
+    when(manager.disputeChat).thenAnswer((_) => disputeStream.stream);
+
+    final db = await newDatabaseFactoryMemory()
+        .openDatabase('dispute_single_req.db');
+
+    container = ProviderContainer(overrides: [
+      sharedPreferencesProvider.overrideWithValue(SharedPreferencesAsync()),
+      eventStorageProvider.overrideWithValue(EventStorage(db: db)),
+      subscriptionManagerProvider.overrideWithValue(manager),
+      mostroServiceProvider
+          .overrideWith((ref) => throw UnimplementedError('unused')),
+      sessionNotifierProvider
+          .overrideWith((ref) => _FixedSessionNotifier(ref, [session])),
+      // The old resolution instantiated an OrderNotifier per session per
+      // event; the disputeId lookup must never need one.
+      orderNotifierProvider.overrideWith(
+        (ref, id) =>
+            throw StateError('session resolution must not build notifiers'),
+      ),
+    ]);
+    addTearDown(container.dispose);
+    addTearDown(disputeStream.close);
+  });
+
+  Future<NostrEvent> adminEnvelope(String text) {
+    final chatKeys = ChatKeys.fromSharedKey(session.adminSharedKey!);
+    final rumor = NostrEventExtensions.createChatRumor(
+      senderKeys: adminKey,
+      content: text,
+    );
+    return rumor.chatWrap(chatKeys);
+  }
+
+  test('events from the shared manager stream reach the notifier', () async {
+    container.read(disputeChatNotifierProvider(disputeId).notifier);
+    await pumpEventQueue(times: 50);
+
+    disputeStream.add(await adminEnvelope('hola admin'));
+
+    final deadline = DateTime.now().add(const Duration(seconds: 5));
+    while (container
+            .read(disputeChatNotifierProvider(disputeId))
+            .messages
+            .isEmpty &&
+        DateTime.now().isBefore(deadline)) {
+      await Future<void>.delayed(const Duration(milliseconds: 20));
+    }
+
+    final messages =
+        container.read(disputeChatNotifierProvider(disputeId)).messages;
+    expect(messages, hasLength(1),
+        reason: 'the notifier must consume SubscriptionManager.disputeChat '
+            'instead of opening its own REQ, and resolve its session via '
+            'session.disputeId without touching order notifiers');
+    expect(messages.single.content, 'hola admin');
+  });
+
+  test('an envelope delivered before the notifier exists is still recovered',
+      () async {
+    // `disputeChat` is a broadcast stream and `DisputeChatNotifier` is built
+    // lazily, only when the Disputes tab renders. Everything the shared REQ
+    // delivers before that — including the backlog replayed when the REQ is
+    // first issued — is dropped for want of a listener. The notifier must ask
+    // for a catch-up re-issue when it attaches, so the relay replays from the
+    // persisted cursor with the listener connected.
+    final backlog = await adminEnvelope('mensaje del admin');
+
+    // Relay replay: the re-issued REQ resends what the cursor still covers.
+    when(manager.refreshDisputeChatSubscription()).thenAnswer((_) {
+      disputeStream.add(backlog);
+    });
+
+    // Delivered while nothing is listening -> dropped by the broadcast stream.
+    disputeStream.add(backlog);
+    await pumpEventQueue(times: 10);
+
+    // Only now does the user open the Disputes tab.
+    container.read(disputeChatNotifierProvider(disputeId).notifier);
+
+    final deadline = DateTime.now().add(const Duration(seconds: 5));
+    while (container
+            .read(disputeChatNotifierProvider(disputeId))
+            .messages
+            .isEmpty &&
+        DateTime.now().isBefore(deadline)) {
+      await Future<void>.delayed(const Duration(milliseconds: 20));
+    }
+
+    verify(manager.refreshDisputeChatSubscription()).called(1);
+    final messages =
+        container.read(disputeChatNotifierProvider(disputeId)).messages;
+    expect(messages, hasLength(1),
+        reason: 'an admin message delivered before the Disputes tab was '
+            'opened must not stay invisible');
+    expect(messages.single.content, 'mensaje del admin');
+  });
+}
+
+/// Session list the notifier can read without touching storage.
+class _FixedSessionNotifier extends SessionNotifier {
+  _FixedSessionNotifier(Ref ref, List<Session> sessions)
+      : super(ref, MockSessionStorage(), MockSettings()) {
+    state = sessions;
+  }
+}