perf: serve mostro message queries from a single in-memory index

lib/data/repositories/mostro_storage.dart

@@ -1,26 +1,105 @@
+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';
 import 'package:mostro_mobile/data/models/mostro_message.dart';
 import 'package:mostro_mobile/data/repositories/base_storage.dart';
 
 class MostroStorage extends BaseStorage<MostroMessage> {
-
-
   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<String, List<MostroMessage>> _byOrder = {};
+  final StreamController<String> _orderChanges = StreamController.broadcast();
+  final Set<String> _writesInFlight = <String>{};
+  Future<void>? _warmup;
+
+  @visibleForTesting
+  int get debugIndexSize => _byOrder.length;
+
+  Future<void> _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, () => <MostroMessage>[]);
+    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<MostroMessage> _historyFor(String orderId) =>
+      List.unmodifiable(_byOrder[orderId] ?? const <MostroMessage>[]);
+
+  Stream<R> _watchOrder<R>(String orderId, R Function() read) {
+    late StreamController<R> controller;
+    StreamSubscription<String>? changes;
+    controller = StreamController<R>(
+      onListen: () async {
+        await _ensureIndex();
+        if (controller.isClosed) return;
+        controller.add(read());
+        changes = _orderChanges.stream
+            .where((changed) => changed == orderId)
+            .listen((_) => controller.add(read()));
+      },
+      onCancel: () async {
+        await changes?.cancel();
+        await controller.close();
+      },
+    );
+    return controller.stream;
+  }
+
   /// Save or update any MostroMessage
   Future<void> 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<String, dynamic> dbMap = message.toJson();
       message.timestamp ??= DateTime.now().millisecondsSinceEpoch;
       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 +110,8 @@ class MostroStorage extends BaseStorage<MostroMessage> {
         stackTrace: stack,
       );
       rethrow;
+    } finally {
+      _writesInFlight.remove(id);
     }
   }
 
@@ -44,22 +125,42 @@ class MostroStorage extends BaseStorage<MostroMessage> {
     }
   }
 
-  /// Delete all messages
-  Future<void> 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<void> 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<void> deleteAllMessages() => deleteAll();
+
   /// Delete all messages by Id regardless of type
   Future<void> deleteAllMessagesByOrderId(String orderId) async {
+    await _ensureIndex();
     await deleteWhere(
       Filter.equals('id', orderId),
     );
+    _byOrder.remove(orderId);
+    _notifyOrder(orderId);
   }
 
   /// Filter messages by payload type
@@ -106,63 +207,30 @@ class MostroStorage extends BaseStorage<MostroMessage> {
 
   /// Get the latest message for an order, regardless of type
   Future<MostroMessage?> 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<MostroMessage?> 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<MostroMessage?> watchLatestMessage(String orderId) =>
+      _watchOrder(orderId, () => _latestFor(orderId));
 
   /// Stream of the latest message for an order whose payload is of type T
-  Stream<MostroMessage?> watchLatestMessageOfType<T>(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<MostroMessage?> watchLatestMessageOfType<T>(String orderId) =>
+      _watchOrder(orderId, () {
+        for (final message in _byOrder[orderId] ?? const <MostroMessage>[]) {
+          if (message.payload is T) return message;
         }
-      }
-      return null;
-    });
-  }
+        return null;
+      });
 
-  // Use the same sorting across all methods that return lists of messages
-  List<SortOrder> _getDefaultSort() => [SortOrder('timestamp', false, true)];
+  /// Stream of all messages for an order (newest first)
+  Stream<List<MostroMessage>> watchAllMessages(String orderId) =>
+      _watchOrder(orderId, () => _historyFor(orderId));
 
-  /// Stream of all messages for an order
-  Stream<List<MostroMessage>> 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<SortOrder> _getDefaultSort() => [SortOrder('timestamp', false, true)];
 
   Stream<MostroMessage?> watchByRequestId(int requestId) {
     final query = store.query(
@@ -178,13 +246,14 @@ class MostroStorage extends BaseStorage<MostroMessage> {
   }
 
   Future<List<MostroMessage>> 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<void> dispose() async {
+    _byOrder.clear();
+    await _orderChanges.close();
   }
 }

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: