From 8c7b09c9ed884a96e5f12985787dff5f41ff0db4 Mon Sep 17 00:00:00 2001 From: grunch Date: Mon, 31 Aug 2026 21:15:32 -0300 Subject: [PATCH 1/2] perf: single dispute-chat REQ and constant-time session resolution The dispute chat notifier opened its own kind-14 REQ per dispute - 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, storage watcher, book listener) per candidate, on every incoming event, send and read-status check. - The notifier now consumes the manager's shared disputeChat stream (per-dispute filtering stays in the existing K_sign pre-filter), so one REQ serves all disputes and the manager's persisted shared cursor bounds the replay. - _getSessionForDispute resolves through the persisted session.disputeId first (constant time, side-effect free); the order-state scan remains only as a fallback for sessions persisted before disputeId existed. - Subscription.cancel tolerates teardown ordering: disposing the container while REQs are open no longer throws from onCancel. --- .../notifiers/dispute_chat_notifier.dart | 47 +++---- .../subscriptions/subscription_manager.dart | 8 +- .../dispute_chat_single_req_test.dart | 132 ++++++++++++++++++ 3 files changed, 160 insertions(+), 27 deletions(-) create mode 100644 test/features/disputes/dispute_chat_single_req_test.dart diff --git a/lib/features/disputes/notifiers/dispute_chat_notifier.dart b/lib/features/disputes/notifiers/dispute_chat_notifier.dart index 8f946813..7fc00d2c 100644 --- a/lib/features/disputes/notifiers/dispute_chat_notifier.dart +++ b/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'; @@ -160,32 +161,16 @@ class DisputeChatNotifier extends StateNotifier 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. + _subscription = ref + .read(subscriptionManagerProvider) + .disputeChat + .listen(_onChatEvent); + logger.i('Consuming shared dispute chat stream for dispute: $disputeId'); } /// Listen for session changes and subscribe when admin shared key is ready @@ -638,6 +623,16 @@ class DisputeChatNotifier extends StateNotifier 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 { diff --git a/lib/features/subscriptions/subscription_manager.dart b/lib/features/subscriptions/subscription_manager.dart index cb2fcfa8..1f65d003 100644 --- a/lib/features/subscriptions/subscription_manager.dart +++ b/lib/features/subscriptions/subscription_manager.dart @@ -471,7 +471,13 @@ class SubscriptionManager { request: request, streamSubscription: streamSubscription, onCancel: () { - ref.read(nostrServiceProvider).unsubscribe(request.subscriptionId!); + // Tolerate teardown ordering: when the container is being + // disposed, the socket (and its REQs) dies with it anyway. + try { + ref.read(nostrServiceProvider).unsubscribe(request.subscriptionId!); + } catch (e) { + logger.w('Skipping relay CLOSE during teardown: $e'); + } }, ); diff --git a/test/features/disputes/dispute_chat_single_req_test.dart b/test/features/disputes/dispute_chat_single_req_test.dart new file mode 100644 index 00000000..54443012 --- /dev/null +++ b/test/features/disputes/dispute_chat_single_req_test.dart @@ -0,0 +1,132 @@ +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 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.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 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.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'); + }); +} + +/// Session list the notifier can read without touching storage. +class _FixedSessionNotifier extends SessionNotifier { + _FixedSessionNotifier(Ref ref, List sessions) + : super(ref, MockSessionStorage(), MockSettings()) { + state = sessions; + } +} From 65ab00fc085074bc5180657a43f60dc6a9297c43 Mon Sep 17 00:00:00 2001 From: grunch Date: Tue, 1 Sep 2026 12:12:41 -0300 Subject: [PATCH 2/2] fix: replay the dispute-chat backlog when a notifier attaches MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `SubscriptionManager.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 the shared REQ delivered before that was discarded: not displayed, not persisted, and the cursor never advanced. That includes the backlog the relay replays when the REQ is first issued. Before this PR the notifier opened its own REQ on creation, so the relay always backfilled it. Adds `SubscriptionManager.refreshDisputeChatSubscription()`, asked for once by the notifier when it attaches: it clears the applied filter key and re-issues the REQ, so the relay replays from the persisted cursor with a listener connected. Still a single REQ — re-issued once, when a consumer shows up — and it also covers a dispute that starts mid-run, which eager notifier creation at startup would miss. Narrows the `onCancel` guard to the `ref.read`, per review: a failing `unsubscribe()` means the relay CLOSE was skipped and the REQ lingers, which is exactly the waste this PR removes, so it must surface rather than be swallowed. Doing so exposed a latent NPE — dart_nostr only assigns `subscriptionId` when it serializes the REQ onto a socket, so a request that never reached a relay has none — now handled as "nothing to CLOSE". Also refreshes the now-stale dispute comment in the foreground transition. --- .../notifiers/dispute_chat_notifier.dart | 22 ++++++++-- .../subscriptions/subscription_manager.dart | 38 ++++++++++++++++-- lib/services/lifecycle_manager.dart | 5 ++- .../dispute_chat_single_req_test.dart | 40 +++++++++++++++++++ 4 files changed, 97 insertions(+), 8 deletions(-) diff --git a/lib/features/disputes/notifiers/dispute_chat_notifier.dart b/lib/features/disputes/notifiers/dispute_chat_notifier.dart index 7fc00d2c..8821ad6f 100644 --- a/lib/features/disputes/notifiers/dispute_chat_notifier.dart +++ b/lib/features/disputes/notifiers/dispute_chat_notifier.dart @@ -111,6 +111,11 @@ class DisputeChatNotifier extends StateNotifier 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. @@ -166,11 +171,20 @@ class DisputeChatNotifier extends StateNotifier with MediaCach // 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. - _subscription = ref - .read(subscriptionManagerProvider) - .disputeChat - .listen(_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 diff --git a/lib/features/subscriptions/subscription_manager.dart b/lib/features/subscriptions/subscription_manager.dart index 1f65d003..8938b5ba 100644 --- a/lib/features/subscriptions/subscription_manager.dart +++ b/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,13 +472,25 @@ class SubscriptionManager { request: request, streamSubscription: streamSubscription, onCancel: () { - // Tolerate teardown ordering: when the container is being - // disposed, the socket (and its REQs) dies with it anyway. + // 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 { - ref.read(nostrServiceProvider).unsubscribe(request.subscriptionId!); + nostrService = ref.read(nostrServiceProvider); } catch (e) { logger.w('Skipping relay CLOSE during teardown: $e'); + return; } + nostrService.unsubscribe(subscriptionId); }, ); @@ -512,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]; diff --git a/lib/services/lifecycle_manager.dart b/lib/services/lifecycle_manager.dart index 5a771b98..3d3628fe 100644 --- a/lib/services/lifecycle_manager.dart +++ b/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); diff --git a/test/features/disputes/dispute_chat_single_req_test.dart b/test/features/disputes/dispute_chat_single_req_test.dart index 54443012..86b360c1 100644 --- a/test/features/disputes/dispute_chat_single_req_test.dart +++ b/test/features/disputes/dispute_chat_single_req_test.dart @@ -121,6 +121,46 @@ void main() { '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.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.