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

lib/data/repositories/mostro_storage.dart

@@ -1,26 +1,119 @@
+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 {
+        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<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 +124,8 @@ class MostroStorage extends BaseStorage<MostroMessage> {
         stackTrace: stack,
       );
       rethrow;
+    } finally {
+      _writesInFlight.remove(id);
     }
   }
 
@@ -44,22 +139,47 @@ 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 {
+    // 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<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(
@@ -173,18 +260,20 @@ class MostroStorage extends BaseStorage<MostroMessage> {
       ),
     );
 
-    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<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();
   }
 }