fix: updateChannelStates invocation sequence as per foreign keys relations.

Signed-off-by: xsahil03x <[email protected]>

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

@@ -143,11 +143,19 @@ abstract class ChatPersistenceClient {
 
   /// Updates the message data of a particular channel [cid] with
   /// the new [messages] data
-  Future<void> updateMessages(String cid, List<Message> messages);
+  Future<void> updateMessages(String cid, List<Message> messages) =>
+      bulkUpdateMessages({cid: messages});
+
+  /// Bulk updates the message data of multiple channels.
+  Future<void> bulkUpdateMessages(Map<String, List<Message>> messages);
 
   /// Updates the pinned message data of a particular channel [cid] with
   /// the new [messages] data
-  Future<void> updatePinnedMessages(String cid, List<Message> messages);
+  Future<void> updatePinnedMessages(String cid, List<Message> messages) =>
+      bulkUpdatePinnedMessages({cid: messages});
+
+  /// Bulk updates the message data of multiple channels.
+  Future<void> bulkUpdatePinnedMessages(Map<String, List<Message>> messages);
 
   /// Returns all the threads by parent message of a particular channel by
   /// providing channel [cid]
@@ -158,11 +166,19 @@ abstract class ChatPersistenceClient {
 
   /// Updates all the members of a particular channle [cid]
   /// with the new [members] data
-  Future<void> updateMembers(String cid, List<Member> members);
+  Future<void> updateMembers(String cid, List<Member> members) =>
+      bulkUpdateMembers({cid: members});
+
+  /// Bulk updates the members data of multiple channels.
+  Future<void> bulkUpdateMembers(Map<String, List<Member>> members);
 
   /// Updates the read data of a particular channel [cid] with
   /// the new [reads] data
-  Future<void> updateReads(String cid, List<Read> reads);
+  Future<void> updateReads(String cid, List<Read> reads) =>
+      bulkUpdateReads({cid: reads});
+
+  /// Bulk updates the read data of multiple channels.
+  Future<void> bulkUpdateReads(Map<String, List<Read>> reads);
 
   /// Updates the users data with the new [users] data
   Future<void> updateUsers(List<User> users);
@@ -188,104 +204,91 @@ abstract class ChatPersistenceClient {
 
   /// Update list of channel states
   Future<void> updateChannelStates(List<ChannelState> channelStates) async {
-    final deleteReactions = deleteReactionsByMessageId(channelStates
+    final reactionsToDelete = <String>[];
-        .expand((it) => it.messages)
+    final pinnedReactionsToDelete = <String>[];
-        .map((m) => m.id)
+    final membersToDelete = <String>[];
-        .toList(growable: false));
 
-    final deletePinnedMessageReactions =
+    final channels = <ChannelModel>[];
-        deletePinnedMessageReactionsByMessageId(channelStates
+    final channelWithMessages = <String, List<Message>>{};
-            .expand((it) => it.pinnedMessages)
+    final channelWithPinnedMessages = <String, List<Message>>{};
-            .map((m) => m.id)
+    final channelWithReads = <String, List<Read>>{};
-            .toList(growable: false));
+    final channelWithMembers = <String, List<Member>>{};
 
-    final cleanedChannelStates =
+    final users = <User>[];
-        channelStates.where((it) => it.channel != null);
+    final reactions = <Reaction>[];
+    final pinnedReactions = <Reaction>[];
 
-    final deleteMembers = deleteMembersByCids(
+    for (final state in channelStates) {
-      cleanedChannelStates.map((it) => it.channel!.cid).toList(growable: false),
+      final channel = state.channel;
-    );
+      if (channel != null) {
+        channels.add(channel);
 
+        final cid = channel.cid;
+        final reads = state.read;
+        final members = state.members;
+        final messages = state.messages;
+        final pinnedMessages = state.pinnedMessages;
+
+        // Preparing deletion data
+        membersToDelete.add(cid);
+        reactionsToDelete.addAll(state.messages.map((it) => it.id));
+        pinnedReactionsToDelete.addAll(state.pinnedMessages.map((it) => it.id));
+
+        // preparing addition data
+        channelWithReads[cid] = reads;
+        channelWithMembers[cid] = members;
+        channelWithMessages[cid] = messages;
+        channelWithPinnedMessages[cid] = pinnedMessages;
+
+        List<Reaction> expandReactions(Message message) {
+          final own = message.ownReactions;
+          final latest = message.latestReactions;
+          return [
+            if (own != null) ...own.where((r) => r.userId != null),
+            if (latest != null) ...latest.where((r) => r.userId != null),
+          ];
+        }
+
+        reactions.addAll(messages.expand(expandReactions));
+        pinnedReactions.addAll(pinnedMessages.expand(expandReactions));
+
+        users.addAll([
+          channel.createdBy,
+          ...reads.map((it) => it.user),
+          ...members.map((it) => it.user),
+          ...reactions.map((it) => it.user),
+          ...pinnedReactions.map((it) => it.user),
+        ].withNullifyer);
+      }
+    }
+
+    // Removing old members and reactions data as they may have
+    // changes over the time.
     await Future.wait([
-      deleteReactions,
+      deleteMembersByCids(membersToDelete),
-      deletePinnedMessageReactions,
+      deleteReactionsByMessageId(reactionsToDelete),
-      deleteMembers,
+      deletePinnedMessageReactionsByMessageId(pinnedReactionsToDelete),
     ]);
 
-    final channels = cleanedChannelStates.map((it) => it.channel).withNullifyer;
+    // Updating first as does not depend on any other table.
-
-    final reactions = cleanedChannelStates
-        .expand((it) => it.messages)
-        .expand((it) => [
-              if (it.ownReactions != null)
-                ...it.ownReactions!.where((r) => r.userId != null),
-              if (it.latestReactions != null)
-                ...it.latestReactions!.where((r) => r.userId != null),
-            ])
-        .withNullifyer;
-
-    final pinnedMessageReactions = cleanedChannelStates
-        .expand((it) => it.pinnedMessages)
-        .expand((it) => [
-              if (it.ownReactions != null)
-                ...it.ownReactions!.where((r) => r.userId != null),
-              if (it.latestReactions != null)
-                ...it.latestReactions!.where((r) => r.userId != null),
-            ])
-        .withNullifyer;
-
-    final users = cleanedChannelStates
-        .map((cs) => [
-              cs.channel?.createdBy,
-              ...cs.messages
-                  .map((m) => [
-                        m.user,
-                        if (m.latestReactions != null)
-                          ...m.latestReactions!.map((r) => r.user),
-                        if (m.ownReactions != null)
-                          ...m.ownReactions!.map((r) => r.user),
-                      ])
-                  .expand((v) => v),
-              ...cs.read.map((r) => r.user),
-              ...cs.members.map((m) => m.user),
-            ])
-        .expand((it) => it)
-        .withNullifyer;
-
-    final updateMessagesFuture = cleanedChannelStates.map((it) {
-      final cid = it.channel!.cid;
-      final messages = it.messages;
-      return updateMessages(cid, messages.toList(growable: false));
-    }).toList(growable: false);
-
-    final updatePinnedMessagesFuture = cleanedChannelStates.map((it) {
-      final cid = it.channel!.cid;
-      final messages = it.pinnedMessages;
-      return updatePinnedMessages(cid, messages.toList(growable: false));
-    }).toList(growable: false);
-
-    final updateReadsFuture = cleanedChannelStates.map((it) {
-      final cid = it.channel!.cid;
-      final reads = it.read;
-      return updateReads(cid, reads.toList(growable: false));
-    }).toList(growable: false);
-
-    final updateMembersFuture = cleanedChannelStates.map((it) {
-      final cid = it.channel!.cid;
-      final members = it.members;
-      return updateMembers(cid, members.toList(growable: false));
-    }).toList(growable: false);
-
     await Future.wait([
-      ...updateMessagesFuture,
-      ...updatePinnedMessagesFuture,
-      ...updateReadsFuture,
-      ...updateMembersFuture,
       updateUsers(users.toList(growable: false)),
       updateChannels(channels.toList(growable: false)),
+    ]);
+
+    // All has a foreign key relation with channels table.
+    await Future.wait([
+      bulkUpdateReads(channelWithReads),
+      bulkUpdateMembers(channelWithMembers),
+      bulkUpdateMessages(channelWithMessages),
+      bulkUpdatePinnedMessages(channelWithPinnedMessages),
+    ]);
+
+    // Both has a foreign key relation with messages, pinnedMessages table.
+    await Future.wait([
       updateReactions(reactions.toList(growable: false)),
       updatePinnedMessageReactions(
-        pinnedMessageReactions.toList(growable: false),
+        pinnedReactions.toList(growable: false),
       ),
     ]);
   }

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

@@ -106,18 +106,6 @@ class TestPersistenceClient extends ChatPersistenceClient {
   Future<void> updateLastSyncAt(DateTime lastSyncAt) =>
       throw UnimplementedError();
 
-  @override
-  Future<void> updateMembers(String cid, List<Member> members) =>
-      Future.value();
-
-  @override
-  Future<void> updateMessages(String cid, List<Message> messages) =>
-      Future.value();
-
-  @override
-  Future<void> updatePinnedMessages(String cid, List<Message> messages) =>
-      Future.value();
-
   @override
   Future<void> updateReactions(List<Reaction> reactions) => Future.value();
 
@@ -126,10 +114,22 @@ class TestPersistenceClient extends ChatPersistenceClient {
       Future.value();
 
   @override
-  Future<void> updateReads(String cid, List<Read> reads) => Future.value();
+  Future<void> updateUsers(List<User> users) => Future.value();
 
   @override
-  Future<void> updateUsers(List<User> users) => Future.value();
+  Future<void> bulkUpdateMembers(Map<String, List<Member>> members) =>
+      Future.value();
+
+  @override
+  Future<void> bulkUpdateMessages(Map<String, List<Message>> messages) =>
+      Future.value();
+
+  @override
+  Future<void> bulkUpdatePinnedMessages(Map<String, List<Message>> messages) =>
+      Future.value();
+
+  @override
+  Future<void> bulkUpdateReads(Map<String, List<Read>> reads) => Future.value();
 }
 
 void main() {

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

@@ -30,13 +30,19 @@ class MemberDao extends DatabaseAccessor<MoorChatDatabase>
       }).get();
 
   /// Updates all the members using the new [memberList] data
-  Future<void> updateMembers(String cid, List<Member> memberList) async =>
+  Future<void> updateMembers(String cid, List<Member> memberList) =>
-      batch(
+      bulkUpdateMembers({cid: memberList});
-        (it) => it.insertAllOnConflictUpdate(
+
-          members,
+  /// Bulk updates the members data of multiple channels
-          memberList.map((m) => m.toEntity(cid: cid)).toList(),
+  Future<void> bulkUpdateMembers(Map<String, List<Member>> channelWithMembers) {
-        ),
+    final entities = channelWithMembers.entries
-      );
+        .map((entry) => entry.value.map(
+              (member) => member.toEntity(cid: entry.key),
+            ))
+        .expand((it) => it)
+        .toList(growable: false);
+    return batch((batch) => batch.insertAllOnConflictUpdate(members, entities));
+  }
 
   /// Deletes all the members whose [Members.channelCid] is present in [cids]
   Future<void> deleteMemberByCids(List<String> cids) async => batch((it) {

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

@@ -169,12 +169,21 @@ class MessageDao extends DatabaseAccessor<MoorChatDatabase>
 
   /// Updates the message data of a particular channel with
   /// the new [messageList] data
-  Future<void> updateMessages(String cid, List<Message> messageList) => batch(
+  Future<void> updateMessages(String cid, List<Message> messageList) =>
-        (batch) {
+      bulkUpdateMessages({cid: messageList});
-          batch.insertAllOnConflictUpdate(
+
-            messages,
+  /// Bulk updates the message data of multiple channels
-            messageList.map((it) => it.toEntity(cid: cid)).toList(),
+  Future<void> bulkUpdateMessages(
-          );
+    Map<String, List<Message>> channelWithMessages,
-        },
+  ) {
-      );
+    final entities = channelWithMessages.entries
+        .map((entry) => entry.value.map(
+              (message) => message.toEntity(cid: entry.key),
+            ))
+        .expand((it) => it)
+        .toList(growable: false);
+    return batch(
+      (batch) => batch.insertAllOnConflictUpdate(messages, entities),
+    );
+  }
 }

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

@@ -170,12 +170,21 @@ class PinnedMessageDao extends DatabaseAccessor<MoorChatDatabase>
 
   /// Updates the message data of a particular channel with
   /// the new [messageList] data
-  Future<void> updateMessages(String cid, List<Message> messageList) => batch(
+  Future<void> updateMessages(String cid, List<Message> messageList) =>
-        (batch) {
+      bulkUpdateMessages({cid: messageList});
-          batch.insertAllOnConflictUpdate(
+
-            pinnedMessages,
+  /// Bulk updates the message data of multiple channels
-            messageList.map((it) => it.toPinnedEntity(cid: cid)).toList(),
+  Future<void> bulkUpdateMessages(
-          );
+    Map<String, List<Message>> channelWithMessages,
-        },
+  ) {
-      );
+    final entities = channelWithMessages.entries
+        .map((entry) => entry.value.map(
+              (message) => message.toPinnedEntity(cid: entry.key),
+            ))
+        .expand((it) => it)
+        .toList(growable: false);
+    return batch(
+      (batch) => batch.insertAllOnConflictUpdate(pinnedMessages, entities),
+    );
+  }
 }

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

@@ -29,10 +29,17 @@ class ReadDao extends DatabaseAccessor<MoorChatDatabase> with _$ReadDaoMixin {
 
   /// Updates the read data of a particular channel with
   /// the new [readList] data
-  Future<void> updateReads(String cid, List<Read> readList) => batch(
+  Future<void> updateReads(String cid, List<Read> readList) =>
-        (it) => it.insertAllOnConflictUpdate(
+      bulkUpdateReads({cid: readList});
-          reads,
+
-          readList.map((r) => r.toEntity(cid: cid)).toList(),
+  /// Bulk updates the reads data of multiple channels
-        ),
+  Future<void> bulkUpdateReads(Map<String, List<Read>> channelWithReads) {
-      );
+    final entities = channelWithReads.entries
+        .map((entry) => entry.value.map(
+              (read) => read.toEntity(cid: entry.key),
+            ))
+        .expand((it) => it)
+        .toList(growable: false);
+    return batch((batch) => batch.insertAllOnConflictUpdate(reads, entities));
+  }
 }

packages/stream_chat_persistence/lib/src/db/moor_chat_database.dart

@@ -58,7 +58,7 @@ class MoorChatDatabase extends _$MoorChatDatabase {
   @override
   MigrationStrategy get migration => MigrationStrategy(
         beforeOpen: (details) async {
-          await customStatement('PRAGMA foreign_keys = ON;');
+          await customStatement('PRAGMA foreign_keys = ON');
         },
         onUpgrade: (openingDetails, before, after) async {
           if (before != after) {

packages/stream_chat_persistence/lib/src/db/moor_chat_database.g.dart

@@ -645,7 +645,7 @@ class MessageEntity extends DataClass implements Insertable<MessageEntity> {
   final String? pinnedByUserId;
 
   /// The channel cid of which this message is part of
-  final String? channelCid;
+  final String channelCid;
 
   /// A Map of [messageText] translations.
   final Map<String, String>? i18n;
@@ -675,7 +675,7 @@ class MessageEntity extends DataClass implements Insertable<MessageEntity> {
       this.pinnedAt,
       this.pinExpires,
       this.pinnedByUserId,
-      this.channelCid,
+      required this.channelCid,
       this.i18n,
       this.extraData});
   factory MessageEntity.fromData(
@@ -728,7 +728,7 @@ class MessageEntity extends DataClass implements Insertable<MessageEntity> {
       pinnedByUserId: const StringType()
           .mapFromDatabaseResponse(data['${effectivePrefix}pinned_by_user_id']),
       channelCid: const StringType()
-          .mapFromDatabaseResponse(data['${effectivePrefix}channel_cid']),
+          .mapFromDatabaseResponse(data['${effectivePrefix}channel_cid'])!,
       i18n: $MessagesTable.$converter5.mapToDart(const StringType()
           .mapFromDatabaseResponse(data['${effectivePrefix}i18n'])),
       extraData: $MessagesTable.$converter6.mapToDart(const StringType()
@@ -800,9 +800,7 @@ class MessageEntity extends DataClass implements Insertable<MessageEntity> {
     if (!nullToAbsent || pinnedByUserId != null) {
       map['pinned_by_user_id'] = Variable<String?>(pinnedByUserId);
     }
-    if (!nullToAbsent || channelCid != null) {
+    map['channel_cid'] = Variable<String>(channelCid);
-      map['channel_cid'] = Variable<String?>(channelCid);
-    }
     if (!nullToAbsent || i18n != null) {
       final converter = $MessagesTable.$converter5;
       map['i18n'] = Variable<String?>(converter.mapToSql(i18n));
@@ -842,7 +840,7 @@ class MessageEntity extends DataClass implements Insertable<MessageEntity> {
       pinnedAt: serializer.fromJson<DateTime?>(json['pinnedAt']),
       pinExpires: serializer.fromJson<DateTime?>(json['pinExpires']),
       pinnedByUserId: serializer.fromJson<String?>(json['pinnedByUserId']),
-      channelCid: serializer.fromJson<String?>(json['channelCid']),
+      channelCid: serializer.fromJson<String>(json['channelCid']),
       i18n: serializer.fromJson<Map<String, String>?>(json['i18n']),
       extraData: serializer.fromJson<Map<String, Object?>?>(json['extraData']),
     );
@@ -873,7 +871,7 @@ class MessageEntity extends DataClass implements Insertable<MessageEntity> {
       'pinnedAt': serializer.toJson<DateTime?>(pinnedAt),
       'pinExpires': serializer.toJson<DateTime?>(pinExpires),
       'pinnedByUserId': serializer.toJson<String?>(pinnedByUserId),
-      'channelCid': serializer.toJson<String?>(channelCid),
+      'channelCid': serializer.toJson<String>(channelCid),
       'i18n': serializer.toJson<Map<String, String>?>(i18n),
       'extraData': serializer.toJson<Map<String, Object?>?>(extraData),
     };
@@ -902,7 +900,7 @@ class MessageEntity extends DataClass implements Insertable<MessageEntity> {
           Value<DateTime?> pinnedAt = const Value.absent(),
           Value<DateTime?> pinExpires = const Value.absent(),
           Value<String?> pinnedByUserId = const Value.absent(),
-          Value<String?> channelCid = const Value.absent(),
+          String? channelCid,
           Value<Map<String, String>?> i18n = const Value.absent(),
           Value<Map<String, Object?>?> extraData = const Value.absent()}) =>
       MessageEntity(
@@ -934,7 +932,7 @@ class MessageEntity extends DataClass implements Insertable<MessageEntity> {
         pinExpires: pinExpires.present ? pinExpires.value : this.pinExpires,
         pinnedByUserId:
             pinnedByUserId.present ? pinnedByUserId.value : this.pinnedByUserId,
-        channelCid: channelCid.present ? channelCid.value : this.channelCid,
+        channelCid: channelCid ?? this.channelCid,
         i18n: i18n.present ? i18n.value : this.i18n,
         extraData: extraData.present ? extraData.value : this.extraData,
       );
@@ -1068,7 +1066,7 @@ class MessagesCompanion extends UpdateCompanion<MessageEntity> {
   final Value<DateTime?> pinnedAt;
   final Value<DateTime?> pinExpires;
   final Value<String?> pinnedByUserId;
-  final Value<String?> channelCid;
+  final Value<String> channelCid;
   final Value<Map<String, String>?> i18n;
   final Value<Map<String, Object?>?> extraData;
   const MessagesCompanion({
@@ -1121,12 +1119,13 @@ class MessagesCompanion extends UpdateCompanion<MessageEntity> {
     this.pinnedAt = const Value.absent(),
     this.pinExpires = const Value.absent(),
     this.pinnedByUserId = const Value.absent(),
-    this.channelCid = const Value.absent(),
+    required String channelCid,
     this.i18n = const Value.absent(),
     this.extraData = const Value.absent(),
   })  : id = Value(id),
         attachments = Value(attachments),
-        mentionedUsers = Value(mentionedUsers);
+        mentionedUsers = Value(mentionedUsers),
+        channelCid = Value(channelCid);
   static Insertable<MessageEntity> custom({
     Expression<String>? id,
     Expression<String?>? messageText,
@@ -1150,7 +1149,7 @@ class MessagesCompanion extends UpdateCompanion<MessageEntity> {
     Expression<DateTime?>? pinnedAt,
     Expression<DateTime?>? pinExpires,
     Expression<String?>? pinnedByUserId,
-    Expression<String?>? channelCid,
+    Expression<String>? channelCid,
     Expression<Map<String, String>?>? i18n,
     Expression<Map<String, Object?>?>? extraData,
   }) {
@@ -1206,7 +1205,7 @@ class MessagesCompanion extends UpdateCompanion<MessageEntity> {
       Value<DateTime?>? pinnedAt,
       Value<DateTime?>? pinExpires,
       Value<String?>? pinnedByUserId,
-      Value<String?>? channelCid,
+      Value<String>? channelCid,
       Value<Map<String, String>?>? i18n,
       Value<Map<String, Object?>?>? extraData}) {
     return MessagesCompanion(
@@ -1317,7 +1316,7 @@ class MessagesCompanion extends UpdateCompanion<MessageEntity> {
       map['pinned_by_user_id'] = Variable<String?>(pinnedByUserId.value);
     }
     if (channelCid.present) {
-      map['channel_cid'] = Variable<String?>(channelCid.value);
+      map['channel_cid'] = Variable<String>(channelCid.value);
     }
     if (i18n.present) {
       final converter = $MessagesTable.$converter5;
@@ -1491,11 +1490,10 @@ class $MessagesTable extends Messages
       typeName: 'TEXT', requiredDuringInsert: false);
   final VerificationMeta _channelCidMeta = const VerificationMeta('channelCid');
   late final GeneratedColumn<String?> channelCid = GeneratedColumn<String?>(
-      'channel_cid', aliasedName, true,
+      'channel_cid', aliasedName, false,
       typeName: 'TEXT',
-      requiredDuringInsert: false,
+      requiredDuringInsert: true,
-      $customConstraints:
+      $customConstraints: 'REFERENCES channels(cid) ON DELETE CASCADE');
-          'NULLABLE REFERENCES channels(cid) ON DELETE CASCADE');
   final VerificationMeta _i18nMeta = const VerificationMeta('i18n');
   late final GeneratedColumnWithTypeConverter<Map<String, String>, String?>
       i18n = GeneratedColumn<String?>('i18n', aliasedName, true,
@@ -1634,6 +1632,8 @@ class $MessagesTable extends Messages
           _channelCidMeta,
           channelCid.isAcceptableOrUnknown(
               data['channel_cid']!, _channelCidMeta));
+    } else if (isInserting) {
+      context.missing(_channelCidMeta);
     }
     context.handle(_i18nMeta, const VerificationResult.success());
     context.handle(_extraDataMeta, const VerificationResult.success());
@@ -1739,7 +1739,7 @@ class PinnedMessageEntity extends DataClass
   final String? pinnedByUserId;
 
   /// The channel cid of which this message is part of
-  final String? channelCid;
+  final String channelCid;
 
   /// A Map of [messageText] translations.
   final Map<String, String>? i18n;
@@ -1769,7 +1769,7 @@ class PinnedMessageEntity extends DataClass
       this.pinnedAt,
       this.pinExpires,
       this.pinnedByUserId,
-      this.channelCid,
+      required this.channelCid,
       this.i18n,
       this.extraData});
   factory PinnedMessageEntity.fromData(
@@ -1825,7 +1825,7 @@ class PinnedMessageEntity extends DataClass
       pinnedByUserId: const StringType()
           .mapFromDatabaseResponse(data['${effectivePrefix}pinned_by_user_id']),
       channelCid: const StringType()
-          .mapFromDatabaseResponse(data['${effectivePrefix}channel_cid']),
+          .mapFromDatabaseResponse(data['${effectivePrefix}channel_cid'])!,
       i18n: $PinnedMessagesTable.$converter5.mapToDart(const StringType()
           .mapFromDatabaseResponse(data['${effectivePrefix}i18n'])),
       extraData: $PinnedMessagesTable.$converter6.mapToDart(const StringType()
@@ -1897,9 +1897,7 @@ class PinnedMessageEntity extends DataClass
     if (!nullToAbsent || pinnedByUserId != null) {
       map['pinned_by_user_id'] = Variable<String?>(pinnedByUserId);
     }
-    if (!nullToAbsent || channelCid != null) {
+    map['channel_cid'] = Variable<String>(channelCid);
-      map['channel_cid'] = Variable<String?>(channelCid);
-    }
     if (!nullToAbsent || i18n != null) {
       final converter = $PinnedMessagesTable.$converter5;
       map['i18n'] = Variable<String?>(converter.mapToSql(i18n));
@@ -1939,7 +1937,7 @@ class PinnedMessageEntity extends DataClass
       pinnedAt: serializer.fromJson<DateTime?>(json['pinnedAt']),
       pinExpires: serializer.fromJson<DateTime?>(json['pinExpires']),
       pinnedByUserId: serializer.fromJson<String?>(json['pinnedByUserId']),
-      channelCid: serializer.fromJson<String?>(json['channelCid']),
+      channelCid: serializer.fromJson<String>(json['channelCid']),
       i18n: serializer.fromJson<Map<String, String>?>(json['i18n']),
       extraData: serializer.fromJson<Map<String, Object?>?>(json['extraData']),
     );
@@ -1970,7 +1968,7 @@ class PinnedMessageEntity extends DataClass
       'pinnedAt': serializer.toJson<DateTime?>(pinnedAt),
       'pinExpires': serializer.toJson<DateTime?>(pinExpires),
       'pinnedByUserId': serializer.toJson<String?>(pinnedByUserId),
-      'channelCid': serializer.toJson<String?>(channelCid),
+      'channelCid': serializer.toJson<String>(channelCid),
       'i18n': serializer.toJson<Map<String, String>?>(i18n),
       'extraData': serializer.toJson<Map<String, Object?>?>(extraData),
     };
@@ -1999,7 +1997,7 @@ class PinnedMessageEntity extends DataClass
           Value<DateTime?> pinnedAt = const Value.absent(),
           Value<DateTime?> pinExpires = const Value.absent(),
           Value<String?> pinnedByUserId = const Value.absent(),
-          Value<String?> channelCid = const Value.absent(),
+          String? channelCid,
           Value<Map<String, String>?> i18n = const Value.absent(),
           Value<Map<String, Object?>?> extraData = const Value.absent()}) =>
       PinnedMessageEntity(
@@ -2031,7 +2029,7 @@ class PinnedMessageEntity extends DataClass
         pinExpires: pinExpires.present ? pinExpires.value : this.pinExpires,
         pinnedByUserId:
             pinnedByUserId.present ? pinnedByUserId.value : this.pinnedByUserId,
-        channelCid: channelCid.present ? channelCid.value : this.channelCid,
+        channelCid: channelCid ?? this.channelCid,
         i18n: i18n.present ? i18n.value : this.i18n,
         extraData: extraData.present ? extraData.value : this.extraData,
       );
@@ -2165,7 +2163,7 @@ class PinnedMessagesCompanion extends UpdateCompanion<PinnedMessageEntity> {
   final Value<DateTime?> pinnedAt;
   final Value<DateTime?> pinExpires;
   final Value<String?> pinnedByUserId;
-  final Value<String?> channelCid;
+  final Value<String> channelCid;
   final Value<Map<String, String>?> i18n;
   final Value<Map<String, Object?>?> extraData;
   const PinnedMessagesCompanion({
@@ -2218,12 +2216,13 @@ class PinnedMessagesCompanion extends UpdateCompanion<PinnedMessageEntity> {
     this.pinnedAt = const Value.absent(),
     this.pinExpires = const Value.absent(),
     this.pinnedByUserId = const Value.absent(),
-    this.channelCid = const Value.absent(),
+    required String channelCid,
     this.i18n = const Value.absent(),
     this.extraData = const Value.absent(),
   })  : id = Value(id),
         attachments = Value(attachments),
-        mentionedUsers = Value(mentionedUsers);
+        mentionedUsers = Value(mentionedUsers),
+        channelCid = Value(channelCid);
   static Insertable<PinnedMessageEntity> custom({
     Expression<String>? id,
     Expression<String?>? messageText,
@@ -2247,7 +2246,7 @@ class PinnedMessagesCompanion extends UpdateCompanion<PinnedMessageEntity> {
     Expression<DateTime?>? pinnedAt,
     Expression<DateTime?>? pinExpires,
     Expression<String?>? pinnedByUserId,
-    Expression<String?>? channelCid,
+    Expression<String>? channelCid,
     Expression<Map<String, String>?>? i18n,
     Expression<Map<String, Object?>?>? extraData,
   }) {
@@ -2303,7 +2302,7 @@ class PinnedMessagesCompanion extends UpdateCompanion<PinnedMessageEntity> {
       Value<DateTime?>? pinnedAt,
       Value<DateTime?>? pinExpires,
       Value<String?>? pinnedByUserId,
-      Value<String?>? channelCid,
+      Value<String>? channelCid,
       Value<Map<String, String>?>? i18n,
       Value<Map<String, Object?>?>? extraData}) {
     return PinnedMessagesCompanion(
@@ -2414,7 +2413,7 @@ class PinnedMessagesCompanion extends UpdateCompanion<PinnedMessageEntity> {
       map['pinned_by_user_id'] = Variable<String?>(pinnedByUserId.value);
     }
     if (channelCid.present) {
-      map['channel_cid'] = Variable<String?>(channelCid.value);
+      map['channel_cid'] = Variable<String>(channelCid.value);
     }
     if (i18n.present) {
       final converter = $PinnedMessagesTable.$converter5;
@@ -2589,11 +2588,10 @@ class $PinnedMessagesTable extends PinnedMessages
       typeName: 'TEXT', requiredDuringInsert: false);
   final VerificationMeta _channelCidMeta = const VerificationMeta('channelCid');
   late final GeneratedColumn<String?> channelCid = GeneratedColumn<String?>(
-      'channel_cid', aliasedName, true,
+      'channel_cid', aliasedName, false,
       typeName: 'TEXT',
-      requiredDuringInsert: false,
+      requiredDuringInsert: true,
-      $customConstraints:
+      $customConstraints: 'REFERENCES channels(cid) ON DELETE CASCADE');
-          'NULLABLE REFERENCES channels(cid) ON DELETE CASCADE');
   final VerificationMeta _i18nMeta = const VerificationMeta('i18n');
   late final GeneratedColumnWithTypeConverter<Map<String, String>, String?>
       i18n = GeneratedColumn<String?>('i18n', aliasedName, true,
@@ -2734,6 +2732,8 @@ class $PinnedMessagesTable extends PinnedMessages
           _channelCidMeta,
           channelCid.isAcceptableOrUnknown(
               data['channel_cid']!, _channelCidMeta));
+    } else if (isInserting) {
+      context.missing(_channelCidMeta);
     }
     context.handle(_i18nMeta, const VerificationResult.success());
     context.handle(_extraDataMeta, const VerificationResult.success());

packages/stream_chat_persistence/lib/src/entity/members.dart

@@ -39,8 +39,5 @@ class Members extends Table {
   DateTimeColumn get updatedAt => dateTime().withDefault(currentDateAndTime)();
 
   @override
-  Set<Column> get primaryKey => {
+  Set<Column> get primaryKey => {userId, channelCid};
-        userId,
-        channelCid,
-      };
 }

packages/stream_chat_persistence/lib/src/entity/messages.dart

@@ -77,8 +77,8 @@ class Messages extends Table {
   TextColumn get pinnedByUserId => text().nullable()();
 
   /// The channel cid of which this message is part of
-  TextColumn get channelCid => text().nullable().customConstraint(
+  TextColumn get channelCid =>
-      'NULLABLE REFERENCES channels(cid) ON DELETE CASCADE')();
+      text().customConstraint('REFERENCES channels(cid) ON DELETE CASCADE')();
 
   /// A Map of [messageText] translations.
   TextColumn get i18n => text().nullable().map(MapConverter<String>())();

packages/stream_chat_persistence/lib/src/mapper/message_mapper.dart

@@ -51,7 +51,7 @@ extension MessageEntityX on MessageEntity {
 /// Useful mapping functions for [Message]
 extension MessageX on Message {
   /// Maps a [Message] into [MessageEntity]
-  MessageEntity toEntity({String? cid}) => MessageEntity(
+  MessageEntity toEntity({required String cid}) => MessageEntity(
         id: id,
         attachments: attachments.map((it) => jsonEncode(it.toData())).toList(),
         channelCid: cid,

packages/stream_chat_persistence/lib/src/mapper/pinned_message_mapper.dart

@@ -51,7 +51,8 @@ extension PinnedMessageEntityX on PinnedMessageEntity {
 /// Useful mapping functions for [Message]
 extension PMessageX on Message {
   /// Maps a [Message] into [PinnedMessageEntity]
-  PinnedMessageEntity toPinnedEntity({String? cid}) => PinnedMessageEntity(
+  PinnedMessageEntity toPinnedEntity({required String cid}) =>
+      PinnedMessageEntity(
         id: id,
         attachments: attachments.map((it) => jsonEncode(it.toData())).toList(),
         channelCid: cid,

packages/stream_chat_persistence/lib/src/stream_chat_persistence_client.dart

@@ -296,25 +296,25 @@ class StreamChatPersistenceClient extends ChatPersistenceClient {
   }
 
   @override
-  Future<void> updateMembers(String cid, List<Member> members) {
+  Future<void> bulkUpdateMembers(Map<String, List<Member>> members) {
     assert(_debugIsConnected, '');
-    _logger.info('updateMembers');
+    _logger.info('bulkUpdateMembers');
-    return _readProtected(() => db!.memberDao.updateMembers(cid, members));
+    return _readProtected(() => db!.memberDao.bulkUpdateMembers(members));
   }
 
   @override
-  Future<void> updateMessages(String cid, List<Message> messages) {
+  Future<void> bulkUpdateMessages(Map<String, List<Message>> messages) {
     assert(_debugIsConnected, '');
-    _logger.info('updateMessages');
+    _logger.info('bulkUpdateMessages');
-    return _readProtected(() => db!.messageDao.updateMessages(cid, messages));
+    return _readProtected(() => db!.messageDao.bulkUpdateMessages(messages));
   }
 
   @override
-  Future<void> updatePinnedMessages(String cid, List<Message> messages) {
+  Future<void> bulkUpdatePinnedMessages(Map<String, List<Message>> messages) {
     assert(_debugIsConnected, '');
-    _logger.info('updatePinnedMessages');
+    _logger.info('bulkUpdatePinnedMessages');
     return _readProtected(
-      () => db!.pinnedMessageDao.updateMessages(cid, messages),
+      () => db!.pinnedMessageDao.bulkUpdateMessages(messages),
     );
   }
 
@@ -335,10 +335,10 @@ class StreamChatPersistenceClient extends ChatPersistenceClient {
   }
 
   @override
-  Future<void> updateReads(String cid, List<Read> reads) {
+  Future<void> bulkUpdateReads(Map<String, List<Read>> reads) {
     assert(_debugIsConnected, '');
-    _logger.info('updateReads');
+    _logger.info('bulkUpdateReads');
-    return _readProtected(() => db!.readDao.updateReads(cid, reads));
+    return _readProtected(() => db!.readDao.bulkUpdateReads(reads));
   }
 
   @override

packages/stream_chat_persistence/test/stream_chat_persistence_client_test.dart

@@ -397,23 +397,26 @@ void main() {
     test('updateMessages', () async {
       const cid = 'testCid';
       final messages = List.generate(3, (index) => Message());
-      when(() => mockDatabase.messageDao.updateMessages(cid, messages))
+
+      when(() => mockDatabase.messageDao.bulkUpdateMessages({cid: messages}))
           .thenAnswer((_) => Future.value());
 
       await client.updateMessages(cid, messages);
-      verify(() => mockDatabase.messageDao.updateMessages(cid, messages))
+      verify(() => mockDatabase.messageDao.bulkUpdateMessages({cid: messages}))
           .called(1);
     });
 
     test('updatePinnedMessages', () async {
       const cid = 'testCid';
       final messages = List.generate(3, (index) => Message());
-      when(() => mockDatabase.pinnedMessageDao.updateMessages(cid, messages))
+      when(
-          .thenAnswer((_) => Future.value());
+        () => mockDatabase.pinnedMessageDao.bulkUpdateMessages({cid: messages}),
+      ).thenAnswer((_) => Future.value());
 
       await client.updatePinnedMessages(cid, messages);
-      verify(() => mockDatabase.pinnedMessageDao.updateMessages(cid, messages))
+      verify(
-          .called(1);
+        () => mockDatabase.pinnedMessageDao.bulkUpdateMessages({cid: messages}),
+      ).called(1);
     });
 
     test('getChannelThreads', () async {
@@ -456,11 +459,11 @@ void main() {
     test('updateMembers', () async {
       const cid = 'testCid';
       final members = List.generate(3, (index) => Member());
-      when(() => mockDatabase.memberDao.updateMembers(cid, members))
+      when(() => mockDatabase.memberDao.bulkUpdateMembers({cid: members}))
           .thenAnswer((_) => Future.value());
 
       await client.updateMembers(cid, members);
-      verify(() => mockDatabase.memberDao.updateMembers(cid, members))
+      verify(() => mockDatabase.memberDao.bulkUpdateMembers({cid: members}))
           .called(1);
     });
 
@@ -473,11 +476,12 @@ void main() {
           lastRead: DateTime.now(),
         ),
       );
-      when(() => mockDatabase.readDao.updateReads(cid, reads))
+      when(() => mockDatabase.readDao.bulkUpdateReads({cid: reads}))
           .thenAnswer((_) => Future.value());
 
       await client.updateReads(cid, reads);
-      verify(() => mockDatabase.readDao.updateReads(cid, reads)).called(1);
+      verify(() => mockDatabase.readDao.bulkUpdateReads({cid: reads}))
+          .called(1);
     });
 
     test('updateUsers', () async {