perf: prune unbounded local storage growth

lib/data/repositories/mostro_storage.dart

@@ -1,26 +1,97 @@
+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();
+  Future<void>? _warmup;
+
+  @visibleForTesting
+  int get debugIndexSize => _byOrder.length;
+
+  /// Order ids currently holding messages (index-backed).
+  Future<List<String>> allOrderIds() async {
+    await _ensureIndex();
+    return _byOrder.keys.toList(growable: false);
+  }
+
+  Future<void> _ensureIndex() {
+    return _warmup ??= () async {
+      final all = await getAll();
+      for (final message in all) {
+        _indexAdd(message, notify: false);
+      }
+      logger.i('Mostro message index warmed: ${_byOrder.length} orders');
+    }();
+  }
+
+  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;
     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}',
       );
@@ -47,7 +118,11 @@ class MostroStorage extends BaseStorage<MostroMessage> {
   /// Delete all messages
   Future<void> deleteAllMessages() async {
     try {
+      await _ensureIndex();
       await 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);
@@ -57,9 +132,12 @@ class MostroStorage extends BaseStorage<MostroMessage> {
 
   /// 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 +184,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 +223,7 @@ class MostroStorage extends BaseStorage<MostroMessage> {
   }
 
   Future<List<MostroMessage>> getAllMessagesForOrderId(String orderId) async {
-    final finder = Finder(
-        filter: Filter.equals('id', orderId),
-        sortOrders: [SortOrder('timestamp', false)]);
-
-    final snapshots = await store.find(db, finder: finder);
-    return snapshots
-        .map((snapshot) => MostroMessage.fromJson(snapshot.value))
-        .toList();
+    await _ensureIndex();
+    return _historyFor(orderId);
   }
 }

lib/data/repositories/notifications_history_repository.dart

@@ -12,6 +12,10 @@ abstract class NotificationsRepository {
   Future<void> markAllAsRead();
   Future<void> deleteNotification(String notificationId);
   Future<void> deleteByOrderId(String orderId);
+
+  /// Delete the given notifications in one transaction. Used to enforce the
+  /// history cap by exact identity, so tied timestamps cannot survive it.
+  Future<int> deleteByIds(Iterable<String> ids);
   Future<void> clearAll();
   Stream<List<NotificationModel>> watchNotifications();
   Future<List<NotificationModel>> getUnreadNotifications();
@@ -118,6 +122,13 @@ class NotificationsStorage extends BaseStorage<NotificationModel>
     await deleteWhere(Filter.equals('orderId', orderId));
   }
 
+  @override
+  Future<int> deleteByIds(Iterable<String> ids) async {
+    final keys = ids.toSet();
+    if (keys.isEmpty) return 0;
+    return deleteWhere(Filter.custom((record) => keys.contains(record.key)));
+  }
+
   @override
   Future<void> clearAll() async {
     await deleteAll();

lib/services/storage_pruner.dart

@@ -0,0 +1,150 @@
+import 'package:mostro_mobile/data/repositories/event_storage.dart';
+import 'package:mostro_mobile/data/repositories/mostro_storage.dart';
+import 'package:mostro_mobile/data/repositories/notifications_history_repository.dart';
+import 'package:mostro_mobile/services/logger_service.dart';
+import 'package:sembast/sembast.dart';
+
+/// Bounded growth for the local databases.
+///
+/// Both Sembast files are fully loaded into RAM and JSON-parsed at every
+/// launch, so unbounded growth taxes every cold start and every store scan.
+/// Three families grew forever: DM reservation records ({id, created_at},
+/// no order_id — unreachable by the session cleanup), chat/message records
+/// for orders whose session is gone, and the notification history. In
+/// "keep forever" session mode nothing pruned at all.
+class StoragePruner {
+  StoragePruner({
+    required this.eventStorage,
+    required this.messageStorage,
+    required this.notificationsStorage,
+  });
+
+  final EventStorage eventStorage;
+  final MostroStorage messageStorage;
+  final NotificationsRepository notificationsStorage;
+
+  /// Floor for reservation retention. It is only a floor: the orders filter
+  /// carries no `since`, so a relay may re-deliver any event a live session's
+  /// trade key matches, and the reservation is the only dedup for Mostro DMs.
+  /// An event for a live session cannot predate that session, so the real
+  /// cutoff is the older of this window and the oldest live session's start.
+  static const Duration reservationRetention = Duration(days: 7);
+
+  /// A full scan of both stores on the main isolate is not worth paying at
+  /// every session-cleanup tick; storage growth is a slow process.
+  static const Duration minimumInterval = Duration(hours: 6);
+
+  /// Orphaned records (no live session) get a grace window before deletion
+  /// so a restore in progress is never raced.
+  static const Duration orphanRetention = Duration(days: 30);
+
+  static const int notificationCap = 300;
+
+  /// Timestamps below this are seconds, above are milliseconds (the boundary
+  /// sits in the year 33658 for seconds and 2001 for milliseconds).
+  static const int _millisecondThreshold = 1000000000000;
+
+  DateTime? _lastRunAt;
+
+  /// Errors propagate: a pruner that silently stopped working would keep the
+  /// growth it exists to bound.
+  Future<void> prune({
+    required Set<String> liveOrderIds,
+    required Set<String> liveDisputeIds,
+    DateTime? oldestLiveSessionAt,
+    DateTime? now,
+  }) async {
+    final at = now ?? DateTime.now();
+    final last = _lastRunAt;
+    if (last != null && at.difference(last) < minimumInterval) return;
+    _lastRunAt = at;
+    await _pruneReservations(at, oldestLiveSessionAt);
+    await _pruneOrphanEvents(at, liveOrderIds, liveDisputeIds);
+    await _pruneOrphanMessages(at, liveOrderIds);
+    await _capNotifications();
+  }
+
+  Future<void> _pruneReservations(
+    DateTime now,
+    DateTime? oldestLiveSessionAt,
+  ) async {
+    var cutoffAt = now.subtract(reservationRetention);
+    if (oldestLiveSessionAt != null &&
+        oldestLiveSessionAt.isBefore(cutoffAt)) {
+      // Keep every reservation a live session could still see replayed.
+      cutoffAt = oldestLiveSessionAt;
+    }
+    final cutoff = cutoffAt.millisecondsSinceEpoch ~/ 1000;
+    final removed = await eventStorage.deleteWhere(Filter.and([
+      Filter.isNull('type'),
+      Filter.lessThan('created_at', cutoff),
+    ]));
+    _logRemoved('DM reservations', removed);
+  }
+
+  Future<void> _pruneOrphanEvents(
+    DateTime now,
+    Set<String> liveOrderIds,
+    Set<String> liveDisputeIds,
+  ) async {
+    final cutoff =
+        now.subtract(orphanRetention).millisecondsSinceEpoch ~/ 1000;
+    final removed = await eventStorage.deleteWhere(Filter.custom((record) {
+      final value = record.value;
+      if (value is! Map) return false;
+      final type = value['type'];
+      if (type != 'chat' && type != 'dispute_chat') return false;
+      final createdAt = value['created_at'];
+      if (createdAt is! int || createdAt >= cutoff) return false;
+      if (type == 'chat') {
+        return !liveOrderIds.contains(value['order_id']);
+      }
+      return !liveDisputeIds.contains(value['dispute_id']);
+    }));
+    _logRemoved('orphaned chat events', removed);
+  }
+
+  Future<void> _pruneOrphanMessages(
+    DateTime now,
+    Set<String> liveOrderIds,
+  ) async {
+    final cutoffMs = now.subtract(orphanRetention).millisecondsSinceEpoch;
+    // Through the storage API so the in-memory index stays coherent.
+    for (final orderId in await messageStorage.allOrderIds()) {
+      if (liveOrderIds.contains(orderId)) continue;
+      final latest = await messageStorage.getLatestMessageById(orderId);
+      // The daemon sends seconds, the app fills in milliseconds when the
+      // field is absent, and both units coexist in the store. An unknown or
+      // non-positive timestamp is never treated as proof of age.
+      final ts = _timestampMs(latest?.timestamp);
+      if (ts != null && ts < cutoffMs) {
+        await messageStorage.deleteAllMessagesByOrderId(orderId);
+        logger.i('Pruned orphaned messages for order $orderId');
+      }
+    }
+  }
+
+  Future<void> _capNotifications() async {
+    final all = await notificationsStorage.getAllNotifications();
+    if (all.length <= notificationCap) return;
+    // Deleting the exact overflow entries rather than everything older than
+    // the cap-th timestamp: a burst of tied timestamps would otherwise leave
+    // the history above the cap on every pass.
+    final sorted = [...all]..sort((a, b) {
+        final byTime = b.timestamp.compareTo(a.timestamp);
+        return byTime != 0 ? byTime : b.id.compareTo(a.id);
+      });
+    final overflow = sorted.skip(notificationCap).map((n) => n.id);
+    final removed = await notificationsStorage.deleteByIds(overflow);
+    _logRemoved('old notifications', removed);
+  }
+
+  static int? _timestampMs(int? raw) {
+    if (raw == null || raw <= 0) return null;
+    return raw < _millisecondThreshold ? raw * 1000 : raw;
+  }
+
+  void _logRemoved(String what, Object? removed) {
+    logger.i('Storage pruner: removed $what ($removed)');
+  }
+}