diff --git a/lib/data/repositories/mostro_storage.dart b/lib/data/repositories/mostro_storage.dart index 659180da..b248c2d5 100644 --- a/lib/data/repositories/mostro_storage.dart +++ b/lib/data/repositories/mostro_storage.dart @@ -1,3 +1,5 @@ +import 'dart:async'; +import 'package:flutter/foundation.dart'; import 'package:mostro_mobile/services/logger_service.dart'; import 'package:mostro_mobile/data/models/payload.dart'; import 'package:sembast/sembast.dart'; @@ -5,15 +7,105 @@ import 'package:mostro_mobile/data/models/mostro_message.dart'; import 'package:mostro_mobile/data/repositories/base_storage.dart'; class MostroStorage extends BaseStorage { - - MostroStorage({required Database db}) : super(db, stringMapStoreFactory.store('orders')); + /// In-memory index by order id (newest first). Every write used to wake + /// one Sembast query listener per OrderNotifier and per visible trade row, + /// each re-filtering the whole unindexed store on the UI isolate. The + /// index is warmed from disk once; watchers are served from memory and + /// demultiplexed per order. + final Map> _byOrder = {}; + final StreamController _orderChanges = StreamController.broadcast(); + final Set _writesInFlight = {}; + Future? _warmup; + + @visibleForTesting + int get debugIndexSize => _byOrder.length; + + Future _ensureIndex() { + return _warmup ??= () async { + try { + final all = await getAll(); + for (final message in all) { + _indexAdd(message, notify: false); + } + logger.i('Mostro message index warmed: ${_byOrder.length} orders'); + } catch (e, stack) { + // A retained failed future would rethrow on every later query for the + // rest of the session: drop it so the next call retries the warm-up. + _warmup = null; + _byOrder.clear(); + logger.e('Mostro message index warm-up failed', + error: e, stackTrace: stack); + rethrow; + } + }(); + } + + void _indexAdd(MostroMessage message, {bool notify = true}) { + final orderId = message.id; + if (orderId == null) return; + final list = _byOrder.putIfAbsent(orderId, () => []); + list.add(message); + list.sort((a, b) => (b.timestamp ?? 0).compareTo(a.timestamp ?? 0)); + if (notify) _notifyOrder(orderId); + } + + void _notifyOrder(String orderId) { + if (!_orderChanges.isClosed) _orderChanges.add(orderId); + } + + MostroMessage? _latestFor(String orderId) { + final list = _byOrder[orderId]; + return (list == null || list.isEmpty) ? null : list.first; + } + + List _historyFor(String orderId) => + List.unmodifiable(_byOrder[orderId] ?? const []); + + Stream _watchOrder(String orderId, R Function() read) { + late StreamController controller; + StreamSubscription? changes; + controller = StreamController( + onListen: () async { + try { + await _ensureIndex(); + } catch (e, stack) { + // onListen's future is not observed by the controller: without this + // the subscriber would wait forever and the rejection would surface + // as an unhandled async error. + if (!controller.isClosed) { + controller.addError(e, stack); + await controller.close(); + } + return; + } + if (controller.isClosed) return; + controller.add(read()); + changes = _orderChanges.stream + .where((changed) => changed == orderId) + .listen((_) => controller.add(read())); + }, + onCancel: () async { + await changes?.cancel(); + // Skip when onListen already closed the controller: its done future + // only completes after onCancel returns, so awaiting close() again + // here would deadlock. + if (!controller.isClosed) await controller.close(); + }, + ); + return controller.stream; + } + /// Save or update any MostroMessage Future addMessage(String key, MostroMessage message) async { final id = key; + // Claimed synchronously so two concurrent writes for the same key cannot + // both pass the existence check below and index the message twice. + if (!_writesInFlight.add(id)) return; try { + await _ensureIndex(); if (await hasItem(id)) return; // Add metadata for easier querying final Map dbMap = message.toJson(); @@ -21,6 +113,7 @@ class MostroStorage extends BaseStorage { dbMap['timestamp'] = message.timestamp; await store.record(id).put(db, dbMap); + _indexAdd(message); logger.i( 'Saved message of type ${message.action} with order id ${message.id}', ); @@ -31,6 +124,8 @@ class MostroStorage extends BaseStorage { stackTrace: stack, ); rethrow; + } finally { + _writesInFlight.remove(id); } } @@ -44,22 +139,47 @@ class MostroStorage extends BaseStorage { } } - /// Delete all messages - Future deleteAllMessages() async { + /// Delete every message, on disk and in the index. + /// + /// Overridden rather than left to callers: the wipe-everything flows + /// (account restore, master-key rotation) call [deleteAll] directly, and a + /// disk-only wipe would leave the index serving deleted messages — merged + /// with the restored ones — for the rest of the session. + @override + Future deleteAll() async { try { - await deleteAll(); + // Serialize with any warm-up already in flight, so it cannot repopulate + // the index after the wipe. A broken index must not block the wipe. + try { + await _ensureIndex(); + } catch (_) {} + await super.deleteAll(); + final orderIds = _byOrder.keys.toList(); + _byOrder.clear(); + orderIds.forEach(_notifyOrder); logger.i('All messages deleted'); } catch (e, stack) { - logger.e('deleteAllMessages failed', error: e, stackTrace: stack); + logger.e('deleteAll failed', error: e, stackTrace: stack); rethrow; } } + /// Delete all messages + Future deleteAllMessages() => deleteAll(); + /// Delete all messages by Id regardless of type Future deleteAllMessagesByOrderId(String orderId) async { + // Awaited first so a warm-up in flight cannot re-index records this call + // is about to delete, but its failure (already logged and reset for + // retry) must not leave the records on disk. + try { + await _ensureIndex(); + } catch (_) {} await deleteWhere( Filter.equals('id', orderId), ); + _byOrder.remove(orderId); + _notifyOrder(orderId); } /// Filter messages by payload type @@ -106,63 +226,30 @@ class MostroStorage extends BaseStorage { /// Get the latest message for an order, regardless of type Future getLatestMessageById(String orderId) async { - final finder = Finder( - filter: Filter.equals('id', orderId), - sortOrders: _getDefaultSort(), - limit: 1, - ); - - final snapshot = await store.findFirst(db, finder: finder); - if (snapshot != null) { - return MostroMessage.fromJson(snapshot.value); - } - return null; + await _ensureIndex(); + return _latestFor(orderId); } /// Stream of the latest message for an order - Stream watchLatestMessage(String orderId) { - final query = store.query( - finder: Finder( - filter: Filter.equals('id', orderId), - sortOrders: _getDefaultSort(), - limit: 1, - ), - ); - - return query.onSnapshots(db).map((snaps) => - snaps.isEmpty ? null : MostroMessage.fromJson(snaps.first.value)); - } + Stream watchLatestMessage(String orderId) => + _watchOrder(orderId, () => _latestFor(orderId)); /// Stream of the latest message for an order whose payload is of type T - Stream watchLatestMessageOfType(String orderId) { - // Watch all messages for the orderId, sorted by timestamp descending - final query = store.query( - finder: Finder( - filter: Filter.equals('id', orderId), - sortOrders: _getDefaultSort(), - ), - ); - return query.onSnapshots(db).map((snaps) { - for (final snap in snaps) { - final msg = MostroMessage.fromJson(snap.value); - if (msg.payload is T) { - return msg; + Stream watchLatestMessageOfType(String orderId) => + _watchOrder(orderId, () { + for (final message in _byOrder[orderId] ?? const []) { + if (message.payload is T) return message; } - } - return null; - }); - } + return null; + }); - // Use the same sorting across all methods that return lists of messages - List _getDefaultSort() => [SortOrder('timestamp', false, true)]; + /// Stream of all messages for an order (newest first) + Stream> watchAllMessages(String orderId) => + _watchOrder(orderId, () => _historyFor(orderId)); - /// Stream of all messages for an order - Stream> watchAllMessages(String orderId) { - return watch( - filter: Filter.equals('id', orderId), - sort: _getDefaultSort(), - ); - } + // Sorting for the remaining DB-backed query (request-id lookups are + // transient one-offs during order creation and stay on Sembast). + List _getDefaultSort() => [SortOrder('timestamp', false, true)]; Stream watchByRequestId(int requestId) { final query = store.query( @@ -173,18 +260,20 @@ class MostroStorage extends BaseStorage { ), ); - return query.onSnapshots(db).map((snapshots) => - snapshots.isNotEmpty ? MostroMessage.fromJson(snapshots.first.value) : null); + return query.onSnapshots(db).map((snapshots) => snapshots.isNotEmpty + ? MostroMessage.fromJson(snapshots.first.value) + : null); } Future> getAllMessagesForOrderId(String orderId) async { - final finder = Finder( - filter: Filter.equals('id', orderId), - sortOrders: [SortOrder('timestamp', false)]); + await _ensureIndex(); + return _historyFor(orderId); + } - final snapshots = await store.find(db, finder: finder); - return snapshots - .map((snapshot) => MostroMessage.fromJson(snapshot.value)) - .toList(); + /// Release the change stream. The app-lifetime provider never disposes, but + /// short-lived instances (tests) would otherwise leak a controller each. + Future dispose() async { + _byOrder.clear(); + await _orderChanges.close(); } } diff --git a/pubspec.lock b/pubspec.lock index 9c358b50..41dda05b 100644 --- a/pubspec.lock +++ b/pubspec.lock @@ -5,10 +5,10 @@ packages: dependency: transitive description: name: _fe_analyzer_shared - sha256: da0d9209ca76bde579f2da330aeb9df62b6319c834fa7baae052021b0462401f + sha256: c209688d9f5a5f26b2fb47a188131a6fb9e876ae9e47af3737c0b4f58a93470d url: "https://pub.dev" source: hosted - version: "85.0.0" + version: "91.0.0" _flutterfire_internals: dependency: transitive description: @@ -21,10 +21,10 @@ packages: dependency: transitive description: name: analyzer - sha256: "974859dc0ff5f37bc4313244b3218c791810d03ab3470a579580279ba971a48d" + sha256: f51c8499b35f9b26820cfe914828a6a98a94efd5cc78b37bb7d03debae3a1d08 url: "https://pub.dev" source: hosted - version: "7.7.1" + version: "8.4.1" app_links: dependency: "direct main" description: @@ -221,10 +221,10 @@ packages: dependency: transitive description: name: characters - sha256: f71061c654a3380576a52b451dd5532377954cf9dbd272a78fc8479606670803 + sha256: faf38497bda5ead2a8c7615f4f7939df04333478bf32e4173fcb06d428b5716b url: "https://pub.dev" source: hosted - version: "1.4.0" + version: "1.4.1" checked_yaml: dependency: transitive description: @@ -342,10 +342,10 @@ packages: dependency: transitive description: name: dart_style - sha256: "8a0e5fba27e8ee025d2ffb4ee820b4e6e2cf5e4246a6b1a477eb66866947e0bb" + sha256: a9c30492da18ff84efe2422ba2d319a89942d93e58eb0b73d32abe822ef54b7b url: "https://pub.dev" source: hosted - version: "3.1.1" + version: "3.1.3" dbus: dependency: transitive description: @@ -1073,26 +1073,26 @@ packages: dependency: transitive description: name: matcher - sha256: dc58c723c3c24bf8d3e2d3ad3f2f9d7bd9cf43ec6feaa64181775e60190153f2 + sha256: dc0b7dc7651697ea4ff3e69ef44b0407ea32c487a39fff6a4004fa585e901861 url: "https://pub.dev" source: hosted - version: "0.12.17" + version: "0.12.19" material_color_utilities: dependency: transitive description: name: material_color_utilities - sha256: f7142bb1154231d7ea5f96bc7bde4bda2a0945d2806bb11670e30b850d56bdec + sha256: "9c337007e82b1889149c82ed242ed1cb24a66044e30979c44912381e9be4c48b" url: "https://pub.dev" source: hosted - version: "0.11.1" + version: "0.13.0" meta: dependency: transitive description: name: meta - sha256: e3641ec5d63ebf0d9b41bd43201a66e3fc79a65db5f61fc181f04cd27aab950c + sha256: "23f08335362185a5ea2ad3a4e597f1375e78bce8a040df5c600c8d3552ef2394" url: "https://pub.dev" source: hosted - version: "1.16.0" + version: "1.17.0" mime: dependency: "direct main" description: @@ -1113,10 +1113,10 @@ packages: dependency: "direct dev" description: name: mockito - sha256: "2314cbe9165bcd16106513df9cf3c3224713087f09723b128928dc11a4379f99" + sha256: eff30d002f0c8bf073b6f929df4483b543133fcafce056870163587b03f1d422 url: "https://pub.dev" source: hosted - version: "5.5.0" + version: "5.6.4" mutation_test: dependency: "direct dev" description: @@ -1655,26 +1655,26 @@ packages: dependency: "direct dev" description: name: test - sha256: "65e29d831719be0591f7b3b1a32a3cda258ec98c58c7b25f7b84241bc31215bb" + sha256: "280d6d890011ca966ad08df7e8a4ddfab0fb3aa49f96ed6de56e3521347a9ae7" url: "https://pub.dev" source: hosted - version: "1.26.2" + version: "1.30.0" test_api: dependency: transitive description: name: test_api - sha256: "522f00f556e73044315fa4585ec3270f1808a4b186c936e612cab0b565ff1e00" + sha256: "8161c84903fd860b26bfdefb7963b3f0b68fee7adea0f59ef805ecca346f0c7a" url: "https://pub.dev" source: hosted - version: "0.7.6" + version: "0.7.10" test_core: dependency: transitive description: name: test_core - sha256: "80bf5a02b60af04b09e14f6fe68b921aad119493e26e490deaca5993fef1b05a" + sha256: "0381bd1585d1a924763c308100f2138205252fb90c9d4eeaf28489ee65ccde51" url: "https://pub.dev" source: hosted - version: "0.6.11" + version: "0.6.16" timeago: dependency: "direct main" description: diff --git a/test/data/repositories/mostro_storage_index_test.dart b/test/data/repositories/mostro_storage_index_test.dart new file mode 100644 index 00000000..71eb0740 --- /dev/null +++ b/test/data/repositories/mostro_storage_index_test.dart @@ -0,0 +1,197 @@ +import 'package:flutter_test/flutter_test.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'; +import 'package:sembast/sembast_memory.dart'; + +/// Every `orders`-store write used to wake one Sembast query listener per +/// OrderNotifier and per visible trade row, each re-filtering the whole +/// unindexed store on the UI isolate: O((notifiers + rows) × messages) per +/// incoming message. The storage now keeps a single in-memory index by order +/// id; watchers are served from it and Sembast remains the persistence +/// layer, warmed once per cold start. +void main() { + late MostroStorage storage; + + MostroMessage message(String orderId, Action action, int timestamp) { + final m = MostroMessage(action: action, id: orderId); + m.timestamp = timestamp; + return m; + } + + setUp(() async { + final db = await newDatabaseFactoryMemory().openDatabase('index_test.db'); + storage = MostroStorage(db: db); + }); + + test('the latest-message watcher tracks adds for its order', () async { + final emissions = []; + final sub = storage.watchLatestMessage('a').listen(emissions.add); + addTearDown(sub.cancel); + await Future.delayed(const Duration(milliseconds: 20)); + + await storage.addMessage('k1', message('a', Action.newOrder, 1000)); + await storage.addMessage('k2', message('a', Action.payInvoice, 2000)); + await Future.delayed(const Duration(milliseconds: 20)); + + expect(emissions.last!.action, Action.payInvoice); + }); + + test('a write for another order does not notify this watcher', () async { + await storage.addMessage('k1', message('a', Action.newOrder, 1000)); + var emissions = 0; + final sub = storage.watchLatestMessage('a').skip(1).listen((_) { + emissions++; + }); + addTearDown(sub.cancel); + await Future.delayed(const Duration(milliseconds: 20)); + + await storage.addMessage('k2', message('b', Action.newOrder, 2000)); + await Future.delayed(const Duration(milliseconds: 20)); + + expect(emissions, 0, + reason: 'per-order streams must be demultiplexed in memory'); + }); + + test('history is served newest-first and per order', () async { + await storage.addMessage('k1', message('a', Action.newOrder, 1000)); + await storage.addMessage('k2', message('a', Action.payInvoice, 3000)); + await storage.addMessage('k3', message('b', Action.newOrder, 2000)); + + final history = await storage.getAllMessagesForOrderId('a'); + + expect(history.map((m) => m.action), [Action.payInvoice, Action.newOrder]); + expect(await storage.getLatestMessageById('b'), + isA().having((m) => m.id, 'id', 'b')); + }); + + test('a cold start warms the index from disk', () async { + await storage.addMessage('k1', message('a', Action.newOrder, 1000)); + + // New storage over the same database: what a restart looks like. + final restarted = MostroStorage(db: storage.db); + final latest = await restarted.getLatestMessageById('a'); + + expect(latest!.action, Action.newOrder); + expect(restarted.debugIndexSize, greaterThan(0)); + }); + + test('deleting an order clears it from index and watchers', () async { + await storage.addMessage('k1', message('a', Action.newOrder, 1000)); + final emissions = []; + final sub = storage.watchLatestMessage('a').listen(emissions.add); + addTearDown(sub.cancel); + await Future.delayed(const Duration(milliseconds: 20)); + + await storage.deleteAllMessagesByOrderId('a'); + await Future.delayed(const Duration(milliseconds: 20)); + + expect(emissions.last, isNull); + expect(await storage.getAllMessagesForOrderId('a'), isEmpty); + }); + + test('deleteAll clears the index, not just the disk', () async { + await storage.addMessage('k1', message('a', Action.newOrder, 1000)); + await storage.addMessage('k2', message('a', Action.payInvoice, 2000)); + final emissions = []; + final sub = storage.watchLatestMessage('a').listen(emissions.add); + addTearDown(sub.cancel); + await Future.delayed(const Duration(milliseconds: 20)); + + // The wipe-everything flows (account restore, master-key rotation) call + // the inherited deleteAll(), not deleteAllMessages(). + await storage.deleteAll(); + await Future.delayed(const Duration(milliseconds: 20)); + + expect(await storage.getLatestMessageById('a'), isNull); + expect(await storage.getAllMessagesForOrderId('a'), isEmpty); + expect(storage.debugIndexSize, 0); + expect(emissions.last, isNull); + }); + + test('a restore does not merge pre-wipe history into the restored order', + () async { + await storage.addMessage('k1', message('a', Action.newOrder, 1000)); + await storage.addMessage('k2', message('a', Action.payInvoice, 2000)); + + // What restore_manager does: wipe, then re-add the restored messages. + await storage.deleteAll(); + await storage.addMessage('k3', message('a', Action.newOrder, 3000)); + + final history = await storage.getAllMessagesForOrderId('a'); + expect(history.map((m) => m.action), [Action.newOrder]); + }); + + test('a failed warm-up is retried instead of poisoning every query', + () async { + await storage.addMessage('k1', message('a', Action.newOrder, 1000)); + final flaky = _FlakyWarmupStorage(db: storage.db); + + await expectLater( + flaky.getLatestMessageById('a'), throwsA(isA())); + + // A retained failed future would rethrow here for the rest of the session. + expect((await flaky.getLatestMessageById('a'))!.action, Action.newOrder); + }); + + test('a watcher whose warm-up fails receives the error and closes', () async { + // onListen is async: a rejected warm-up used to escape as an unhandled + // error while the subscriber waited for data that never came. + final flaky = _FlakyWarmupStorage(db: storage.db); + + await expectLater( + flaky.watchLatestMessage('a'), + emitsInOrder([emitsError(isA()), emitsDone]), + ); + }); + + test('a failed warm-up does not stop an order deletion reaching disk', + () async { + await storage.addMessage('k1', message('a', Action.newOrder, 1000)); + final flaky = _FlakyWarmupStorage(db: storage.db); + + await flaky.deleteAllMessagesByOrderId('a'); + + // The retry warms from disk, so an empty result proves the delete ran. + expect(await flaky.getAllMessagesForOrderId('a'), isEmpty); + }); + + test('concurrent writes of the same key index the message once', () async { + await Future.wait([ + storage.addMessage('k1', message('a', Action.newOrder, 1000)), + storage.addMessage('k1', message('a', Action.newOrder, 1000)), + ]); + + expect((await storage.getAllMessagesForOrderId('a')).length, 1); + }); + + test('duplicate keys are ignored without notifying watchers twice', () async { + await storage.addMessage('k1', message('a', Action.newOrder, 1000)); + var emissions = 0; + final sub = storage.watchLatestMessage('a').skip(1).listen((_) { + emissions++; + }); + addTearDown(sub.cancel); + await Future.delayed(const Duration(milliseconds: 20)); + + await storage.addMessage('k1', message('a', Action.newOrder, 1000)); + await Future.delayed(const Duration(milliseconds: 20)); + + expect(emissions, 0); + }); +} + +/// Fails the first index warm-up read, then behaves normally. +class _FlakyWarmupStorage extends MostroStorage { + _FlakyWarmupStorage({required super.db}); + + int _failuresLeft = 1; + + @override + Future> getAll() { + if (_failuresLeft-- > 0) { + return Future.error(StateError('transient read failure')); + } + return super.getAll(); + } +}