diff --git a/integration_test/test_helpers.dart b/integration_test/test_helpers.dart index ddabe916..378cb32a 100644 --- a/integration_test/test_helpers.dart +++ b/integration_test/test_helpers.dart @@ -267,6 +267,9 @@ class FakeMostroService implements MostroService { @override final Ref ref; + @override + Set get debugHeldEventIds => const {}; + @override void init({List? keys}) {} diff --git a/lib/features/subscriptions/subscription_manager.dart b/lib/features/subscriptions/subscription_manager.dart index cb2fcfa8..9494606d 100644 --- a/lib/features/subscriptions/subscription_manager.dart +++ b/lib/features/subscriptions/subscription_manager.dart @@ -295,6 +295,16 @@ class SubscriptionManager { .whereType(); await ref.read(chatCursorStoreProvider).warmUp(orderIds); } + if (type == SubscriptionType.orders) { + // Best-effort: a prefs/storage failure must not prevent the REQ. + try { + await ref + .read(ordersCursorStoreProvider) + .warmUp([ref.read(settingsProvider).mostroPublicKey]); + } catch (e) { + logger.w('Orders cursor warm-up unavailable: $e'); + } + } final filter = _createFilterForType(type, sessions); if (filter == null) { @@ -352,6 +362,21 @@ class SubscriptionManager { } } + /// Earliest start time across [sessions], or the default lookback when that + /// is older — whichever reaches further back. A session older than the + /// lookback widens the window to cover it, and a newer one cannot narrow it + /// below the lookback. Used as the bootstrap `since` when no cursor is + /// stored yet. + DateTime _sessionsFloor(List sessions) { + final defaultFloor = + DateTime.now().subtract(NostrEventExtensions.chatDefaultLookback); + if (sessions.isEmpty) return defaultFloor; + final oldest = sessions + .map((s) => s.startTime) + .reduce((a, b) => a.isBefore(b) ? a : b); + return oldest.isBefore(defaultFloor) ? oldest : defaultFloor; + } + NostrFilter? _createFilterForType( SubscriptionType type, List sessions) { switch (type) { @@ -365,10 +390,30 @@ class SubscriptionManager { // and re-subscribe when the node info arrives after this subscription. final transport = _resolveOrdersTransport(); _appliedOrdersTransport = transport; + final mostroPubkey = ref.read(settingsProvider).mostroPublicKey; + // Persisted cursor bounds the replay; fresh installs fall back to the + // default lookback (older history is served by the restore flow). + DateTime? cursorSince; + try { + cursorSince = + ref.read(ordersCursorStoreProvider).cachedSinceFor(mostroPubkey); + } catch (e) { + logger.w('Orders cursor unavailable, using default lookback: $e'); + } + // No cursor yet means a fresh install *or* the first launch after + // upgrading to the cursor build. An upgrading install can hold orders + // far older than the default lookback (non-terminal orders are kept + // well past 30 days) and normal startup does not run the restore + // flow, so the window must also reach back to the oldest live + // session: no message of an active order predates its session. + final ordersSince = cursorSince ?? + _sessionsFloor(sessions) + .subtract(ChatCursorStore.cursorOverlap); return buildOrdersFilter( transport, tradeKeys, - ref.read(settingsProvider).mostroPublicKey, + mostroPubkey, + since: ordersSince, ); case SubscriptionType.chat: // Kind 14 chat envelope: filter by the K_sign authors derived from @@ -670,19 +715,34 @@ class SubscriptionManager { NostrFilter buildOrdersFilter( Transport transport, List tradeKeys, - String mostroPubkey, -) { + String mostroPubkey, { + DateTime? since, +}) { switch (transport) { case Transport.giftWrap: + // Legacy transport: gift wrap timestamps are randomized ±48 h, so a + // cursor since would silently drop messages. Left unbounded until the + // 1059 branch is removed. return NostrFilter( kinds: [1059], p: tradeKeys, ); case Transport.nip44: + // kind 14 carries real timestamps, so the persisted cursor (minus its + // overlap) bounds the replay. Deliberately no limit on top of it: the + // relay answers a capped filter with the *newest* n and silently drops + // the rest of the window, and events that are never delivered never + // enter _retryableEvents, so nothing holds the cursor back — it + // advances past them and they are lost for good. That window is widest + // exactly when it matters (the first launch after upgrading, or a user + // offline for a long stretch). The filter is already scoped to one + // node's messages addressed to this user's trade keys, so `since` + // alone keeps it small. return NostrFilter( kinds: [14], authors: [mostroPubkey], p: tradeKeys, + since: since, ); } } diff --git a/lib/services/chat_cursor_store.dart b/lib/services/chat_cursor_store.dart index 32f786f4..12bd04c0 100644 --- a/lib/services/chat_cursor_store.dart +++ b/lib/services/chat_cursor_store.dart @@ -22,6 +22,10 @@ class ChatCursorStore { /// generalization; keeping it preserves cursors stored by older builds. static const disputeKeyPrefix = 'dispute_chat_since_'; + /// Orders (node message) namespace, keyed by the node pubkey: one live + /// orders subscription exists per connected node. + static const ordersKeyPrefix = 'orders_since_'; + /// Peer (buyer-seller) chat namespace, keyed by orderId. Shared with the /// background isolate, which builds its own store without Riverpod. static const peerKeyPrefix = 'chat_since_'; @@ -120,3 +124,11 @@ final chatCursorStoreProvider = Provider( keyPrefix: ChatCursorStore.peerKeyPrefix, ), ); + +/// Orders (node kind-14 message) cursors, keyed by the node pubkey. +final ordersCursorStoreProvider = Provider( + (ref) => ChatCursorStore( + ref.watch(sharedPreferencesProvider), + keyPrefix: ChatCursorStore.ordersKeyPrefix, + ), +); diff --git a/lib/services/mostro_service.dart b/lib/services/mostro_service.dart index 3df7c3bc..f907cdbd 100644 --- a/lib/services/mostro_service.dart +++ b/lib/services/mostro_service.dart @@ -3,7 +3,9 @@ import 'dart:convert'; import 'package:collection/collection.dart'; import 'package:dart_nostr/dart_nostr.dart'; +import 'package:flutter/foundation.dart'; import 'package:flutter_riverpod/flutter_riverpod.dart'; +import 'package:mostro_mobile/services/chat_cursor_store.dart'; import 'package:mostro_mobile/services/logger_service.dart'; import 'package:mostro_mobile/data/enums.dart'; import 'package:mostro_mobile/data/models.dart'; @@ -119,11 +121,70 @@ class MostroService { /// message write) — the daemon's message then never notified, was skipped /// by every later replay, and the order sat at a stale status across /// restarts. - Future _markEventProcessed(NostrEvent event) { - return ref.read(eventStorageProvider).putItem(event.id!, { - 'id': event.id, + Future _markEventProcessed(NostrEvent event) async { + final eventId = event.id!; + await ref.read(eventStorageProvider).putItem(eventId, { + 'id': eventId, 'created_at': event.createdAt!.millisecondsSinceEpoch ~/ 1000, }); + // Only once the durable marker is written may the cursor move: a failed + // putItem leaves the event unmarked, and a cursor advanced past it would + // drop it from every later replay. + _advanceOrdersCursor(event); + } + + /// Events seen but deliberately left unmarked (no matching session yet, or + /// a failure mid-processing), keyed by id. They still need replaying, so + /// the shared node cursor must not move past the oldest of them. + final Map _retryableEvents = {}; + + /// How long an unmarked event holds the cursor back. Bounded so an event + /// that can never be processed (a trade key whose session is gone) cannot + /// freeze the cursor — and with it the replay window — forever. + static const _retryHoldWindow = Duration(hours: 1); + + /// Records an event that was not marked processed, so [_advanceOrdersCursor] + /// keeps the replay window covering it. + void _holdEventForRetry(NostrEvent event) { + if (event.kind != 14 || event.id == null || event.createdAt == null) return; + _pruneExpiredHolds(); + _retryableEvents[event.id!] = event.createdAt!; + } + + /// Drops holds older than [_retryHoldWindow]. Runs on both the hold and the + /// advance path: pruning only when an event is accepted would let the map + /// grow unpruned through a run in which every event is held. + void _pruneExpiredHolds() { + _retryableEvents.removeWhere( + (_, at) => at.isBefore(DateTime.now().subtract(_retryHoldWindow)), + ); + } + + /// Ids currently holding the cursor back. + @visibleForTesting + Set get debugHeldEventIds => _retryableEvents.keys.toSet(); + + /// Advances the orders `since` cursor for a processed event. + /// + /// Only kind 14 counts: the cursor feeds the NIP-44 filter, and gift wrap + /// (1059) timestamps are randomized, so letting them move it would push + /// `since` past kind-14 messages once the node switches transport. + /// + /// The cursor is a contiguous watermark: it never moves past an event still + /// awaiting a retry, otherwise one trade's newer response would evict + /// another trade's older, still-unprocessed one from the replay window. + void _advanceOrdersCursor(NostrEvent event) { + _retryableEvents.remove(event.id); + if (event.kind != 14) return; + final accepted = event.createdAt!; + _pruneExpiredHolds(); + final blocked = _retryableEvents.values.any((at) => !at.isAfter(accepted)); + if (blocked) return; + unawaited( + ref + .read(ordersCursorStoreProvider) + .advance(_settings.mostroPublicKey, accepted), + ); } Future _onData(NostrEvent event) async { @@ -149,6 +210,7 @@ class MostroService { // Deliberately NOT marked processed: the session may simply not exist // yet (startup ordering, a child order being linked), and a later // replay must be able to retry this event. + _holdEventForRetry(event); logger.w('No matching session found for recipient: ${event.recipient}'); return; } @@ -172,7 +234,10 @@ class MostroService { decryptedId = decryptedEvent.id; } - if (content == null) return; + if (content == null) { + _holdEventForRetry(event); + return; + } final result = jsonDecode(content); @@ -224,6 +289,7 @@ class MostroService { // transient failure. A permanently undecryptable event costs one // decrypt attempt per replay, which the dedup above bounds to one // relay copy at a time. + _holdEventForRetry(event); logger.e('Error processing event', error: e); } } diff --git a/test/features/disputes/dispute_chat_duplicate_envelope_test.dart b/test/features/disputes/dispute_chat_duplicate_envelope_test.dart index 230b9708..34f08be6 100644 --- a/test/features/disputes/dispute_chat_duplicate_envelope_test.dart +++ b/test/features/disputes/dispute_chat_duplicate_envelope_test.dart @@ -174,7 +174,13 @@ void main() { nostrService.controller.add(forged); nostrService.controller.add(real); - await pumpEventQueue(times: 200); + // chatUnwrap verifies and decrypts on a worker isolate, whose spawn takes + // real wall-clock time: draining the event queue alone can return before + // the valid envelope has been accepted. Poll until it lands instead. + await _waitFor( + () => container.read(disputeChatNotifierProvider(disputeId)).messages + .isNotEmpty, + ); final messages = container.read(disputeChatNotifierProvider(disputeId)).messages; @@ -185,3 +191,15 @@ void main() { expect(notifier.mounted, isTrue); }); } + +/// Polls [condition] until it holds or [timeout] elapses, yielding to the +/// event loop between attempts so isolate results can be delivered. +Future _waitFor( + bool Function() condition, { + Duration timeout = const Duration(seconds: 10), +}) async { + final deadline = DateTime.now().add(timeout); + while (!condition() && DateTime.now().isBefore(deadline)) { + await Future.delayed(const Duration(milliseconds: 10)); + } +} diff --git a/test/features/subscriptions/orders_filter_test.dart b/test/features/subscriptions/orders_filter_test.dart index 3d715eac..cffb82ce 100644 --- a/test/features/subscriptions/orders_filter_test.dart +++ b/test/features/subscriptions/orders_filter_test.dart @@ -31,5 +31,35 @@ void main() { expect(filter.authors, [mostroPubkey]); expect(filter.p, tradeKeys); }); + + // Without a since every (re)subscription replayed the node's full message + // history from every relay; kind 14 carries real timestamps, so a + // persisted cursor can bound the replay tightly. + test('v2 (nip44) bounds the replay with since and no limit', () { + final since = DateTime.fromMillisecondsSinceEpoch(1700000000 * 1000); + final filter = buildOrdersFilter( + Transport.nip44, + tradeKeys, + mostroPubkey, + since: since, + ); + + expect(filter.since, since); + // A limit on top of since would be answered with the *newest* n, + // silently dropping the rest of the window. Those events are never + // delivered, so they never hold the cursor back and are lost for good. + expect(filter.limit, isNull); + }); + + test('v1 (giftWrap) ignores since: its timestamps are randomized', () { + final filter = buildOrdersFilter( + Transport.giftWrap, + tradeKeys, + mostroPubkey, + since: DateTime.now(), + ); + + expect(filter.since, isNull); + }); }); } diff --git a/test/features/subscriptions/orders_since_bootstrap_test.dart b/test/features/subscriptions/orders_since_bootstrap_test.dart new file mode 100644 index 00000000..4f63dc9a --- /dev/null +++ b/test/features/subscriptions/orders_since_bootstrap_test.dart @@ -0,0 +1,144 @@ +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/features/settings/settings.dart'; +import 'package:mostro_mobile/features/settings/settings_notifier.dart'; +import 'package:mostro_mobile/features/settings/settings_provider.dart'; +import 'package:mostro_mobile/features/subscriptions/subscription_manager.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/nostr_service_provider.dart'; +import 'package:mostro_mobile/shared/providers/order_repository_provider.dart'; +import 'package:mostro_mobile/shared/providers/session_notifier_provider.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'; + +/// First launch after upgrading to the cursor build: sessions exist but no +/// `orders_since_` preference does. A flat default lookback would silently +/// drop responses to orders older than it — non-terminal orders are kept far +/// longer, and normal startup does not run the restore flow — so the +/// bootstrap window must also reach back to the oldest live session. +const _mostroPubkey = 'mostro-pubkey'; +const _tradeKey = + 'abcdef1234567890abcdef1234567890abcdef1234567890abcdef1234567890'; + +void main() { + late MockNostrService nostrService; + late MockOpenOrdersRepository orderRepository; + late ProviderContainer container; + late _FakeSessionNotifier sessions; + late SubscriptionManager manager; + late List issuedRequests; + late StreamController relayGenerations; + + Session sessionStartedAt(DateTime startTime) => Session( + masterKey: NostrKeyPairs( + private: + '1234567890abcdef1234567890abcdef1234567890abcdef1234567890abcdef'), + tradeKey: NostrKeyPairs(private: _tradeKey), + keyIndex: 0, + fullPrivacy: false, + startTime: startTime, + )..orderId = 'order-a'; + + setUp(() { + SharedPreferencesAsyncPlatform.instance = + InMemorySharedPreferencesAsync.empty(); + nostrService = MockNostrService(); + orderRepository = MockOpenOrdersRepository(); + issuedRequests = []; + var nextSubscriptionId = 0; + when(nostrService.subscribeToEvents(any)).thenAnswer((invocation) { + final request = invocation.positionalArguments.first as NostrRequest; + request.subscriptionId ??= 'sub-${nextSubscriptionId++}'; + issuedRequests.add(request); + return const Stream.empty(); + }); + when(nostrService.unsubscribe(any)).thenAnswer((_) async {}); + relayGenerations = StreamController.broadcast(); + when(nostrService.relayGenerationStream) + .thenAnswer((_) => relayGenerations.stream); + when(nostrService.relayGeneration).thenAnswer((_) => 0); + when(orderRepository.mostroInstanceStream) + .thenAnswer((_) => const Stream.empty()); + when(orderRepository.mostroInstance).thenReturn(null); + + container = ProviderContainer(overrides: [ + nostrServiceProvider.overrideWithValue(nostrService), + orderRepositoryProvider.overrideWithValue(orderRepository), + settingsProvider.overrideWith((ref) => _FixedSettingsNotifier()), + sessionNotifierProvider.overrideWith((ref) { + sessions = _FakeSessionNotifier(ref); + return sessions; + }), + ]); + }); + + tearDown(() { + manager.unsubscribeAll(); + container.dispose(); + relayGenerations.close(); + }); + + Future ordersSinceFor(DateTime sessionStart) async { + container.read(sessionNotifierProvider); + manager = container.read(subscriptionManagerProvider); + sessions.emit([sessionStartedAt(sessionStart)]); + await Future.delayed(const Duration(milliseconds: 50)); + + final ordersFilter = issuedRequests + .expand((r) => r.filters) + .firstWhere((f) => + (f.kinds ?? const []).contains(14) && + (f.authors ?? const []).contains(_mostroPubkey)); + expect(ordersFilter.since, isNotNull); + return ordersFilter.since!; + } + + test('with no cursor the window reaches back to the oldest session', + () async { + final start = DateTime.now().subtract(const Duration(days: 45)); + + final since = await ordersSinceFor(start); + + expect(since.isAfter(start), isFalse, + reason: 'a response to a 45-day-old order must still be replayed'); + }); + + test('a young session does not narrow the window below the lookback', + () async { + final since = await ordersSinceFor(DateTime.now()); + + final lookbackFloor = DateTime.now() + .subtract(NostrEventExtensions.chatDefaultLookback) + .subtract(const Duration(minutes: 1)); + expect(since.isAfter(lookbackFloor), isFalse, + reason: 'fresh installs keep the default lookback'); + }); +} + +class _FixedSettingsNotifier extends SettingsNotifier { + _FixedSettingsNotifier() : super(MockSharedPreferencesAsync()) { + state = Settings( + relays: const [], + fullPrivacyMode: false, + mostroPublicKey: _mostroPubkey, + ); + } +} + +class _FakeSessionNotifier extends SessionNotifier { + _FakeSessionNotifier(Ref ref) + : super(ref, MockSessionStorage(), MockSettings()) { + state = const []; + } + + void emit(List sessions) => state = sessions; +} diff --git a/test/services/mostro_service_orders_cursor_test.dart b/test/services/mostro_service_orders_cursor_test.dart new file mode 100644 index 00000000..b1600755 --- /dev/null +++ b/test/services/mostro_service_orders_cursor_test.dart @@ -0,0 +1,211 @@ +import 'dart:async'; +import 'dart:convert'; + +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/session.dart'; +import 'package:mostro_mobile/data/repositories/event_storage.dart'; +import 'package:mostro_mobile/features/settings/settings.dart'; +import 'package:mostro_mobile/features/settings/settings_notifier.dart'; +import 'package:mostro_mobile/features/settings/settings_provider.dart'; +import 'package:mostro_mobile/features/subscriptions/subscription_manager_provider.dart'; +import 'package:mostro_mobile/services/chat_cursor_store.dart'; +import 'package:mostro_mobile/services/mostro_service.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/nostr_utils.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 orders `since` cursor is shared by every session of one node, so it +/// must only ever move to a point past which nothing still needs replaying: +/// it feeds the kind-14 filter alone, and it must not step over an event +/// that was deliberately left unmarked for a later retry. +const _nodePriv = + '0000000000000000000000000000000000000000000000000000000000000003'; +const _tradePriv = + '0000000000000000000000000000000000000000000000000000000000000004'; + +void main() { + final nodeKeys = NostrKeyPairs(private: _nodePriv); + final tradeKeys = NostrKeyPairs(private: _tradePriv); + + late Database db; + late StreamController ordersController; + late MockSubscriptionManagerSpy subscriptionManager; + late _FakeSessionNotifier sessions; + late ProviderContainer container; + late ChatCursorStore cursorStore; + + /// A node message whose payload is an empty tuple: accepted and marked + /// processed by the shortest path through `_processEvent`. + Future nodeMessage({DateTime? createdAt}) async { + final encrypted = + await NostrUtils.encryptNIP44(jsonEncode([]), _nodePriv, tradeKeys.public); + return NostrEvent.fromPartialData( + kind: 14, + content: encrypted, + keyPairs: nodeKeys, + createdAt: createdAt, + tags: [ + ['p', tradeKeys.public], + ], + ); + } + + /// A kind-14 addressed to a trade key no session holds: left unmarked so a + /// later replay can retry it once the session exists. + NostrEvent orphanMessage({required DateTime createdAt}) => + NostrEvent.fromPartialData( + kind: 14, + content: 'undecryptable', + keyPairs: nodeKeys, + createdAt: createdAt, + tags: [ + ['p', 'a' * 64], + ], + ); + + setUp(() async { + SharedPreferencesAsyncPlatform.instance = + InMemorySharedPreferencesAsync.empty(); + db = await newDatabaseFactoryMemory().openDatabase('orders_cursor.db'); + ordersController = StreamController.broadcast(); + subscriptionManager = MockSubscriptionManagerSpy(); + when(subscriptionManager.orders) + .thenAnswer((_) => ordersController.stream); + + container = ProviderContainer(overrides: [ + sharedPreferencesProvider.overrideWithValue(SharedPreferencesAsync()), + eventStorageProvider.overrideWithValue(EventStorage(db: db)), + subscriptionManagerProvider.overrideWithValue(subscriptionManager), + settingsProvider + .overrideWith((ref) => _FixedSettingsNotifier(nodeKeys.public)), + sessionNotifierProvider.overrideWith((ref) { + sessions = _FakeSessionNotifier(ref); + return sessions; + }), + mostroServiceProvider.overrideWith((ref) => MostroService(ref)..init()), + ]); + container.read(sessionNotifierProvider); + container.read(mostroServiceProvider); + cursorStore = container.read(ordersCursorStoreProvider); + + sessions.emit([ + Session( + masterKey: tradeKeys, + tradeKey: tradeKeys, + keyIndex: 0, + fullPrivacy: false, + startTime: DateTime.now(), + )..orderId = 'order-a', + ]); + }); + + tearDown(() async { + await ordersController.close(); + container.dispose(); + await db.close(); + }); + + MostroService service() => container.read(mostroServiceProvider); + + /// Delivers [event] and waits for the pipeline to actually settle, rather + /// than for a fixed delay: processing runs an off-isolate NIP-44 decrypt, a + /// Sembast write and a SharedPreferences write, none of which this test can + /// await directly. A held event settles once it holds the cursor back; a + /// processed one settles once its durable marker is written (which is what + /// _markEventProcessed writes before advancing the cursor). + Future deliver(NostrEvent event, {bool held = false}) async { + ordersController.add(event); + final eventStore = EventStorage(db: db); + await _waitFor(() async => held + ? service().debugHeldEventIds.contains(event.id) + : await eventStore.hasItem(event.id!)); + } + + test('a processed kind-14 event advances the orders cursor', () async { + final event = await nodeMessage(); + + await deliver(event); + + // The advance itself is fire-and-forget after the marker is written. + await _waitFor( + () async => await cursorStore.cursorFor(nodeKeys.public) != null); + final cursor = await cursorStore.cursorFor(nodeKeys.public); + expect(cursor, isNotNull); + expect(cursor!.millisecondsSinceEpoch ~/ 1000, + event.createdAt!.millisecondsSinceEpoch ~/ 1000); + }); + + test('a processed gift wrap does not advance the kind-14 cursor', () async { + // Gift wrap timestamps are randomized +/- 48 h; letting one move this + // cursor would push `since` past real kind-14 messages. + final wrap = await NostrUtils.createNIP59Event( + jsonEncode([]), + tradeKeys.public, + _nodePriv, + ); + + await deliver(wrap); + + expect(await cursorStore.cursorFor(nodeKeys.public), isNull, + reason: 'only the kind-14 transport owns this cursor'); + }); + + test('an older unmarked event holds the cursor back', () async { + final now = DateTime.now(); + // Arrives for a trade key whose session does not exist yet: not marked, + // so a later replay must still cover it. + await deliver( + orphanMessage(createdAt: now.subtract(const Duration(minutes: 30))), + held: true, + ); + + // A newer message for another session must not evict it from the window. + await deliver(await nodeMessage(createdAt: now)); + + expect(await cursorStore.cursorFor(nodeKeys.public), isNull, + reason: 'the cursor is a contiguous watermark, not a high-water mark'); + }); +} + +/// Polls [condition] until it holds or [timeout] elapses, yielding to the +/// event loop between attempts. +Future _waitFor( + Future Function() condition, { + Duration timeout = const Duration(seconds: 10), +}) async { + final deadline = DateTime.now().add(timeout); + while (!await condition() && DateTime.now().isBefore(deadline)) { + await Future.delayed(const Duration(milliseconds: 10)); + } +} + +class _FixedSettingsNotifier extends SettingsNotifier { + _FixedSettingsNotifier(String mostroPublicKey) + : super(MockSharedPreferencesAsync()) { + state = Settings( + relays: const [], + fullPrivacyMode: false, + mostroPublicKey: mostroPublicKey, + ); + } +} + +class _FakeSessionNotifier extends SessionNotifier { + _FakeSessionNotifier(Ref ref) + : super(ref, MockSessionStorage(), MockSettings()) { + state = const []; + } + + void emit(List sessions) => state = sessions; +}