diff --git a/docs/architecture/SESSION_RECOVERY_ARCHITECTURE.md b/docs/architecture/SESSION_RECOVERY_ARCHITECTURE.md index 4895c91f1..72faba795 100644 --- a/docs/architecture/SESSION_RECOVERY_ARCHITECTURE.md +++ b/docs/architecture/SESSION_RECOVERY_ARCHITECTURE.md @@ -353,12 +353,14 @@ for (final orderDetail in ordersResponse.orders) { // ... other fields ); - // Build synthetic MostroMessage + // Build synthetic MostroMessage. The ordering timestamp is the single + // restore-start anchor (_restoreStartTime), NOT orderDetail.createdAt — + // see "Restore-Time Live Event Buffering and Replay" below for why. final mostroMessage = MostroMessage( id: orderDetail.id, action: action, payload: order, - timestamp: orderDetail.createdAt ?? DateTime.now().millisecondsSinceEpoch, + timestamp: _restoreStartTime, ); // Save message to storage and update state @@ -511,37 +513,135 @@ Action _getActionFromStatus(Status status, Role? userRole) { ### Restore Mode Protection -**File**: `lib/features/restore/restore_manager.dart:466-468` +**File**: `lib/features/restore/restore_manager.dart` -During recovery, a global flag prevents processing of old messages: +During recovery, a global flag (`isRestoringProvider`) marks the window in +which historical order/dispute state is being rebuilt from Mostro's restore +response: ```dart -// Enable restore mode to block all old message processing +// Enable restore mode to block synthetic/live processing races ref.read(isRestoringProvider.notifier).state = true; -_logger.i('Restore: enabled restore mode - blocking all old message processing'); +logger.i('Restore: enabled restore mode - blocking all old message processing'); ``` -**File**: `lib/services/mostro_service.dart:44-96` +`isRestoringProvider` is cleared on **both** the success path and the catch +block of `restore()`, so the transition back to `false` is outcome-agnostic +— see D3 below. + +**File**: `lib/services/mostro_service.dart` ```dart bool _isRestorePayload(Map json) { // Check if this is a restore-specific payload that should be ignored // during normal operation - + final wrapper = json['restore'] ?? json['order']; if (wrapper == null || wrapper is! Map) return false; - + final payload = wrapper['payload']; if (payload == null || payload is! Map) return false; - + // Check for restore-specific fields if (payload.containsKey('restore_data')) return true; if (payload.containsKey('trade_index')) return true; - + return false; } ``` +`_isRestorePayload` only recognizes the restore protocol's OWN response +payloads (restore data, orders list, trade index) arriving on the temporary +subscription — it has nothing to do with regular live order/dispute events +addressed to the user's real sessions. Those are handled by the +buffer-and-replay mechanism below. + +### Restore-Time Live Event Buffering and Replay + +**File**: `lib/services/mostro_service.dart` + +Fixes GitHub #584: previously, any live event (order update, dispute +message) that arrived on the user's real sessions while a restore was in +progress was either dropped outright or — in an earlier, buggy port of this +fix — buffered too late (after the session-match check), so events for a +session restore had not recreated yet were still lost. Live events are now +buffered unconditionally while `isRestoringProvider` is true and replayed, +in arrival order, once restore ends. + +**Architecture decisions**: + +- **D1 — Buffer check precedes session-match**: `_onData` checks + `isRestoringProvider` immediately after the dedup reserve, **before** the + `matchingSession == null` early return and before any decrypt. Restore + recreates sessions incrementally (one `saveSession()` call per order), so + an event addressed to a session that has not been recreated yet must still + be preserved, not dropped as "no matching session". +- **D2 — Keyed buffer**: `_restoreBuffer` is a `Map` + keyed by `event.id`, which gives dedup-by-id (re-buffering the same id + keeps a single entry in its original arrival position — Dart `Map` is + insertion-ordered) as a defensive backstop behind the top-level + `eventStore.hasItem`/`putItem` dedup check. +- **D3 — Outcome-agnostic flush trigger**: `MostroService.init()` registers + `ref.listen(isRestoringProvider, ...)` and flushes the buffer on any + `true → false` transition. This covers `RestoreService.restore()`'s + success path and its catch block identically — the flush does not need to + know why restore ended. +- **D4 — Dedup entry cleared before replay**: `_flushRestoreBuffer()` calls + `eventStore.deleteItem(event.id)` immediately before replaying each event + through `_onData`, because `_onData` reserved that id when the event was + first buffered. Without this, the dedup check at the top of `_onData` + would silently drop the replay. +- **D5 — Single restore-start anchor for synthetic messages**: synthetic + order/dispute messages built in `RestoreService.restore()` use one + timestamp, `_restoreStartTime` — captured once at the start of + `initRestoreProcess()`, before restore mode is enabled — instead of + `orderDetail.createdAt`. `createdAt` reflects when the order was + originally created, not when this snapshot was taken; for a long-lived + order, using it as the ordering timestamp would let intervening historical + replay outrank the current-state snapshot. The restore-start anchor sorts + newer than all pre-restore history yet older than any live event that + arrives during restore. The real creation time is still preserved in the + `Order`/`Dispute` payload for display; only the ordering timestamp + changes. Real Nostr timestamps are always second-precision, never exact + to the millisecond, so `_restoreStartTime` is floored to just before the + current second (`_floorToPreviousSecond`) rather than used raw — otherwise + a live event created in the same second could sort older than the anchor + despite happening after it. +- **D6 — Deferred fiat-sent classification, reconciled once after flush**: + `RestoreService.restore()` no longer checks `storage.getAllMessagesForOrderId()` + inline while building disputed/cooperativelyCanceled snapshots — the + confirming `fiatSent`/`fiatSentOk` message may still be sitting unflushed + in `MostroService._restoreBuffer` at that point. Affected orders are + tracked and rechecked once, in a `finally` block, after awaiting the new + public `MostroService.flushRestoreBuffer()` (single-flight guarded against + the existing reactive listener). The tracked snapshot is only replayed if + no stored message for that order is newer, so a buffered live update + flushed on the same pass is never overwritten by the stale snapshot. + +**Target `_onData` control flow** (landmark-relative — decrypt, session-match, +DM/restore-payload skips, and the timestamp fallback already existed; only +the buffer check's *position*, relative to session-match, is new): + +```dart +_onData(event): + 1. if eventStore.hasItem(id): return // dedup check + 2. eventStore.putItem(id, ...) // dedup reserve + 3. if isRestoringProvider: buffer[id] = event; return // <== reorder fix + 4. matchingSession = ...; if null: return // now AFTER buffer + 5. decrypt (v1 gift-wrap unWrap / v2 NIP-44 direct) + 6. jsonDecode; skip DM payloads; skip restore payloads + 7. msg = MostroMessage.fromJson(...) + 8. msg.timestamp ??= innerRumorCreatedAt ?? event.createdAt + 9. messageStorage.addMessage(...); link child order if applicable +``` + +On flush, `isRestoringProvider` is `false`, so replayed events flow past +step 3 straight to step 4 onward and receive their real protocol timestamp: +the inner rumor's `created_at` for v1 gift wrap (canonical per NIP-59 — the +outer wrap/seal timestamps are randomized for privacy), or the event's own +`created_at` for v2 NIP-44 direct (kind 14 has no seal/rumor layer, so its +own timestamp is already the real send time). + ### Session Validation The system validates that recreated sessions match the expected order data: @@ -718,8 +818,8 @@ sequenceDiagram --- -**Last Modified**: November 25, 2025 -**Version**: 1.0.0 +**Last Modified**: July 8, 2026 +**Version**: 1.1.0 **Author**: Architecture Documentation **Related Files**: - `lib/features/restore/restore_manager.dart` diff --git a/integration_test/test_helpers.dart b/integration_test/test_helpers.dart index ddabe916d..bec2f9aa3 100644 --- a/integration_test/test_helpers.dart +++ b/integration_test/test_helpers.dart @@ -315,11 +315,27 @@ class FakeMostroService implements MostroService { @override void updateSettings(Settings settings) {} - + @override void dispose() { // TODO: implement dispose } + + @override + Future onDataForTesting( + NostrEvent event, { + int? bufferedReceivedAt, + }) async {} + + @override + Future flushRestoreBuffer() async {} + + @override + Future flushRestoreBufferForTesting() async {} + + @override + Map + get restoreBufferForTesting => {}; } Future pumpTestApp(WidgetTester tester) async { diff --git a/lib/data/models/mostro_message.dart b/lib/data/models/mostro_message.dart index 7d8a93232..79691af0f 100644 --- a/lib/data/models/mostro_message.dart +++ b/lib/data/models/mostro_message.dart @@ -15,6 +15,7 @@ class MostroMessage { int? tradeIndex; T? _payload; int? timestamp; + int? receivedAt; MostroMessage({ required this.action, @@ -23,6 +24,7 @@ class MostroMessage { T? payload, this.tradeIndex, this.timestamp, + this.receivedAt, }) : _payload = payload; Map toJson({int? version}) { @@ -46,6 +48,11 @@ class MostroMessage { factory MostroMessage.fromJson(Map json) { final timestamp = json['timestamp']; + // receivedAt is only present when reconstructing from a stored Sembast + // record (injected by MostroStorage.addMessage); live wire payloads and + // restore's synthetic messages never carry this key, so it parses as + // null for those. + final receivedAt = json['receivedAt']; // IMPORTANT : Use 'order', 'restore' or 'cant-do' key as per protocol json = json['order'] ?? json['restore'] ?? json['cant-do'] ?? json; final num requestId = json['request_id'] ?? 0; @@ -59,6 +66,7 @@ class MostroMessage { ? Payload.fromJson(json['payload']) as T? : null, timestamp: timestamp, + receivedAt: receivedAt, ); } diff --git a/lib/data/repositories/mostro_storage.dart b/lib/data/repositories/mostro_storage.dart index 659180dad..85752fe23 100644 --- a/lib/data/repositories/mostro_storage.dart +++ b/lib/data/repositories/mostro_storage.dart @@ -14,16 +14,26 @@ class MostroStorage extends BaseStorage { Future addMessage(String key, MostroMessage message) async { final id = key; try { - if (await hasItem(id)) return; - // Add metadata for easier querying - final Map dbMap = message.toJson(); - message.timestamp ??= DateTime.now().millisecondsSinceEpoch; - dbMap['timestamp'] = message.timestamp; - - await store.record(id).put(db, dbMap); - logger.i( - 'Saved message of type ${message.action} with order id ${message.id}', - ); + // The existence check and the write happen inside the same + // transaction so a concurrent addMessage for the same key (e.g. the + // same event redelivered by a second relay) can't slip past the + // check before the first call finishes writing. + final wrote = await db.transaction((txn) async { + if (await store.record(id).exists(txn)) return false; + // Add metadata for easier querying + final Map dbMap = message.toJson(); + message.timestamp ??= DateTime.now().millisecondsSinceEpoch; + dbMap['timestamp'] = message.timestamp; + dbMap['receivedAt'] = message.receivedAt; + + await store.record(id).put(txn, dbMap); + return true; + }); + if (wrote) { + logger.i( + 'Saved message of type ${message.action} with order id ${message.id}', + ); + } } catch (e, stack) { logger.e( 'addMessage failed for $id', diff --git a/lib/features/order/models/order_state.dart b/lib/features/order/models/order_state.dart index 4f27f8f65..4686cba00 100644 --- a/lib/features/order/models/order_state.dart +++ b/lib/features/order/models/order_state.dart @@ -158,20 +158,16 @@ class OrderState { // Handle dispute status updates based on action Dispute? updatedDispute = message.getPayload() ?? dispute; - // If we got a dispute from the message payload, ensure it has the message timestamp - // This is critical for correct sorting in the dispute list if (updatedDispute != null && message.getPayload() != null) { - // Use message timestamp if dispute doesn't have a createdAt or if message has a timestamp - // Note: Nostr timestamps are in seconds, so convert to milliseconds - if (message.timestamp != null) { - final tsMs = message.timestamp! * 1000; - if (updatedDispute.createdAt == null || - updatedDispute.createdAt!.millisecondsSinceEpoch != tsMs) { - updatedDispute = updatedDispute.copyWith( - createdAt: DateTime.fromMillisecondsSinceEpoch(tsMs), - ); - logger.i('Updated dispute ${updatedDispute.disputeId} createdAt from message timestamp: ${updatedDispute.createdAt}'); - } + // Only fill createdAt when the dispute doesn't already carry one — a + // dispute's creation time is fixed and must never be re-stamped by a + // later message's timestamp (which, during restore, is an ordering + // anchor, not a real creation time). + if (updatedDispute.createdAt == null && message.timestamp != null) { + updatedDispute = updatedDispute.copyWith( + createdAt: DateTime.fromMillisecondsSinceEpoch(message.timestamp!), + ); + logger.i('Set dispute ${updatedDispute.disputeId} createdAt from message timestamp: ${updatedDispute.createdAt}'); } } diff --git a/lib/features/order/notifiers/abstract_mostro_notifier.dart b/lib/features/order/notifiers/abstract_mostro_notifier.dart index 6be5d051b..12ff73ec8 100644 --- a/lib/features/order/notifiers/abstract_mostro_notifier.dart +++ b/lib/features/order/notifiers/abstract_mostro_notifier.dart @@ -110,19 +110,19 @@ class AbstractMostroNotifier extends StateNotifier { if (mounted) { state = state.updateWith(msg); } - if (msg.timestamp != null && - msg.timestamp! > + if (msg.receivedAt != null && + msg.receivedAt! > DateTime.now() .subtract(const Duration(seconds: 60)) .millisecondsSinceEpoch) { logger.i( - 'Message timestamp check passed, calling handleEvent for ${msg.action}'); + 'Notification eligible for ${msg.action}: live event, calling handleEvent'); unawaited(handleEvent(msg, previousStatus: previousStatus, wasUserInitiatedCancel: wasUserInitiatedCancel)); } else { - logger.w( - 'Message timestamp check failed for ${msg.action}. Timestamp: ${msg.timestamp}, Current: ${DateTime.now().millisecondsSinceEpoch}, Threshold: ${DateTime.now().subtract(const Duration(seconds: 60)).millisecondsSinceEpoch}'); + logger.i( + 'Notification skipped for ${msg.action}: not a live event (receivedAt: ${msg.receivedAt})'); // Handle dispute actions even if timestamp is old, since they're critical for UI state // but bypass navigation/notification side effects @@ -180,8 +180,8 @@ class AbstractMostroNotifier extends StateNotifier { final navProvider = ref.read(navigationProvider.notifier); // Check if this is a recent event for notification/navigation purposes - final isRecent = event.timestamp != null && - event.timestamp! > + final isRecent = event.receivedAt != null && + event.receivedAt! > DateTime.now() .subtract(const Duration(seconds: 60)) .millisecondsSinceEpoch; @@ -203,7 +203,7 @@ class AbstractMostroNotifier extends StateNotifier { ); } else if (notificationData != null && bypassTimestampGate) { logger.i( - 'Skipping notification for old event: ${event.action} (timestamp: ${event.timestamp})'); + 'Skipping notification for old event: ${event.action} (receivedAt: ${event.receivedAt})'); } /// Handle incoming events and update state accordingly diff --git a/lib/features/order/notifiers/order_notifier.dart b/lib/features/order/notifiers/order_notifier.dart index e30465330..65f916106 100644 --- a/lib/features/order/notifiers/order_notifier.dart +++ b/lib/features/order/notifiers/order_notifier.dart @@ -230,6 +230,18 @@ class OrderNotifier extends AbstractMostroNotifier { state = state.copyWith(fiatWasSent: true); } + // Corrects a stale no-fiat cooperative-cancel action after the fact, + // regardless of what set it (a restore snapshot or a live event that + // ran before setFiatWasSent()). + void upgradeCooperativeCancelToFiatSent() { + if (!mounted) return; + if (state.action == Action.cooperativeCancelNoFiatByYou) { + state = state.copyWith(action: Action.cooperativeCancelFiatSentByYou); + } else if (state.action == Action.cooperativeCancelNoFiatByPeer) { + state = state.copyWith(action: Action.cooperativeCancelFiatSentByPeer); + } + } + /// Update dispute in state (used during restore) void updateDispute(Dispute dispute) { if (mounted) { diff --git a/lib/features/restore/restore_manager.dart b/lib/features/restore/restore_manager.dart index 4e8d1230b..26acc6bae 100644 --- a/lib/features/restore/restore_manager.dart +++ b/lib/features/restore/restore_manager.dart @@ -62,6 +62,24 @@ class RestoreService { bool _operationInProgress = false; Completer? _operationCompleter; + // Single anchor timestamp for every synthetic message built by this restore + // run, captured once at the start of initRestoreProcess() (before restore + // mode is enabled). Ensures synthetic snapshots sort newer than pre-restore + // history yet older than any live event arriving during restore, regardless + // of the order's own (possibly stale) createdAt. Defaults to construction + // time as a safe fallback in case restore() is ever invoked directly. + int _restoreStartTime = + _floorToPreviousSecond(DateTime.now().millisecondsSinceEpoch); + + // Nostr timestamps are always whole seconds, never millisecond-precise; + // flooring the anchor to just before the current second guarantees it + // can never outrank a live event created in this same second. + static int _floorToPreviousSecond(int ms) => (ms ~/ 1000) * 1000 - 1; + + @visibleForTesting + static int floorToPreviousSecondForTesting(int ms) => + _floorToPreviousSecond(ms); + RestoreService(this.ref); Future importMnemonicAndRestore(String mnemonic) async { @@ -593,6 +611,9 @@ class RestoreService { OrdersResponse ordersResponse, List disputes, ) async { + // Orders needing a fiat-sent recheck once the restore buffer drains. + final pendingReconciliation = []; + try { if (_masterKey == null) { throw Exception('Master key not initialized'); @@ -779,82 +800,56 @@ class RestoreService { ); if (dispute != null) { - // For disputed orders, check if fiat was sent before the dispute - // so future cooperative cancel actions get remapped correctly - final disputeMessages = await storage.getAllMessagesForOrderId( - orderDetail.id, - ); - final hadFiatSent = disputeMessages.any( - (m) => - m.action == Action.fiatSent || - m.action == Action.fiatSentOk, - ); - if (hadFiatSent) { - notifier.setFiatWasSent(); - logger.i( - 'Restore: fiatWasSent=true for disputed order ${orderDetail.id}', - ); - } + // Fiat-sent status is checked later, once the buffer has flushed. // Create dispute message with Dispute payload (per Mostro protocol) + // Timestamp is the restore-start anchor (not orderDetail.createdAt, + // which reflects the order's original creation time, not this + // snapshot's currency) so it outranks pre-restore history while + // still sorting behind any live event arriving during restore. final disputeMessage = MostroMessage( id: orderDetail.id, action: action, payload: dispute, - timestamp: - orderDetail.createdAt ?? - DateTime.now().millisecondsSinceEpoch, + timestamp: _restoreStartTime, ); // Save dispute message to storage final disputeKey = - '${orderDetail.id}_restore_${action.value}_${DateTime.now().millisecondsSinceEpoch}'; + '${orderDetail.id}_restore_${action.value}_$_restoreStartTime'; await storage.addMessage(disputeKey, disputeMessage); // Update state with dispute message notifier.updateStateFromMessage(disputeMessage); + pendingReconciliation.add(orderDetail.id); logger.i( 'Restore: created dispute message for order ${orderDetail.id}', ); } else { - // For cooperativelyCanceled orders, check message history to - // determine if fiat was sent before the cancel was initiated. - // This sets fiatWasSent so updateWith can remap to the correct - // semantic action variant. - if (order.status == Status.cooperativelyCanceled) { - final messages = await storage.getAllMessagesForOrderId( - orderDetail.id, - ); - final hadFiatSent = messages.any( - (m) => - m.action == Action.fiatSent || - m.action == Action.fiatSentOk, - ); - if (hadFiatSent) { - notifier.setFiatWasSent(); - logger.i( - 'Restore: fiatWasSent=true for cooperativelyCanceled order ${orderDetail.id}', - ); - } - } + // Fiat-sent status is checked later, once the buffer has flushed. // Create regular order message with Order payload + // Timestamp is the restore-start anchor (not orderDetail.createdAt, + // which reflects the order's original creation time, not this + // snapshot's currency) so it outranks pre-restore history while + // still sorting behind any live event arriving during restore. final mostroMessage = MostroMessage( id: orderDetail.id, action: action, payload: order, - timestamp: - orderDetail.createdAt ?? - DateTime.now().millisecondsSinceEpoch, + timestamp: _restoreStartTime, ); // Save order message to storage final key = - '${orderDetail.id}_restore_${action.value}_${DateTime.now().millisecondsSinceEpoch}'; + '${orderDetail.id}_restore_${action.value}_$_restoreStartTime'; await storage.addMessage(key, mostroMessage); // Update state with order message notifier.updateStateFromMessage(mostroMessage); + if (order.status == Status.cooperativelyCanceled) { + pendingReconciliation.add(orderDetail.id); + } } } catch (e, stack) { logger.e( @@ -877,9 +872,59 @@ class RestoreService { ref.read(isRestoringProvider.notifier).state = false; logger.e('Restore: error during restore', error: e, stackTrace: stack); rethrow; + } finally { + // Restore mode is already off here, so the flush's replayed events + // reach storage instead of re-buffering. Never let this throw — it + // would mask a real restore error being rethrown above. + try { + await ref.read(mostroServiceProvider).flushRestoreBuffer(); + await _reconcileFiatSent(pendingReconciliation); + } catch (e, stack) { + logger.e( + 'Restore: post-restore flush/reconciliation failed', + error: e, + stackTrace: stack, + ); + } + } + } + + // Rechecks fiat-sent status for orders deferred during the restore loop, + // now that the buffer has drained. Per-order failures are isolated, same + // as MostroService._flushRestoreBuffer. + Future _reconcileFiatSent(List pending) async { + if (pending.isEmpty) return; + final storage = ref.read(mostroStorageProvider); + logger.i('Restore: reconciling fiat-sent status for ${pending.length} orders'); + for (final orderId in pending) { + try { + final messages = await storage.getAllMessagesForOrderId(orderId); + final hadFiatSent = messages.any( + (m) => + m.action == Action.fiatSent || m.action == Action.fiatSentOk, + ); + if (!hadFiatSent) continue; + + final notifier = ref.read(orderNotifierProvider(orderId).notifier); + notifier.setFiatWasSent(); + notifier.upgradeCooperativeCancelToFiatSent(); + logger.i( + 'Restore: reconciled fiatWasSent=true for order $orderId', + ); + } catch (e, stack) { + logger.e( + 'Restore: fiat reconciliation failed for order $orderId', + error: e, + stackTrace: stack, + ); + } } } + @visibleForTesting + Future reconcileFiatSentForTesting(List pending) => + _reconcileFiatSent(pending); + //Workflow: // 1. Clear existing data // 2. Create temporary subscription to key index 1 for restore notifications @@ -898,6 +943,10 @@ class RestoreService { _operationInProgress = true; _operationCompleter = Completer(); + // Snapshot the restore-start anchor before anything else, so every + // synthetic message this run produces shares one ordering timestamp. + _restoreStartTime = + _floorToPreviousSecond(DateTime.now().millisecondsSinceEpoch); // Hold the shared session lock for the whole restore so order/take flows // cannot interleave with the session reset (and rebuild) below. final releaseSessionLock = diff --git a/lib/services/mostro_service.dart b/lib/services/mostro_service.dart index 7be83dbcd..e82794a2f 100644 --- a/lib/services/mostro_service.dart +++ b/lib/services/mostro_service.dart @@ -2,11 +2,13 @@ import 'dart:async'; 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/logger_service.dart'; import 'package:mostro_mobile/data/enums.dart'; import 'package:mostro_mobile/data/models.dart'; +import 'package:mostro_mobile/features/restore/restore_mode_provider.dart'; import 'package:mostro_mobile/features/settings/settings.dart'; import 'package:mostro_mobile/features/subscriptions/subscription_manager_provider.dart'; import 'package:mostro_mobile/shared/providers.dart'; @@ -21,12 +23,28 @@ class MostroService { Settings _settings; StreamSubscription? _ordersSubscription; + ProviderSubscription? _restoreListener; + + // Live events received while a restore is in progress are buffered here + // (keyed by event id, so re-delivery of the same id keeps a single entry + // in its original arrival position) and replayed once restore ends. + // `receivedAtMs` is the TRUE original buffering time: it is preserved + // (never re-stamped) if the same event id is buffered again, so the + // eventual replay stamps the message with when it was first seen, not + // when the buffer happens to flush. + final Map _restoreBuffer = + {}; + + // Shared by the reactive listener and flushRestoreBuffer() so both await + // the same drain instead of racing. + Future? _activeFlush; MostroService(this.ref) : _settings = ref.read(settingsProvider); void init() { // Cancel any existing subscription to prevent leaks on re-init _ordersSubscription?.cancel(); + _restoreListener?.close(); // Subscribe to the orders stream from SubscriptionManager // The SubscriptionManager will automatically manage subscriptions based on SessionNotifier changes @@ -44,10 +62,20 @@ class MostroService { }, cancelOnError: false, ); + + // Flush buffered live events once restore ends, regardless of outcome + // (RestoreService.restore() clears isRestoringProvider on both its + // success path and its catch block). + _restoreListener = ref.listen(isRestoringProvider, (previous, next) { + if (previous == true && next == false) { + unawaited(flushRestoreBuffer()); + } + }); } void dispose() { _ordersSubscription?.cancel(); + _restoreListener?.close(); logger.i('MostroService disposed'); } @@ -105,7 +133,7 @@ class MostroService { return false; } - Future _onData(NostrEvent event) async { + Future _onData(NostrEvent event, {int? bufferedReceivedAt}) async { final eventStore = ref.read(eventStorageProvider); if (await eventStore.hasItem(event.id!)) return; @@ -116,12 +144,41 @@ class MostroService { 'created_at': event.createdAt!.millisecondsSinceEpoch ~/ 1000, }); + // Buffer live events while a restore is in progress. This runs BEFORE the + // session-match check below so events addressed to a session restore has + // not recreated yet are preserved instead of being dropped by "no + // matching session". Buffered events are replayed through _onData once + // restore completes (success or error path alike); see _flushRestoreBuffer. + if (ref.read(isRestoringProvider)) { + // Preserve-if-present: only a genuinely first-time buffering of this + // event id stamps `now()`. Any redelivery (the dedup reservation is + // released right below, specifically to allow redelivery) must keep + // the original buffering instant. + final existing = _restoreBuffer[event.id!]; + _restoreBuffer[event.id!] = ( + event: event, + receivedAtMs: + existing?.receivedAtMs ?? DateTime.now().millisecondsSinceEpoch, + ); + // Undo the dedup reservation: only the in-memory buffer holds this + // event now, and it does not survive process death. Releasing the + // reservation lets a later redelivery re-enter and flush normally + // instead of being dropped forever as "already seen". + await eventStore.deleteItem(event.id!); + logger.i('Restore: buffered live event ${event.id}'); + return; + } + final sessions = ref.read(sessionNotifierProvider); final matchingSession = sessions.firstWhereOrNull( (s) => s.tradeKey.public == event.recipient, ); if (matchingSession == null) { logger.w('No matching session found for recipient: ${event.recipient}'); + // Undo the dedup reservation: this event was never actually + // processed, so a later retry (once the session exists) must not + // be silently dropped as "already seen". + await eventStore.deleteItem(event.id!); return; } final privateKey = matchingSession.tradeKey.private; @@ -132,6 +189,9 @@ class MostroService { // decrypts straight to the tuple. Both converge on jsonDecode below. String? content; String? decryptedId; + // Inner rumor's created_at is the real send time (outer gift wrap is + // NIP-59 randomized for privacy); used for timestamp anchoring below. + DateTime? innerCreatedAt; if (event.kind == 14) { content = await NostrUtils.decryptNIP44DirectEvent( event, @@ -142,6 +202,7 @@ class MostroService { final decryptedEvent = await event.unWrap(privateKey); content = decryptedEvent.content; decryptedId = decryptedEvent.id; + innerCreatedAt = decryptedEvent.createdAt; } if (content == null) return; @@ -169,6 +230,22 @@ class MostroService { final msg = MostroMessage.fromJson(result[0]); + // For v1 gift-wrap (kind 1059) use the inner rumor's created_at (real + // send time; the outer wrap is NIP-59 randomized). For v2 NIP-44 direct + // (kind 14) innerCreatedAt is null (no rumor layer), so fall back to + // event.createdAt, which is already the real send time. + msg.timestamp ??= + innerCreatedAt?.millisecondsSinceEpoch ?? + event.createdAt?.millisecondsSinceEpoch; + + // True client arrival time: for a live event this is "now"; for a + // replayed restore-buffered event it is the original buffering + // instant, threaded back in by _flushRestoreBuffer. Always an + // explicit overwrite (never `??=`) so a peer-supplied value parsed by + // MostroMessage.fromJson can never survive — see INVARIANT R3-005. + msg.receivedAt = + bufferedReceivedAt ?? DateTime.now().millisecondsSinceEpoch; + final messageStorage = ref.read(mostroStorageProvider); // Use the inner rumor id if available (v1), otherwise fall back to the @@ -188,6 +265,49 @@ class MostroService { } } + /// Replays every event buffered during restore through [_onData], in + /// arrival order, once restore has ended. Clears each event's dedup entry + /// immediately before replaying it, since [_onData] reserved that entry + /// when the event was first buffered. + Future _flushRestoreBuffer() async { + if (_restoreBuffer.isEmpty) return; + final entries = + List<({NostrEvent event, int receivedAtMs})>.from( + _restoreBuffer.values, + ); + _restoreBuffer.clear(); + logger.i('Restore: flushing ${entries.length} buffered live events'); + final eventStore = ref.read(eventStorageProvider); + for (final entry in entries) { + try { + await eventStore.deleteItem(entry.event.id!); + await _onData(entry.event, bufferedReceivedAt: entry.receivedAtMs); + } catch (e) { + logger.e( + 'Restore: failed to replay buffered event ${entry.event.id}', + error: e, + ); + } + } + } + + // Awaitable flush for production callers. Idempotent when the buffer is empty. + Future flushRestoreBuffer() => + _activeFlush ??= _flushRestoreBuffer().whenComplete(() { + _activeFlush = null; + }); + + @visibleForTesting + Future onDataForTesting(NostrEvent event, {int? bufferedReceivedAt}) => + _onData(event, bufferedReceivedAt: bufferedReceivedAt); + + @visibleForTesting + Future flushRestoreBufferForTesting() => _flushRestoreBuffer(); + + @visibleForTesting + Map + get restoreBufferForTesting => _restoreBuffer; + Future _maybeLinkChildOrder( MostroMessage message, Session session, diff --git a/test/data/repositories/mostro_storage_test.dart b/test/data/repositories/mostro_storage_test.dart new file mode 100644 index 000000000..d69dd1bc6 --- /dev/null +++ b/test/data/repositories/mostro_storage_test.dart @@ -0,0 +1,65 @@ +import 'package:flutter_test/flutter_test.dart'; +import 'package:sembast/sembast_memory.dart'; +import 'package:mostro_mobile/data/models/enums/action.dart'; +import 'package:mostro_mobile/data/models/mostro_message.dart'; +import 'package:mostro_mobile/data/repositories/mostro_storage.dart'; + +void main() { + late MostroStorage storage; + + setUp(() async { + final db = await databaseFactoryMemory.openDatabase('mostro_storage_test.db'); + storage = MostroStorage(db: db); + }); + + group('MostroStorage.addMessage receivedAt persistence', () { + test('write-once: a second addMessage call for the same key does not ' + 'overwrite the original receivedAt', () async { + final first = MostroMessage( + action: Action.fiatSent, + id: 'order-2', + receivedAt: 1000, + ); + final second = MostroMessage( + action: Action.fiatSent, + id: 'order-2', + receivedAt: 9999, + ); + + await storage.addMessage('key-2', first); + await storage.addMessage('key-2', second); + + final reloaded = await storage.getLatestMessageById('order-2'); + + expect(reloaded, isNotNull); + expect(reloaded!.receivedAt, 1000); + }); + + test( + 'concurrent addMessage calls for the same key: only the first-started ' + 'write is retained, the second is a no-op', () async { + final first = MostroMessage( + action: Action.fiatSent, + id: 'order-3', + receivedAt: 1111, + ); + final second = MostroMessage( + action: Action.fiatSent, + id: 'order-3', + receivedAt: 2222, + ); + + // Neither call is awaited before the other starts, so their + // exists-check-then-write sequences genuinely overlap; the + // transaction wrapping addMessage is what keeps this deterministic. + final firstCall = storage.addMessage('key-3', first); + final secondCall = storage.addMessage('key-3', second); + await Future.wait([firstCall, secondCall]); + + final reloaded = await storage.getLatestMessageById('order-3'); + + expect(reloaded, isNotNull); + expect(reloaded!.receivedAt, 1111); + }); + }); +} diff --git a/test/features/order/notifiers/abstract_mostro_notifier_freshness_gate_test.dart b/test/features/order/notifiers/abstract_mostro_notifier_freshness_gate_test.dart new file mode 100644 index 000000000..57e125f69 --- /dev/null +++ b/test/features/order/notifiers/abstract_mostro_notifier_freshness_gate_test.dart @@ -0,0 +1,200 @@ +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:sembast/sembast_memory.dart'; + +import 'package:mostro_mobile/data/enums.dart'; +import 'package:mostro_mobile/data/models.dart'; +import 'package:mostro_mobile/features/notifications/providers/notifications_provider.dart'; +import 'package:mostro_mobile/features/order/notifiers/abstract_mostro_notifier.dart'; +import 'package:mostro_mobile/shared/providers/mostro_database_provider.dart'; +import 'package:mostro_mobile/shared/providers/mostro_storage_provider.dart'; +import 'package:mostro_mobile/shared/providers/navigation_notifier_provider.dart'; +import 'package:mostro_mobile/shared/providers/session_notifier_provider.dart'; + +import '../../../mocks.dart'; +import '../../../mocks.mocks.dart'; + +/// Regression coverage for the notification freshness gate swap: both gate +/// call sites in `AbstractMostroNotifier` (subscribe()'s check and +/// handleEvent()'s `isRecent`) must key off `receivedAt` (true client arrival +/// time) instead of `timestamp` (protocol/send time). The single most +/// important scenario here is the reconnect-after-offline case: an old +/// protocol `timestamp` must NOT suppress a message that is genuinely new to +/// this client. +void main() { + TestWidgetsFlutterBinding.ensureInitialized(); + + const orderId = 'freshness-gate-order'; + var dbCounter = 0; + + late ProviderContainer container; + late StreamController messageController; + late AbstractMostroNotifier notifier; + + Session buildSession() { + return Session( + masterKey: NostrKeyPairs( + private: + '1234567890abcdef1234567890abcdef1234567890abcdef1234567890abcdef'), + tradeKey: NostrKeyPairs( + private: + 'abcdef1234567890abcdef1234567890abcdef1234567890abcdef1234567890'), + keyIndex: 0, + fullPrivacy: false, + startTime: DateTime.now(), + orderId: orderId, + role: Role.buyer, + ); + } + + Future flushAsyncOperations() async { + for (var i = 0; i < 10; i++) { + await Future.delayed(Duration.zero); + } + } + + setUp(() async { + dbCounter++; + final db = await databaseFactoryMemory + .openDatabase('freshness_gate_test_$dbCounter.db'); + messageController = StreamController.broadcast(); + + final mockKeyManager = MockKeyManager(); + final mockSessionStorage = MockSessionStorage(); + final mockSettings = MockSettings(); + final mockRef = MockRef(); + final mockSessionNotifier = MockSessionNotifier( + mockRef, mockKeyManager, mockSessionStorage, mockSettings); + mockSessionNotifier.setMockSession(buildSession()); + + container = ProviderContainer(overrides: [ + mostroDatabaseProvider.overrideWithValue(db), + mostroMessageStreamProvider + .overrideWith((ref, id) => messageController.stream), + sessionNotifierProvider.overrideWith((ref) => mockSessionNotifier), + ]); + + final ref = container.read(Provider((ref) => ref)); + notifier = AbstractMostroNotifier(orderId, ref); + notifier.subscribe(); + }); + + tearDown(() async { + notifier.dispose(); + container.dispose(); + await messageController.close(); + }); + + group('AbstractMostroNotifier freshness gate (receivedAt-based)', () { + test('recent receivedAt passes the gate and fires a notification', + () async { + final now = DateTime.now().millisecondsSinceEpoch; + final message = MostroMessage( + id: orderId, + action: Action.paymentFailed, + payload: + PaymentFailed(paymentAttempts: 1, paymentRetriesInterval: 60), + timestamp: now, + receivedAt: now, + ); + + messageController.add(message); + await flushAsyncOperations(); + + final temporary = container.read(currentTemporaryNotificationProvider); + expect(temporary.show, isTrue); + expect(temporary.action, Action.paymentFailed); + }); + + test( + 'old protocol timestamp with a recent receivedAt still passes the ' + 'gate (reconnect-after-offline regression)', () async { + final oldTimestamp = DateTime.now() + .subtract(const Duration(minutes: 10)) + .millisecondsSinceEpoch; + final recentReceivedAt = DateTime.now().millisecondsSinceEpoch; + final message = MostroMessage( + id: orderId, + action: Action.paymentFailed, + payload: + PaymentFailed(paymentAttempts: 1, paymentRetriesInterval: 60), + timestamp: oldTimestamp, + receivedAt: recentReceivedAt, + ); + + messageController.add(message); + await flushAsyncOperations(); + + final temporary = container.read(currentTemporaryNotificationProvider); + expect(temporary.show, isTrue, + reason: 'A message with an old protocol timestamp but a fresh ' + 'receivedAt must still be notified — this is the exact bug ' + 'this change fixes.'); + expect(temporary.action, Action.paymentFailed); + }); + + test( + 'old receivedAt fails the gate: no notification, but state still ' + 'updates unconditionally', () async { + final oldReceivedAt = DateTime.now() + .subtract(const Duration(minutes: 10)) + .millisecondsSinceEpoch; + final message = MostroMessage( + id: orderId, + action: Action.paymentFailed, + payload: + PaymentFailed(paymentAttempts: 2, paymentRetriesInterval: 30), + timestamp: DateTime.now().millisecondsSinceEpoch, + receivedAt: oldReceivedAt, + ); + + messageController.add(message); + await flushAsyncOperations(); + + final temporary = container.read(currentTemporaryNotificationProvider); + expect(temporary.show, isFalse); + + expect(notifier.state.action, Action.paymentFailed); + expect(notifier.state.status, Status.paymentFailed); + }); + + test( + 'dispute action bypasses the gate for state processing despite an ' + 'old receivedAt, but still suppresses notification and navigation', + () async { + final oldReceivedAt = DateTime.now() + .subtract(const Duration(minutes: 10)) + .millisecondsSinceEpoch; + final dispute = Dispute(disputeId: 'dispute-1', status: 'initiated'); + final message = MostroMessage( + id: orderId, + action: Action.disputeInitiatedByPeer, + payload: dispute, + timestamp: oldReceivedAt, + receivedAt: oldReceivedAt, + ); + + final pathBefore = container.read(navigationProvider).path; + + messageController.add(message); + await flushAsyncOperations(); + + // Notification is suppressed for the old event. + expect(container.read(currentTemporaryNotificationProvider).show, + isFalse); + // Navigation is suppressed too (isRecent is false, bypassTimestampGate + // is true). + expect(container.read(navigationProvider).path, pathBefore); + // Yet handleEvent still processed the dispute action (state-only): + // orderId/action are only stamped by handleEvent's dispute branch, not + // by the unconditional updateWith() merge, so this proves the bypass + // actually ran despite the old receivedAt. + expect(notifier.state.dispute, isNotNull); + expect(notifier.state.dispute!.orderId, orderId); + expect(notifier.state.dispute!.action, 'dispute-initiated-by-peer'); + }); + }); +} diff --git a/test/features/restore/restore_manager_reconciliation_test.dart b/test/features/restore/restore_manager_reconciliation_test.dart new file mode 100644 index 000000000..bda2d7287 --- /dev/null +++ b/test/features/restore/restore_manager_reconciliation_test.dart @@ -0,0 +1,189 @@ +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/enums/action.dart'; +import 'package:mostro_mobile/data/models/enums/order_type.dart'; +import 'package:mostro_mobile/data/models/enums/role.dart'; +import 'package:mostro_mobile/data/models/mostro_message.dart'; +import 'package:mostro_mobile/data/models/order.dart'; +import 'package:mostro_mobile/features/key_manager/key_manager_provider.dart'; +import 'package:mostro_mobile/features/order/providers/order_notifier_provider.dart'; +import 'package:mostro_mobile/features/restore/restore_manager.dart'; +import 'package:mostro_mobile/features/settings/settings.dart'; +import 'package:mostro_mobile/features/settings/settings_provider.dart'; +import 'package:mostro_mobile/shared/providers/mostro_database_provider.dart'; +import 'package:mostro_mobile/shared/providers/mostro_service_provider.dart'; +import 'package:mostro_mobile/shared/providers/mostro_storage_provider.dart'; +import 'package:mostro_mobile/shared/providers/order_repository_provider.dart'; +import 'package:mostro_mobile/shared/providers/session_notifier_provider.dart'; +import 'package:mostro_mobile/shared/providers/session_storage_provider.dart'; +import 'package:mostro_mobile/shared/providers/storage_providers.dart'; + +import '../../mocks.dart'; +import '../../mocks.mocks.dart'; + +/// Regression coverage requested by review: _reconcileFiatSent must upgrade a +/// stale cooperativeCancelNoFiatByPeer/generic snapshot to +/// cooperativeCancelFiatSentByPeer once a confirming fiat-sent message is +/// found in storage after the restore buffer has drained. +void main() { + TestWidgetsFlutterBinding.ensureInitialized(); + + group('RestoreService - post-flush fiat-sent reconciliation', () { + late ProviderContainer container; + late MockMostroService mockMostroService; + late MockSharedPreferencesAsync mockPreferences; + late MockOpenOrdersRepository mockOrdersRepository; + late MockDatabase mockDatabase; + late MockSessionStorage mockSessionStorage; + late MockKeyManager mockKeyManager; + late MockSessionNotifier mockSessionNotifier; + late MockMostroStorage mockMostroStorage; + late MockRef ref; + + const testOrderId = 'reconciliation-order-id'; + + Order buildOrder() => const Order( + id: testOrderId, + kind: OrderType.sell, + fiatCode: 'USD', + fiatAmount: 100, + paymentMethod: 'Lightning', + amount: 100, + ); + + setUp(() { + mockMostroService = MockMostroService(); + mockPreferences = MockSharedPreferencesAsync(); + mockOrdersRepository = MockOpenOrdersRepository(); + mockDatabase = MockDatabase(); + mockSessionStorage = MockSessionStorage(); + mockKeyManager = MockKeyManager(); + mockMostroStorage = MockMostroStorage(); + ref = MockRef(); + + final testSettings = MockSettings(); + + mockSessionNotifier = MockSessionNotifier( + ref, mockKeyManager, mockSessionStorage, testSettings); + + when(mockKeyManager.masterKeyPair).thenReturn( + NostrKeyPairs( + private: + '1234567890abcdef1234567890abcdef1234567890abcdef1234567890abcdef'), + ); + when(mockKeyManager.getCurrentKeyIndex()).thenAnswer((_) async => 0); + + // Default: no confirming fiat-sent message on file. Overridden per-test. + when(mockMostroStorage.getAllMessagesForOrderId(any)) + .thenAnswer((_) async => []); + + container = ProviderContainer(overrides: [ + mostroServiceProvider.overrideWithValue(mockMostroService), + orderRepositoryProvider.overrideWithValue(mockOrdersRepository), + sharedPreferencesProvider.overrideWithValue(mockPreferences), + mostroDatabaseProvider.overrideWithValue(mockDatabase), + eventDatabaseProvider.overrideWithValue(mockDatabase), + sessionStorageProvider.overrideWithValue(mockSessionStorage), + keyManagerProvider.overrideWithValue(mockKeyManager), + sessionNotifierProvider.overrideWith((ref) => mockSessionNotifier), + settingsProvider.overrideWith((ref) { + final mockSettings = MockSettingsNotifier(); + mockSettings.state = Settings( + relays: ['wss://relay.damus.io'], + fullPrivacyMode: false, + mostroPublicKey: + '9d9d0455a96871f2dc4289b8312429db2e925f167b37c77bf7b28014be235980', + defaultFiatCode: 'USD', + ); + return mockSettings; + }), + mostroStorageProvider.overrideWithValue(mockMostroStorage), + ]); + }); + + tearDown(() { + container.dispose(); + }); + + test( + 'order regains the release action once a confirming fiat-sent message is found after flush', + () async { + // Arrange: apply the base cooperativelyCanceled snapshot exactly like + // restore()'s per-order loop does (generic action, fiatWasSent still + // false at this point). + final coopCancelMessage = MostroMessage( + id: testOrderId, + action: Action.cooperativeCancelInitiatedByPeer, + payload: buildOrder(), + ); + final notifier = + container.read(orderNotifierProvider(testOrderId).notifier); + notifier.updateStateFromMessage(coopCancelMessage); + + final beforeState = container.read(orderNotifierProvider(testOrderId)); + expect(beforeState.fiatWasSent, isFalse); + expect( + beforeState.getActions(Role.seller), + isNot(contains(Action.release)), + ); + + // Simulate the confirming message becoming visible only after the + // restore buffer is drained (the race this feature fixes). + when(mockMostroStorage.getAllMessagesForOrderId(testOrderId)) + .thenAnswer((_) async => [ + MostroMessage( + id: testOrderId, + action: Action.fiatSent, + ), + ]); + + final restoreService = container.read(restoreServiceProvider); + + // Act + await restoreService.reconcileFiatSentForTesting([testOrderId]); + + // Assert + final afterState = container.read(orderNotifierProvider(testOrderId)); + expect(afterState.fiatWasSent, isTrue); + expect(afterState.action, equals(Action.cooperativeCancelFiatSentByPeer)); + expect(afterState.getActions(Role.seller), contains(Action.release)); + }); + + test( + 'corrects a stale NoFiat action even when a live event (not our own snapshot) produced it before reconciliation ran', + () async { + // Simulates the Codex-flagged race: a buffered live cooperativeCancel + // event is flushed and processed by _onData/updateWith BEFORE + // _reconcileFiatSent runs, so it gets remapped with fiatWasSent still + // false — same NoFiat outcome as the synthetic snapshot case above, + // just from a different source message. + final liveCoopCancelMessage = MostroMessage( + id: testOrderId, + action: Action.cooperativeCancelInitiatedByPeer, + payload: buildOrder(), + ); + final notifier = + container.read(orderNotifierProvider(testOrderId).notifier); + notifier.updateStateFromMessage(liveCoopCancelMessage); + expect( + container.read(orderNotifierProvider(testOrderId)).action, + equals(Action.cooperativeCancelNoFiatByPeer), + ); + + when(mockMostroStorage.getAllMessagesForOrderId(testOrderId)) + .thenAnswer((_) async => [ + MostroMessage(id: testOrderId, action: Action.fiatSent), + ]); + + final restoreService = container.read(restoreServiceProvider); + await restoreService.reconcileFiatSentForTesting([testOrderId]); + + final afterState = container.read(orderNotifierProvider(testOrderId)); + expect(afterState.fiatWasSent, isTrue); + expect(afterState.action, equals(Action.cooperativeCancelFiatSentByPeer)); + expect(afterState.getActions(Role.seller), contains(Action.release)); + }); + }); +} diff --git a/test/mocks.dart b/test/mocks.dart index 0778bb9df..3c6911b4c 100644 --- a/test/mocks.dart +++ b/test/mocks.dart @@ -4,6 +4,7 @@ import 'package:flutter_riverpod/flutter_riverpod.dart'; import 'package:mockito/annotations.dart'; import 'package:mostro_mobile/data/models/session.dart'; import 'package:mostro_mobile/data/models/enums/role.dart'; +import 'package:mostro_mobile/data/repositories/event_storage.dart'; import 'package:mostro_mobile/data/repositories/open_orders_repository.dart'; import 'package:mostro_mobile/data/repositories/session_storage.dart'; import 'package:mostro_mobile/data/repositories/mostro_storage.dart'; @@ -38,6 +39,7 @@ import 'mocks.mocks.dart'; SessionStorage, KeyManager, MostroStorage, + EventStorage, Settings, Ref, ProviderSubscription, diff --git a/test/services/mostro_service_test.dart b/test/services/mostro_service_test.dart index 6a72045f2..d63ac5bfe 100644 --- a/test/services/mostro_service_test.dart +++ b/test/services/mostro_service_test.dart @@ -19,6 +19,7 @@ import 'package:mostro_mobile/features/settings/settings_provider.dart'; import 'package:mostro_mobile/shared/providers/mostro_storage_provider.dart'; import 'package:mostro_mobile/shared/providers/session_notifier_provider.dart'; import 'package:mostro_mobile/data/models/mostro_message.dart'; +import 'package:mostro_mobile/data/models/nostr_event.dart'; import 'package:mostro_mobile/data/models/order.dart'; import 'package:mostro_mobile/features/order/models/order_state.dart'; import 'package:mostro_mobile/features/order/providers/order_notifier_provider.dart'; @@ -28,11 +29,63 @@ import 'package:mostro_mobile/data/models/enums/order_type.dart'; import 'package:mostro_mobile/data/models/enums/action.dart'; import 'package:mostro_mobile/data/repositories/open_orders_repository.dart'; import 'package:mostro_mobile/shared/providers/order_repository_provider.dart'; +import 'package:mostro_mobile/data/repositories/event_storage.dart'; +import 'package:mostro_mobile/features/restore/restore_mode_provider.dart'; +import 'package:mostro_mobile/shared/providers/mostro_service_provider.dart'; +import 'package:mostro_mobile/shared/utils/nostr_utils.dart'; +import 'package:sembast/sembast_memory.dart'; import '../mocks.dart'; import '../mocks.mocks.dart'; import 'mostro_service_helper_functions.dart'; +/// Builds a real, correctly-signed kind-14 (NIP-44 direct) event simulating a +/// live message sent FROM the Mostro daemon TO the client's trade key. Reuses +/// [MostroMessage.wrapNip44] as-is (full-privacy mode, no master key), which +/// is the exact production wire format `_onData` expects to decrypt. +Future buildLiveDaemonEvent({ + required NostrKeyPairs mostroKeyPair, + required NostrKeyPairs recipientTradeKey, + required MostroMessage message, +}) { + return message.wrapNip44( + tradeKey: mostroKeyPair, + recipientPubKey: recipientTradeKey.public, + ); +} + +/// Builds a kind-14 event whose decrypted wire payload carries a spoofed +/// top-level `receivedAt` key alongside the real `order` wrapper — the shape +/// `_onData` would see if a peer ever supplied that client-local field. Used +/// to prove the client's own explicit assignment always overwrites it. +Future buildSpoofedReceivedAtEvent({ + required NostrKeyPairs mostroKeyPair, + required NostrKeyPairs recipientTradeKey, + required MostroMessage message, + required int spoofedReceivedAt, +}) async { + final messageMap = { + 'order': message.toJson(version: 2), + 'receivedAt': spoofedReceivedAt, + }; + final tuple = jsonEncode([messageMap, null, null]); + final encrypted = await NostrUtils.encryptNIP44( + tuple, + mostroKeyPair.private, + recipientTradeKey.public, + ); + + return NostrEvent.fromPartialData( + kind: 14, + content: encrypted, + keyPairs: mostroKeyPair, + tags: [ + ['p', recipientTradeKey.public], + ], + createdAt: DateTime.now(), + ); +} + void main() { // Provide dummy values for Mockito provideDummy(Settings( @@ -49,6 +102,10 @@ void main() { // Add dummy for MostroStorage provideDummy(MockMostroStorage()); + // Add dummy for EventStorage (never invoked directly — its constructor only + // stores the Database reference, so a mock db suffices as a placeholder). + provideDummy(EventStorage(db: MockDatabase())); + // Add dummy for NostrService provideDummy(MockNostrService()); @@ -564,6 +621,368 @@ void main() { expect(payload, isNull); }); }); + + group('MostroService._onData receivedAt stamping', () { + late NostrKeyPairs mostroKeyPair; + late NostrKeyPairs clientTradeKey; + late MockSettings liveSettings; + late MockMostroStorage capturingStorage; + late Database eventDb; + late EventStorage realEventStorage; + + setUp(() async { + mostroKeyPair = NostrUtils.generateKeyPair(); + clientTradeKey = NostrUtils.generateKeyPair(); + + liveSettings = MockSettings(); + when(liveSettings.mostroPublicKey).thenReturn(mostroKeyPair.public); + when(mockRef.read(settingsProvider)).thenReturn(liveSettings); + + capturingStorage = MockMostroStorage(); + when(capturingStorage.addMessage(any, any)) + .thenAnswer((_) async {}); + when(mockRef.read(mostroStorageProvider)).thenReturn(capturingStorage); + + eventDb = await databaseFactoryMemory.openDatabase( + 'mostro_service_test_events_${DateTime.now().microsecondsSinceEpoch}.db', + ); + realEventStorage = EventStorage(db: eventDb); + when(mockRef.read(eventStorageProvider)).thenReturn(realEventStorage); + + when(mockRef.read(isRestoringProvider)).thenReturn(false); + + final session = Session( + startTime: DateTime.now(), + masterKey: clientTradeKey, + keyIndex: 1, + tradeKey: clientTradeKey, + fullPrivacy: true, + ); + when(mockRef.read(sessionNotifierProvider)).thenReturn([session]); + + // Reconstruct the service so its internal `_settings` field (captured + // at construction) reflects `liveSettings`, not the outer setUp's. + mostroService = MostroService(mockRef); + }); + + test( + 'a wire payload carrying a receivedAt-shaped key is overwritten by ' + 'the client explicit assignment', () async { + final message = MostroMessage(action: Action.fiatSent, id: 'order-c'); + final spoofed = 1; // implausibly old, would fail any freshness check + final event = await buildSpoofedReceivedAtEvent( + mostroKeyPair: mostroKeyPair, + recipientTradeKey: clientTradeKey, + message: message, + spoofedReceivedAt: spoofed, + ); + + final before = DateTime.now().millisecondsSinceEpoch; + await mostroService.onDataForTesting(event); + final after = DateTime.now().millisecondsSinceEpoch; + + final captured = verify( + capturingStorage.addMessage(any, captureAny), + ).captured; + final stamped = captured.single as MostroMessage; + expect(stamped.receivedAt, isNot(spoofed)); + expect(stamped.receivedAt!, greaterThanOrEqualTo(before)); + expect(stamped.receivedAt!, lessThanOrEqualTo(after)); + }); + }); + + group('MostroService restore buffer preserve-if-present and flush timing', + () { + late NostrKeyPairs mostroKeyPair; + late NostrKeyPairs clientTradeKey; + late MockSettings liveSettings; + late MockMostroStorage capturingStorage; + late Database eventDb; + late EventStorage realEventStorage; + + setUp(() async { + mostroKeyPair = NostrUtils.generateKeyPair(); + clientTradeKey = NostrUtils.generateKeyPair(); + + liveSettings = MockSettings(); + when(liveSettings.mostroPublicKey).thenReturn(mostroKeyPair.public); + when(mockRef.read(settingsProvider)).thenReturn(liveSettings); + + capturingStorage = MockMostroStorage(); + when(capturingStorage.addMessage(any, any)) + .thenAnswer((_) async {}); + when(mockRef.read(mostroStorageProvider)).thenReturn(capturingStorage); + + eventDb = await databaseFactoryMemory.openDatabase( + 'mostro_service_test_restore_${DateTime.now().microsecondsSinceEpoch}.db', + ); + realEventStorage = EventStorage(db: eventDb); + when(mockRef.read(eventStorageProvider)).thenReturn(realEventStorage); + + final session = Session( + startTime: DateTime.now(), + masterKey: clientTradeKey, + keyIndex: 1, + tradeKey: clientTradeKey, + fullPrivacy: true, + ); + when(mockRef.read(sessionNotifierProvider)).thenReturn([session]); + + mostroService = MostroService(mockRef); + }); + + test( + 'redelivering the same event id while restoring preserves the ' + 'first-buffered receivedAtMs', () async { + when(mockRef.read(isRestoringProvider)).thenReturn(true); + + final message = MostroMessage(action: Action.fiatSent, id: 'order-d'); + final event = await buildLiveDaemonEvent( + mostroKeyPair: mostroKeyPair, + recipientTradeKey: clientTradeKey, + message: message, + ); + + final beforeFirst = DateTime.now().millisecondsSinceEpoch; + await mostroService.onDataForTesting(event); + final firstStamp = + mostroService.restoreBufferForTesting[event.id]!.receivedAtMs; + expect(firstStamp, greaterThanOrEqualTo(beforeFirst)); + + await Future.delayed(const Duration(milliseconds: 5)); + final beforeSecond = DateTime.now().millisecondsSinceEpoch; + + // Redeliver the same event id: production code releases the dedup + // reservation after buffering, so this re-enters _onData. + await mostroService.onDataForTesting(event); + final secondStamp = + mostroService.restoreBufferForTesting[event.id]!.receivedAtMs; + + expect(secondStamp, equals(firstStamp)); + expect(secondStamp, lessThan(beforeSecond)); + }); + }); + + group('MostroService restore buffer and reorder', () { + late MockRef restoreRef; + late MockSettings restoreSettings; + late MockEventStorage mockEventStorage; + late MockMostroStorage mockMostroStorage; + late MockSubscriptionManager restoreSubscriptionManager; + late MostroService restoreService; + late NostrKeyPairs mostroRestoreKeyPair; + late NostrKeyPairs userTradeKeyPair; + late List> addedMessages; + late Set reservedEventIds; + + setUp(() { + restoreRef = MockRef(); + restoreSettings = MockSettings(); + mockEventStorage = MockEventStorage(); + mockMostroStorage = MockMostroStorage(); + addedMessages = []; + reservedEventIds = {}; + + mostroRestoreKeyPair = NostrUtils.generateKeyPair(); + userTradeKeyPair = NostrUtils.generateKeyPair(); + + when(restoreSettings.mostroPublicKey) + .thenReturn(mostroRestoreKeyPair.public); + when(restoreRef.read(settingsProvider)).thenReturn(restoreSettings); + + // Real in-memory dedup semantics, so the buffer/flush/replay tests don't + // need per-test re-stubbing of hasItem/putItem/deleteItem. + when(restoreRef.read(eventStorageProvider)).thenReturn(mockEventStorage); + when(mockEventStorage.hasItem(any)).thenAnswer((invocation) async { + final id = invocation.positionalArguments[0] as String; + return reservedEventIds.contains(id); + }); + when(mockEventStorage.putItem(any, any)).thenAnswer((invocation) async { + final id = invocation.positionalArguments[0] as String; + reservedEventIds.add(id); + }); + when(mockEventStorage.deleteItem(any)).thenAnswer((invocation) async { + final id = invocation.positionalArguments[0] as String; + reservedEventIds.remove(id); + }); + + when(restoreRef.read(mostroStorageProvider)) + .thenReturn(mockMostroStorage); + when(mockMostroStorage.addMessage(any, any)) + .thenAnswer((invocation) async { + addedMessages.add(invocation.positionalArguments); + }); + + when(restoreRef.read(sessionNotifierProvider)).thenReturn([]); + when(restoreRef.read(isRestoringProvider)).thenReturn(false); + + // SubscriptionManager's constructor (which MockSubscriptionManager + // inherits unchanged) registers a session listener eagerly and reads + // the connected node's info stream. + when(restoreRef.listen>( + any, + any, + onError: anyNamed('onError'), + fireImmediately: anyNamed('fireImmediately'), + )).thenReturn(MockProviderSubscription>()); + + final restoreOrderRepo = MockOpenOrdersRepository(); + when(restoreOrderRepo.mostroInstanceStream) + .thenAnswer((_) => const Stream.empty()); + when(restoreRef.read(orderRepositoryProvider)) + .thenReturn(restoreOrderRepo); + + restoreSubscriptionManager = MockSubscriptionManager(restoreRef); + when(restoreRef.read(subscriptionManagerProvider)) + .thenReturn(restoreSubscriptionManager); + + // init() registers a ref.listen(isRestoringProvider, ...) callback; + // it must be stubbed for init() to succeed, even though these tests + // drive the flush directly via flushRestoreBufferForTesting() rather + // than through the reactive listener. + when(restoreRef.listen(isRestoringProvider, any)) + .thenReturn(MockProviderSubscription()); + + restoreService = MostroService(restoreRef); + restoreService.init(); + + // SubscriptionManager's constructor makes its own unrelated reads of + // sessionNotifierProvider (e.g. _initializeExistingSessions); clear that + // interaction history so verifyNever assertions below only observe + // calls made by the test body itself. + clearInteractions(restoreRef); + }); + + tearDown(() { + restoreService.dispose(); + restoreSubscriptionManager.dispose(); + }); + + void setMatchingSession() { + final session = Session( + startTime: DateTime.now(), + masterKey: NostrUtils.generateKeyPair(), + keyIndex: 1, + tradeKey: userTradeKeyPair, + fullPrivacy: false, + ); + when(restoreRef.read(sessionNotifierProvider)).thenReturn([session]); + } + + NostrEvent bareEvent({required int kind, required String id}) { + return NostrEvent( + id: id, + kind: kind, + content: 'irrelevant-not-decrypted-while-buffered', + pubkey: mostroRestoreKeyPair.public, + sig: '', + createdAt: DateTime.now(), + tags: [ + ['p', userTradeKeyPair.public], + ], + ); + } + + test('buffers a v1 kind-1059 event during restore without processing it', + () async { + when(restoreRef.read(isRestoringProvider)).thenReturn(true); + final event = bareEvent(kind: 1059, id: 'evt-v1-buffer'); + + await restoreService.onDataForTesting(event); + + expect( + restoreService.restoreBufferForTesting.keys, + contains('evt-v1-buffer'), + ); + expect(addedMessages, isEmpty); + verifyNever(restoreRef.read(sessionNotifierProvider)); + }); + + test( + 'buffers an event before the session-match check, even when no session exists yet', + () async { + when(restoreRef.read(isRestoringProvider)).thenReturn(true); + when(restoreRef.read(sessionNotifierProvider)).thenReturn([]); + final event = bareEvent(kind: 1059, id: 'evt-no-session'); + + await restoreService.onDataForTesting(event); + + expect( + restoreService.restoreBufferForTesting.keys, + contains('evt-no-session'), + reason: + 'reorder fix: buffering must precede the matchingSession==null early return', + ); + }); + + test( + 'replays buffered events in arrival order, clearing dedup before each replay', + () async { + setMatchingSession(); + when(restoreRef.read(isRestoringProvider)).thenReturn(true); + + final eventA = await MostroMessage(action: Action.newOrder, id: 'order-a') + .wrap(tradeKey: mostroRestoreKeyPair, recipientPubKey: userTradeKeyPair.public); + final eventB = await MostroMessage(action: Action.newOrder, id: 'order-b') + .wrap(tradeKey: mostroRestoreKeyPair, recipientPubKey: userTradeKeyPair.public); + final eventC = await MostroMessage(action: Action.newOrder, id: 'order-c') + .wrap(tradeKey: mostroRestoreKeyPair, recipientPubKey: userTradeKeyPair.public); + + await restoreService.onDataForTesting(eventA); + await restoreService.onDataForTesting(eventB); + await restoreService.onDataForTesting(eventC); + expect( + restoreService.restoreBufferForTesting.keys.toList(), + [eventA.id, eventB.id, eventC.id], + ); + + when(restoreRef.read(isRestoringProvider)).thenReturn(false); + await restoreService.flushRestoreBufferForTesting(); + + verifyInOrder([ + mockEventStorage.deleteItem(eventA.id!), + mockEventStorage.deleteItem(eventB.id!), + mockEventStorage.deleteItem(eventC.id!), + ]); + expect(restoreService.restoreBufferForTesting, isEmpty); + expect( + addedMessages.map((args) => (args[1] as MostroMessage).id).toList(), + ['order-a', 'order-b', 'order-c'], + ); + }); + + test( + "v1 replayed event uses the inner rumor's real timestamp, never the gift-wrap outer time", + () async { + setMatchingSession(); + when(restoreRef.read(isRestoringProvider)).thenReturn(true); + + final message = MostroMessage(action: Action.newOrder, id: 'order-ts-v1'); + final event = await message.wrap( + tradeKey: mostroRestoreKeyPair, + recipientPubKey: userTradeKeyPair.public, + ); + final expectedInnerCreatedAt = + (await event.unWrap(userTradeKeyPair.private)) + .createdAt! + .millisecondsSinceEpoch; + + await restoreService.onDataForTesting(event); + + when(restoreRef.read(isRestoringProvider)).thenReturn(false); + await restoreService.flushRestoreBufferForTesting(); + + expect(addedMessages, hasLength(1)); + final replayedMessage = addedMessages.first[1] as MostroMessage; + expect(replayedMessage.timestamp, expectedInnerCreatedAt); + expect( + replayedMessage.timestamp, + isNot(event.createdAt!.millisecondsSinceEpoch), + reason: + 'the outer gift-wrap timestamp is NIP-59 randomized and must not be used for ordering', + ); + }); + }); } // Testable service that captures publishOrder calls