Merge branch 'refs/heads/feature/FLU-485_optimize_read_message_from_db' into feature/FLU-485_optimize_read_message_from_db_part2

Author: VelikovPetar

packages/stream_chat_persistence/CHANGELOG.md

@@ -8,12 +8,9 @@
 - 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.
 - Improve the message read times from DB.
 
-✅ Added
-
-- `MessageDao.getMessagesByCid` now honours `PaginationParams.lessThanOrEqual` and `PaginationParams.greaterThanOrEqual` (inclusive of the cursor message), in addition to the existing strict `lessThan`/`greaterThan`.
-
 🐞 Fixed
 
+- `MessageDao.getMessagesByCid` now honours `PaginationParams.lessThanOrEqual` and `PaginationParams.greaterThanOrEqual` (inclusive of the cursor message), in addition to the existing strict `lessThan`/`greaterThan`.
 - `MessageDao.getMessagesByCid` now treats `PaginationParams.greaterThan` as strict (exclusive of the cursor), matching the `PaginationParams` contract and the existing `lessThan` behaviour.
 - `MessageDao.getMessagesByCid` with a forward cursor (`greaterThan`/`greaterThanOrEqual`) and a `limit` now returns the messages immediately AFTER the pivot, instead of the channel tail — mirroring how `lessThan` already returned the messages immediately before the pivot.
 

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

@@ -249,27 +249,15 @@ class MessageDao extends DatabaseAccessor<DriftChatDatabase>
     PaginationParams? messagePagination,
   }) async {
     final (
-      lessThanCutoff,
-      lessThanOrEqualCutoff,
-      greaterThanCutoff,
-      greaterThanOrEqualCutoff,
+      lessThanCursor,
+      lessThanOrEqualCursor,
+      greaterThanCursor,
+      greaterThanOrEqualCursor,
     ) = await (
-      switch (messagePagination?.lessThan) {
-        final id? => _lookupMessageCreatedAt(id),
-        _ => Future<DateTime?>.value(),
-      },
-      switch (messagePagination?.lessThanOrEqual) {
-        final id? => _lookupMessageCreatedAt(id),
-        _ => Future<DateTime?>.value(),
-      },
-      switch (messagePagination?.greaterThan) {
-        final id? => _lookupMessageCreatedAt(id),
-        _ => Future<DateTime?>.value(),
-      },
-      switch (messagePagination?.greaterThanOrEqual) {
-        final id? => _lookupMessageCreatedAt(id),
-        _ => Future<DateTime?>.value(),
-      },
+      _lookupCursor(messagePagination?.lessThan),
+      _lookupCursor(messagePagination?.lessThanOrEqual),
+      _lookupCursor(messagePagination?.greaterThan),
+      _lookupCursor(messagePagination?.greaterThanOrEqual),
     ).wait;
 
     // When the caller is paginating forward (greaterThan / greaterThanOrEqual
@@ -278,9 +266,9 @@ class MessageDao extends DatabaseAccessor<DriftChatDatabase>
     // recent (closest to a `lessThan` cursor, or the channel tail when no
     // cursor is set). The final result is always reshaped to ASC for display.
     final isForwardPagination =
-        (greaterThanCutoff != null || greaterThanOrEqualCutoff != null) &&
-            lessThanCutoff == null &&
-            lessThanOrEqualCutoff == null;
+        (greaterThanCursor != null || greaterThanOrEqualCursor != null) &&
+            lessThanCursor == null &&
+            lessThanOrEqualCursor == null;
 
     final orderBy = isForwardPagination
         ? [
@@ -303,17 +291,37 @@ class MessageDao extends DatabaseAccessor<DriftChatDatabase>
       ..where(messages.parentId.isNull() | messages.showInChannel.equals(true))
       ..orderBy(orderBy);
 
-    if (lessThanCutoff case final t?) {
-      query.where(messages.createdAt.isSmallerThanValue(t));
+    // Cursor predicates compare the full `(createdAt, id)` tuple — the same
+    // key used in ORDER BY — so messages sharing a `createdAt` with the cursor
+    // fall on the correct side of the boundary. Filtering on `createdAt` alone
+    // would skip or repeat those siblings across pages.
+    if (lessThanCursor case final c?) {
+      query.where(
+        messages.createdAt.isSmallerThanValue(c.createdAt) |
+            (messages.createdAt.equals(c.createdAt) &
+                messages.id.isSmallerThanValue(c.id)),
+      );
     }
-    if (lessThanOrEqualCutoff case final t?) {
-      query.where(messages.createdAt.isSmallerOrEqualValue(t));
+    if (lessThanOrEqualCursor case final c?) {
+      query.where(
+        messages.createdAt.isSmallerThanValue(c.createdAt) |
+            (messages.createdAt.equals(c.createdAt) &
+                messages.id.isSmallerOrEqualValue(c.id)),
+      );
     }
-    if (greaterThanCutoff case final t?) {
-      query.where(messages.createdAt.isBiggerThanValue(t));
+    if (greaterThanCursor case final c?) {
+      query.where(
+        messages.createdAt.isBiggerThanValue(c.createdAt) |
+            (messages.createdAt.equals(c.createdAt) &
+                messages.id.isBiggerThanValue(c.id)),
+      );
     }
-    if (greaterThanOrEqualCutoff case final t?) {
-      query.where(messages.createdAt.isBiggerOrEqualValue(t));
+    if (greaterThanOrEqualCursor case final c?) {
+      query.where(
+        messages.createdAt.isBiggerThanValue(c.createdAt) |
+            (messages.createdAt.equals(c.createdAt) &
+                messages.id.isBiggerOrEqualValue(c.id)),
+      );
     }
 
     if (messagePagination != null) {
@@ -347,17 +355,21 @@ class MessageDao extends DatabaseAccessor<DriftChatDatabase>
     );
   }
 
-  /// Returns the `createdAt` of the message with [id] in the local cache,
-  /// or `null` if the message isn't cached or isn't visible in the channel
-  /// (i.e. a thread reply with `showInChannel = false`).
-  Future<DateTime?> _lookupMessageCreatedAt(String id) {
-    return (selectOnly(messages)
+  /// Returns the `(createdAt, id)` cursor for the message with [id] in the
+  /// local cache, or `null` if [id] is null, the message isn't cached, or
+  /// isn't visible in the channel (i.e. a thread reply with
+  /// `showInChannel = false`).
+  Future<({DateTime createdAt, String id})?> _lookupCursor(String? id) async {
+    if (id == null) return null;
+    final createdAt = await (selectOnly(messages)
           ..addColumns([messages.createdAt])
           ..where(messages.id.equals(id))
           ..where(
             messages.parentId.isNull() | messages.showInChannel.equals(true),
           ))
         .map((row) => row.read(messages.createdAt))
         .getSingleOrNull();
+    if (createdAt == null) return null;
+    return (createdAt: createdAt, id: id);
   }
 }

packages/stream_chat_persistence/test/src/dao/message_dao_test.dart

@@ -610,6 +610,81 @@ void main() {
       expect(fetchedMessages.first.id, 'testMessageId${cid}5');
       expect(fetchedMessages.last.id, 'testMessageId${cid}14');
     });
+
+    test('cursor with tied createdAt does not skip or duplicate siblings',
+        () async {
+      // Three messages share an identical `createdAt`. The SQL ORDER BY uses
+      // the `(createdAt, id)` tuple, so within the trio the relative order is
+      // by id (lexicographic). A cursor at `msg_tieB` must split the trio
+      // cleanly: `msg_tieA` lands on the "before" side, `msg_tieC` on the
+      // "after" side. A `createdAt`-only WHERE predicate would collapse all
+      // three into the cursor's bucket and drop or keep them together.
+      final users = [User(id: 'tieUser')];
+      await database.userDao.updateUsers(users);
+      await database.channelDao.updateChannels([ChannelModel(cid: cid)]);
+
+      final tie = DateTime.now();
+      final earlier = tie.subtract(const Duration(seconds: 1));
+      final later = tie.add(const Duration(seconds: 1));
+
+      Message m(String id, DateTime t) => Message(
+            id: id,
+            user: users.first,
+            createdAt: t,
+            updatedAt: t,
+            text: id,
+          );
+
+      await messageDao.updateMessages(cid, [
+        m('msg_pre', earlier),
+        m('msg_tieA', tie),
+        m('msg_tieB', tie),
+        m('msg_tieC', tie),
+        m('msg_post', later),
+      ]);
+
+      final before = await messageDao.getMessagesByCid(
+        cid,
+        messagePagination: const PaginationParams(
+          limit: 100,
+          lessThan: 'msg_tieB',
+        ),
+      );
+      expect(before.map((m) => m.id).toList(), ['msg_pre', 'msg_tieA']);
+
+      final after = await messageDao.getMessagesByCid(
+        cid,
+        messagePagination: const PaginationParams(
+          limit: 100,
+          greaterThan: 'msg_tieB',
+        ),
+      );
+      expect(after.map((m) => m.id).toList(), ['msg_tieC', 'msg_post']);
+
+      final atOrBefore = await messageDao.getMessagesByCid(
+        cid,
+        messagePagination: const PaginationParams(
+          limit: 100,
+          lessThanOrEqual: 'msg_tieB',
+        ),
+      );
+      expect(
+        atOrBefore.map((m) => m.id).toList(),
+        ['msg_pre', 'msg_tieA', 'msg_tieB'],
+      );
+
+      final atOrAfter = await messageDao.getMessagesByCid(
+        cid,
+        messagePagination: const PaginationParams(
+          limit: 100,
+          greaterThanOrEqual: 'msg_tieB',
+        ),
+      );
+      expect(
+        atOrAfter.map((m) => m.id).toList(),
+        ['msg_tieB', 'msg_tieC', 'msg_post'],
+      );
+    });
   });
 
   test('updateMessages', () async {