perf(persistence): Reduce the number of read message per channel from DB when paginating (part 1)

packages/stream_chat_persistence/CHANGELOG.md

@@ -1,3 +1,12 @@
+## Upcoming Changes
+
+🚀 Performance
+
+- Read only the messages matching the `PaginationParams` from DB when calling `MessageDao.getMessagesByCid` instead of reading all messages for the channel and applying pagination in memory.
+- Read only the reactions matching the `userId` from DB when calling `ReactionDao.getReactionsByUserId` instead of reading all reactions for the message and filtering in memory.
+- Read only the reactions matching the `userId` from DB when calling `PinnedMessageReactionDao.getReactionsByUserId` instead of reading all reactions for the message and filtering in memory.
+
+
 ## 9.24.0
 
 - Updated `stream_chat` dependency to [`9.24.0`](https://pub.dev/packages/stream_chat/changelog).

packages/stream_chat_persistence/lib/src/dao/message_dao.dart

@@ -1,5 +1,3 @@
-import 'dart:math';
-
 import 'package:drift/drift.dart';
 import 'package:stream_chat/stream_chat.dart';
 import 'package:stream_chat_persistence/src/db/drift_chat_database.dart';
@@ -171,6 +169,15 @@ class MessageDao extends DatabaseAccessor<DriftChatDatabase>
     bool fetchDraft = true,
     PaginationParams? messagePagination,
   }) async {
+    final lessThanCutoff = await switch (messagePagination?.lessThan) {
+      final id? => _lookupMessageCreatedAt(id),
+      _ => null,
+    };
+    final greaterThanCutoff = await switch (messagePagination?.greaterThan) {
+      final id? => _lookupMessageCreatedAt(id),
+      _ => null,
+    };
+
     final query = select(messages).join([
       leftOuterJoin(_users, messages.userId.equalsExp(_users.id)),
       leftOuterJoin(
@@ -180,44 +187,28 @@ class MessageDao extends DatabaseAccessor<DriftChatDatabase>
     ])
       ..where(messages.channelCid.equals(cid))
       ..where(messages.parentId.isNull() | messages.showInChannel.equals(true))
-      ..orderBy([OrderingTerm.asc(messages.createdAt)]);
+      ..orderBy([
+        OrderingTerm.desc(messages.createdAt),
+        OrderingTerm.desc(messages.id),
+      ]);
 
-    final result = await query.get();
-    if (result.isEmpty) return [];
-
-    final msgList = await Future.wait(
-      result.map(
-        (row) => _messageFromJoinRow(
-          row,
-          fetchDraft: fetchDraft,
-        ),
-      ),
-    );
+    if (lessThanCutoff case final t?) {
+      query.where(messages.createdAt.isSmallerThanValue(t));
+    }
+    if (greaterThanCutoff case final t?) {
+      query.where(messages.createdAt.isBiggerOrEqualValue(t));
+    }
 
-    if (msgList.isNotEmpty) {
-      if (messagePagination?.lessThan != null) {
-        final lessThanIndex = msgList.indexWhere(
-          (m) => m.id == messagePagination!.lessThan,
-        );
-        if (lessThanIndex != -1) {
-          msgList.removeRange(lessThanIndex, msgList.length);
-        }
-      }
-      if (messagePagination?.greaterThan != null) {
-        final greaterThanIndex = msgList.indexWhere(
-          (m) => m.id == messagePagination!.greaterThan,
-        );
-        if (greaterThanIndex != -1) {
-          msgList.removeRange(0, greaterThanIndex);
-        }
-      }
-      if (messagePagination?.limit != null) {
-        return msgList
-            .skip(max(0, msgList.length - messagePagination!.limit))
-            .toList();
-      }
+    if (messagePagination != null) {
+      query.limit(messagePagination.limit);
     }
-    return msgList;
+
+    final rows = (await query.get()).reversed.toList();
+    if (rows.isEmpty) return [];
+
+    return Future.wait(
+      rows.map((row) => _messageFromJoinRow(row, fetchDraft: fetchDraft)),
+    );
   }
 
   /// Updates the message data of a particular channel with
@@ -241,4 +232,16 @@ class MessageDao extends DatabaseAccessor<DriftChatDatabase>
       (batch) => batch.insertAllOnConflictUpdate(messages, entities),
     );
   }
+
+  /// Returns the `createdAt` of the message with [id] in the local cache,
+  /// or `null` if the message isn't cached. Used by [getMessagesByCid] to
+  /// resolve `lessThan` / `greaterThan` pagination cursors into SQL-comparable
+  /// timestamps so the filter runs in SQL instead of after the fact in Dart.
+  Future<DateTime?> _lookupMessageCreatedAt(String id) {
+    return (selectOnly(messages)
+      ..addColumns([messages.createdAt])
+      ..where(messages.id.equals(id)))
+        .map((row) => row.read(messages.createdAt))
+        .getSingleOrNull();
+  }
 }

packages/stream_chat_persistence/lib/src/dao/pinned_message_reaction_dao.dart

@@ -16,17 +16,10 @@ class PinnedMessageReactionDao extends DatabaseAccessor<DriftChatDatabase>
 
   /// Returns all the reactions of a particular message by matching
   /// [Reactions.messageId] with [messageId]
-  Future<List<Reaction>> getReactions(String messageId) =>
-      (select(pinnedMessageReactions).join([
-        leftOuterJoin(users, pinnedMessageReactions.userId.equalsExp(users.id)),
-      ])
-            ..where(pinnedMessageReactions.messageId.equals(messageId))
-            ..orderBy([OrderingTerm.asc(pinnedMessageReactions.createdAt)]))
-          .map((rows) {
-        final userEntity = rows.readTableOrNull(users);
-        final reactionEntity = rows.readTable(pinnedMessageReactions);
-        return reactionEntity.toReaction(user: userEntity?.toUser());
-      }).get();
+  Future<List<Reaction>> getReactions(String messageId) {
+    final where = pinnedMessageReactions.messageId.equals(messageId);
+    return _selectReactions(where);
+  }
 
   /// Returns all the reactions of a particular message
   /// added by a particular user by matching
@@ -35,9 +28,10 @@ class PinnedMessageReactionDao extends DatabaseAccessor<DriftChatDatabase>
   Future<List<Reaction>> getReactionsByUserId(
     String messageId,
     String userId,
-  ) async {
-    final reactions = await getReactions(messageId);
-    return reactions.where((it) => it.userId == userId).toList();
+  ) {
+    final where = pinnedMessageReactions.messageId.equals(messageId) &
+        pinnedMessageReactions.userId.equals(userId);
+    return _selectReactions(where);
   }
 
   /// Updates the reactions data with the new [reactionList] data
@@ -57,4 +51,17 @@ class PinnedMessageReactionDao extends DatabaseAccessor<DriftChatDatabase>
           (r) => r.messageId.isIn(messageIds),
         );
       });
+
+  Future<List<Reaction>> _selectReactions(Expression<bool> where) {
+    final rows = select(pinnedMessageReactions).join([
+      leftOuterJoin(users, pinnedMessageReactions.userId.equalsExp(users.id)),
+    ])
+      ..where(where)
+      ..orderBy([OrderingTerm.asc(pinnedMessageReactions.createdAt)]);
+    return rows.map((row) {
+      final reactionEntity = row.readTable(pinnedMessageReactions);
+      final userEntity = row.readTableOrNull(users);
+      return reactionEntity.toReaction(user: userEntity?.toUser());
+    }).get();
+  }
 }

packages/stream_chat_persistence/lib/src/dao/reaction_dao.dart

@@ -16,17 +16,10 @@ class ReactionDao extends DatabaseAccessor<DriftChatDatabase>
 
   /// Returns all the reactions of a particular message by matching
   /// [Reactions.messageId] with [messageId]
-  Future<List<Reaction>> getReactions(String messageId) =>
-      (select(reactions).join([
-        leftOuterJoin(users, reactions.userId.equalsExp(users.id)),
-      ])
-            ..where(reactions.messageId.equals(messageId))
-            ..orderBy([OrderingTerm.asc(reactions.createdAt)]))
-          .map((rows) {
-        final userEntity = rows.readTableOrNull(users);
-        final reactionEntity = rows.readTable(reactions);
-        return reactionEntity.toReaction(user: userEntity?.toUser());
-      }).get();
+  Future<List<Reaction>> getReactions(String messageId) {
+    final where = reactions.messageId.equals(messageId);
+    return _selectReactions(where);
+  }
 
   /// Returns all the reactions of a particular message
   /// added by a particular user by matching
@@ -35,9 +28,10 @@ class ReactionDao extends DatabaseAccessor<DriftChatDatabase>
   Future<List<Reaction>> getReactionsByUserId(
     String messageId,
     String userId,
-  ) async {
-    final reactions = await getReactions(messageId);
-    return reactions.where((it) => it.userId == userId).toList();
+  ) {
+    final where =
+        reactions.messageId.equals(messageId) & reactions.userId.equals(userId);
+    return _selectReactions(where);
   }
 
   /// Updates the reactions data with the new [reactionList] data
@@ -57,4 +51,16 @@ class ReactionDao extends DatabaseAccessor<DriftChatDatabase>
           (r) => r.messageId.isIn(messageIds),
         );
       });
+
+  Future<List<Reaction>> _selectReactions(Expression<bool> where) {
+    final rows = select(reactions)
+        .join([leftOuterJoin(users, reactions.userId.equalsExp(users.id))])
+      ..where(where)
+      ..orderBy([OrderingTerm.asc(reactions.createdAt)]);
+    return rows.map((row) {
+      final reactionEntity = row.readTable(reactions);
+      final userEntity = row.readTableOrNull(users);
+      return reactionEntity.toReaction(user: userEntity?.toUser());
+    }).get();
+  }
 }