feat(persistence): add support for DraftMessages

packages/stream_chat/lib/src/db/chat_persistence_client.dart

@@ -4,6 +4,7 @@ import 'package:stream_chat/src/core/api/sort_order.dart';
 import 'package:stream_chat/src/core/models/attachment_file.dart';
 import 'package:stream_chat/src/core/models/channel_model.dart';
 import 'package:stream_chat/src/core/models/channel_state.dart';
+import 'package:stream_chat/src/core/models/draft.dart';
 import 'package:stream_chat/src/core/models/event.dart';
 import 'package:stream_chat/src/core/models/filter.dart';
 import 'package:stream_chat/src/core/models/member.dart';
@@ -78,32 +79,40 @@ abstract class ChatPersistenceClient {
     PaginationParams? messagePagination,
   });
 
+  /// Get stored [Draft] message by providing channel [cid].
+  Future<Draft?> getDraftMessageByCid(String cid);
+
+  /// Get stored [Draft] message by providing parent message [id].
+  Future<Draft?> getDraftMessageByParentId(String parentId);
+
   /// Get [ChannelState] data by providing channel [cid]
   Future<ChannelState> getChannelStateByCid(
     String cid, {
     PaginationParams? messagePagination,
     PaginationParams? pinnedMessagePagination,
   }) async {
-    final data = await Future.wait([
+    final (members, reads, channel, messages, pinnedMessages, draft) = await (
       getMembersByCid(cid),
       getReadsByCid(cid),
       getChannelByCid(cid),
       getMessagesByCid(cid, messagePagination: messagePagination),
       getPinnedMessagesByCid(cid, messagePagination: pinnedMessagePagination),
-    ]);
+      getDraftMessageByCid(cid),
+    ).wait;
 
-    final members = data[0] as List<Member>?;
-    final membership = userId == null
-        ? null
-        : members?.firstWhereOrNull((it) => it.userId == userId);
+    final membership = switch (userId) {
+      final userId? => members?.firstWhereOrNull((it) => it.userId == userId),
+      _ => null,
+    };
 
     return ChannelState(
       members: members,
       membership: membership,
-      read: data[1] as List<Read>?,
-      channel: data[2] as ChannelModel?,
-      messages: data[3] as List<Message>?,
-      pinnedMessages: data[4] as List<Message>?,
+      read: reads,
+      channel: channel,
+      messages: messages,
+      pinnedMessages: pinnedMessages,
+      draft: draft,
     );
   }
 
@@ -157,6 +166,9 @@ abstract class ChatPersistenceClient {
   /// Remove a channel by [channelId]
   Future<void> deleteChannels(List<String> cids);
 
+  /// Removes all the draft messages by draft [messageIds]
+  Future<void> deleteDraftMessagesByIds(List<String> messageIds);
+
   /// Updates the message data of a particular channel [cid] with
   /// the new [messages] data
   Future<void> updateMessages(String cid, List<Message> messages) =>
@@ -214,6 +226,9 @@ abstract class ChatPersistenceClient {
   /// Updates the poll votes data with the new [pollVotes] data
   Future<void> updatePollVotes(List<PollVote> pollVotes);
 
+  /// Updates the draft messages data with the new [draftMessages] data
+  Future<void> updateDraftMessages(List<Draft> draftMessages);
+
   /// Deletes all the reactions by [messageIds]
   Future<void> deleteReactionsByMessageId(List<String> messageIds);
 
@@ -272,6 +287,9 @@ abstract class ChatPersistenceClient {
     final pollVotes = <PollVote>[];
     final pollVotesToDelete = <String>[];
 
+    final drafts = <Draft>[];
+    final draftsToDelete = <String>[];
+
     for (final state in channelStates) {
       final channel = state.channel;
       // Continue if channel is not available.
@@ -315,6 +333,14 @@ abstract class ChatPersistenceClient {
 
       pollVotes.addAll(polls.expand(_expandPollVotes));
 
+      drafts.addAll([
+        state.draft,
+        ...?messages?.map((it) => it.draft),
+        ...?pinnedMessages?.map((it) => it.draft),
+      ].nonNulls);
+
+      draftsToDelete.addAll(drafts.map((it) => it.message.id));
+
       users.addAll([
         channel.createdBy,
         ...?messages?.map((it) => it.user),
@@ -335,6 +361,7 @@ abstract class ChatPersistenceClient {
       deleteReactionsByMessageId(reactionsToDelete),
       deletePinnedMessageReactionsByMessageId(pinnedReactionsToDelete),
       deletePollVotesByPollIds(pollVotesToDelete),
+      deleteDraftMessagesByIds(draftsToDelete),
     ]);
 
     // Updating first as does not depend on any other table.
@@ -357,6 +384,7 @@ abstract class ChatPersistenceClient {
       updateReactions(reactions),
       updatePinnedMessageReactions(pinnedReactions),
       updatePollVotes(pollVotes),
+      updateDraftMessages(drafts),
     ]);
   }
 

packages/stream_chat/test/src/db/chat_persistence_client_test.dart

@@ -2,6 +2,8 @@ import 'package:stream_chat/src/core/api/requests.dart';
 import 'package:stream_chat/src/core/api/sort_order.dart';
 import 'package:stream_chat/src/core/models/channel_model.dart';
 import 'package:stream_chat/src/core/models/channel_state.dart';
+import 'package:stream_chat/src/core/models/draft.dart';
+import 'package:stream_chat/src/core/models/draft_message.dart';
 import 'package:stream_chat/src/core/models/event.dart';
 import 'package:stream_chat/src/core/models/filter.dart';
 import 'package:stream_chat/src/core/models/member.dart';
@@ -55,6 +57,10 @@ class TestPersistenceClient extends ChatPersistenceClient {
   @override
   Future<void> deletePollVotesByPollIds(List<String> pollIds) => Future.value();
 
+  @override
+  Future<void> deleteDraftMessagesByIds(List<String> messageIds) =>
+      Future.value();
+
   @override
   Future<void> disconnect({bool flush = false}) => throw UnimplementedError();
 
@@ -98,6 +104,21 @@ class TestPersistenceClient extends ChatPersistenceClient {
   @override
   Future<List<Read>> getReadsByCid(String cid) async => [];
 
+  @override
+  Future<Draft?> getDraftMessageByCid(String cid) async => Draft(
+        channelCid: cid,
+        createdAt: DateTime.now(),
+        message: DraftMessage(id: 'message-id', text: 'message-text'),
+      );
+
+  @override
+  Future<Draft?> getDraftMessageByParentId(String parentId) async => Draft(
+        channelCid: 'test:cid',
+        createdAt: DateTime.now(),
+        parentId: parentId,
+        message: DraftMessage(id: 'message-id', text: 'message-text'),
+      );
+
   @override
   Future<List<Message>> getReplies(String parentId,
           {PaginationParams? options}) =>
@@ -152,6 +173,9 @@ class TestPersistenceClient extends ChatPersistenceClient {
 
   @override
   Future<void> updatePolls(List<Poll> polls) => Future.value();
+
+  @override
+  Future<void> updateDraftMessages(List<Draft> draftMessages) => Future.value();
 }
 
 void main() {
@@ -194,6 +218,20 @@ void main() {
       persistenceClient.updatePolls([poll]);
     });
 
+    test('deleteDraftMessagesByIds', () {
+      const messageIds = ['message-id'];
+      persistenceClient.deleteDraftMessagesByIds(messageIds);
+    });
+
+    test('updateDraftMessages', () async {
+      final draft = Draft(
+        channelCid: 'test:cid',
+        createdAt: DateTime.now(),
+        message: DraftMessage(id: 'message-id', text: 'message-text'),
+      );
+      persistenceClient.updateDraftMessages([draft]);
+    });
+
     test('updateChannelThreads', () async {
       const cid = 'test:cid';
       final user = User(id: 'test-user-id');

packages/stream_chat_persistence/CHANGELOG.md

@@ -1,6 +1,7 @@
 ## Upcoming
 
 - Added `pinnedAt` and `archivedAt` fields on `Member`.
+- Added support for DraftMessages.
 
 ## 9.7.0
 

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

@@ -1,6 +1,7 @@
 export 'channel_dao.dart';
 export 'channel_query_dao.dart';
 export 'connection_event_dao.dart';
+export 'draft_message_dao.dart';
 export 'member_dao.dart';
 export 'message_dao.dart';
 export 'pinned_message_dao.dart';

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

@@ -0,0 +1,109 @@
+import 'package:drift/drift.dart';
+import 'package:stream_chat/stream_chat.dart';
+import 'package:stream_chat_persistence/src/db/drift_chat_database.dart';
+import 'package:stream_chat_persistence/src/entity/entity.dart';
+import 'package:stream_chat_persistence/src/mapper/mapper.dart';
+
+part 'draft_message_dao.g.dart';
+
+/// The Data Access Object for operations in [DraftMessages] table.
+@DriftAccessor(tables: [DraftMessages, Messages])
+class DraftMessageDao extends DatabaseAccessor<DriftChatDatabase>
+    with _$DraftMessageDaoMixin {
+  /// Creates a new draft message dao instance
+  DraftMessageDao(this._db) : super(_db);
+
+  final DriftChatDatabase _db;
+
+  Future<Draft> _draftFromEntity(DraftMessageEntity entity) async {
+    // We do not want to fetch the draft message of the parent and quoted
+    // message because it will create a circular dependency and will
+    // result in infinite loop.
+    const fetchDraft = false;
+
+    final parentMessage = await switch (entity.parentId) {
+      final id? => _db.messageDao.getMessageById(id, fetchDraft: fetchDraft),
+      _ => null,
+    };
+
+    final quotedMessage = await switch (entity.quotedMessageId) {
+      final id? => _db.messageDao.getMessageById(id, fetchDraft: fetchDraft),
+      _ => null,
+    };
+
+    final poll = await switch (entity.pollId) {
+      final id? => _db.pollDao.getPollById(id),
+      _ => null,
+    };
+
+    return entity.toDraft(
+      parentMessage: parentMessage,
+      quotedMessage: quotedMessage,
+      poll: poll,
+    );
+  }
+
+  /// Returns the draft message by matching [DraftMessages.id] with [id]
+  Future<Draft?> getDraftMessageById(String id) async {
+    final query = select(draftMessages)..where((tbl) => tbl.id.equals(id));
+
+    final result = await query.getSingleOrNull();
+    if (result == null) return null;
+
+    return _draftFromEntity(result);
+  }
+
+  /// Returns the draft message by matching [DraftMessages.channelCid].
+  ///
+  /// Note: This will skip the thread draft messages.
+  Future<Draft?> getDraftMessageByCid(String cid) async {
+    final query = select(draftMessages)
+      ..where((tbl) => tbl.channelCid.equals(cid) & tbl.parentId.isNull());
+
+    final result = await query.getSingleOrNull();
+    if (result == null) return null;
+
+    return _draftFromEntity(result);
+  }
+
+  /// Returns the draft message by matching [DraftMessages.parentId].
+  Future<Draft?> getDraftMessageByParentId(String parentId) async {
+    final query = select(draftMessages)
+      ..where((tbl) => tbl.parentId.equals(parentId));
+
+    final result = await query.getSingleOrNull();
+    if (result == null) return null;
+
+    return _draftFromEntity(result);
+  }
+
+  /// Updates the draft message data of a particular channel with
+  /// the new [messageList] data.
+  Future<void> updateDraftMessages(List<Draft> draftMessageList) {
+    return transaction(() async {
+      for (final draftMessage in draftMessageList) {
+        final entity = draftMessage.toEntity();
+
+        // Find and delete existing drafts with the same channelCid
+        // and parentId (if any).
+        final deleteQuery = delete(draftMessages)
+          ..where((tbl) {
+            var filter = tbl.channelCid.equals(entity.channelCid);
+            if (entity.parentId case final parentId?) {
+              filter &= tbl.parentId.equals(parentId);
+            }
+
+            return filter;
+          });
+
+        await deleteQuery.go();
+        await into(draftMessages).insertOnConflictUpdate(entity);
+      }
+    });
+  }
+
+  /// Deletes all the draft messages whose [DraftMessages.id] is present in
+  /// [messageIds].
+  Future<void> deleteDraftMessagesByIds(List<String> messageIds) =>
+      (delete(draftMessages)..where((tbl) => tbl.id.isIn(messageIds))).go();
+}