[StreamChatPersistence] Add initial implementation

Signed-off-by: Sahil Kumar <[email protected]>

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

@@ -0,0 +1,57 @@
+import 'package:moor/moor.dart';
+import 'package:stream_chat/stream_chat.dart';
+import 'package:stream_chat_persistence/src/db/moor_chat_database.dart';
+import 'package:stream_chat_persistence/src/entity/channels.dart';
+import 'package:stream_chat_persistence/src/entity/users.dart';
+import '../mapper/mapper.dart';
+
+part 'channel_dao.g.dart';
+
+///
+@UseDao(tables: [Channels, Users])
+class ChannelDao extends DatabaseAccessor<MoorChatDatabase>
+    with _$ChannelDaoMixin {
+  ///
+  ChannelDao(MoorChatDatabase db) : super(db);
+
+  /// Get channel by cid
+  Future<ChannelModel> getChannelByCid(String cid) async {
+    return (select(channels)..where((c) => c.cid.equals(cid))).join([
+      leftOuterJoin(users, channels.createdById.equalsExp(users.id)),
+    ]).map((rows) {
+      final channel = rows.readTable(channels);
+      final createdBy = rows.readTable(users);
+      return channel.toChannelModel(createdBy: createdBy?.toUser());
+    }).getSingle();
+  }
+
+  /// Delete all channels by matching cid in [cids]
+  ///
+  /// This will automatically delete the following linked records
+  /// 1. Channel Reads
+  /// 2. Channel Members
+  /// 3. Channel Messages -> Messages Reactions
+  Future<void> deleteChannelByCids(List<String> cids) async {
+    return (delete(channels)..where((tbl) => tbl.cid.isIn(cids))).go();
+  }
+
+  /// Get the channel cids saved in the storage
+  Future<List<String>> get cids {
+    return (select(channels)
+          ..orderBy([(c) => OrderingTerm.desc(c.lastMessageAt)])
+          ..limit(250))
+        .map((c) => c.cid)
+        .get();
+  }
+
+  ///
+  Future<void> updateChannels(List<ChannelModel> channelList) {
+    return batch(
+      (it) => it.insertAll(
+        channels,
+        channelList.map((c) => c.toEntity()).toList(),
+        mode: InsertMode.insertOrReplace,
+      ),
+    );
+  }
+}

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

@@ -0,0 +1,126 @@
+import 'dart:convert';
+
+import 'package:moor/moor.dart';
+import 'package:stream_chat/stream_chat.dart';
+import 'package:stream_chat_persistence/src/db/moor_chat_database.dart';
+import 'package:stream_chat_persistence/src/entity/channel_queries.dart';
+import 'package:stream_chat_persistence/src/entity/channels.dart';
+import 'package:stream_chat_persistence/src/entity/users.dart';
+import '../mapper/mapper.dart';
+
+part 'channel_query_dao.g.dart';
+
+///
+@UseDao(tables: [ChannelQueries, Channels, Users])
+class ChannelQueryDao extends DatabaseAccessor<MoorChatDatabase>
+    with _$ChannelQueryDaoMixin {
+  ///
+  ChannelQueryDao(this._db) : super(_db);
+
+  final MoorChatDatabase _db;
+
+  String _computeHash(Map<String, dynamic> filter) {
+    if (filter == null) {
+      return 'allchannels';
+    }
+    final hash = base64Encode(utf8.encode('filter: ${jsonEncode(filter)}'));
+    return hash;
+  }
+
+  /// Update list of channel queries
+  /// If [clearQueryCache] is true before the insert
+  /// the list of matching rows will be deleted
+  Future<void> updateChannelQueries(
+    Map<String, dynamic> filter,
+    List<String> cids,
+    bool clearQueryCache,
+  ) async {
+    final hash = _computeHash(filter);
+    if (clearQueryCache) {
+      await (delete(channelQueries)
+            ..where((query) => query.queryHash.equals(hash)))
+          .go();
+    }
+
+    return batch((batch) {
+      batch.insertAll(
+        channelQueries,
+        cids.map((cid) {
+          return ChannelQueryEntity(
+            queryHash: hash,
+            channelCid: cid,
+          );
+        }).toList(),
+        mode: InsertMode.insertOrReplace,
+      );
+    });
+  }
+
+  /// Get list of channels by filter, sort and paginationParams
+  Future<List<ChannelState>> getChannelStates({
+    Map<String, dynamic> filter,
+    List<SortOption> sort = const [],
+    PaginationParams paginationParams,
+  }) async {
+    final hash = _computeHash(filter);
+    final cachedChannels = await Future.wait(await (select(channelQueries)
+          ..where((c) => c.queryHash.equals(hash)))
+        .get()
+        .then((channelQueries) {
+      final cids = channelQueries.map((c) => c.channelCid).toList();
+      final query = select(channels)..where((c) => c.cid.isIn(cids));
+
+      sort = sort
+          ?.where((s) => ChannelModel.topLevelFields.contains(s.field))
+          ?.toList();
+
+      if (sort != null && sort.isNotEmpty) {
+        query.orderBy(sort.map((s) {
+          final orderExpression = CustomExpression('channels.${s.field}');
+          return (c) => OrderingTerm(
+                expression: orderExpression,
+                mode: s.direction == 1 ? OrderingMode.asc : OrderingMode.desc,
+              );
+        }).toList());
+      }
+
+      if (paginationParams != null) {
+        query.limit(
+          paginationParams.limit ?? 10,
+          offset: paginationParams.offset,
+        );
+      }
+
+      return query.join([
+        leftOuterJoin(users, channels.createdById.equalsExp(users.id)),
+      ]).map((row) async {
+        final userEntity = row.readTable(users);
+        final channelEntity = row.readTable(channels);
+
+        final cid = channelEntity.cid;
+        final members = await _db.memberDao.getMembersByCid(cid);
+        final reads = await _db.readDao.getReadsByCid(cid);
+        final messages = await _db.messageDao.getMessagesByCid(cid);
+
+        return channelEntity.toChannelState(
+          createdBy: userEntity?.toUser(),
+          members: members,
+          reads: reads,
+          messages: messages,
+        );
+      }).get();
+    }));
+
+    if (sort?.isEmpty != false && cachedChannels?.isNotEmpty == true) {
+      cachedChannels
+          .sort((a, b) => b.channel.updatedAt.compareTo(a.channel.updatedAt));
+      cachedChannels.sort((a, b) {
+        final dateA = a.channel.lastMessageAt ?? a.channel.createdAt;
+        final dateB = b.channel.lastMessageAt ?? b.channel.createdAt;
+        return dateB.compareTo(dateA);
+      });
+    }
+
+    return cachedChannels;
+  }
+}

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

@@ -0,0 +1,58 @@
+import 'package:moor/moor.dart';
+import 'package:stream_chat/stream_chat.dart';
+import 'package:stream_chat_persistence/src/db/moor_chat_database.dart';
+import 'package:stream_chat_persistence/src/entity/connection_events.dart';
+import 'package:stream_chat_persistence/src/entity/users.dart';
+import '../mapper/mapper.dart';
+
+part 'connection_event_dao.g.dart';
+
+///
+@UseDao(tables: [ConnectionEvents, Users])
+class ConnectionEventDao extends DatabaseAccessor<MoorChatDatabase>
+    with _$ConnectionEventDaoMixin {
+  ///
+  ConnectionEventDao(MoorChatDatabase db) : super(db);
+
+  /// Get the latest stored connection event
+  Future<Event> get connectionEvent {
+    return select(connectionEvents).join([
+      leftOuterJoin(users, connectionEvents.ownUserId.equalsExp(users.id)),
+    ]).map((rows) {
+      final event = rows.readTable(connectionEvents);
+      final user = rows.readTable(users);
+      return event.toEvent(user: user?.toUser());
+    }).getSingle();
+  }
+
+  /// Get the latest stored lastSyncAt
+  Future<DateTime> get lastSyncAt {
+    return select(connectionEvents).getSingle().then((r) => r?.lastSyncAt);
+  }
+
+  /// Update stored connection event with latest data
+  Future<int> updateConnectionEvent(Event event) async {
+    final connectionInfo = await select(connectionEvents).getSingle();
+    return into(connectionEvents).insert(
+      ConnectionEventEntity(
+        id: 1,
+        lastSyncAt: connectionInfo?.lastSyncAt,
+        lastEventAt: event.createdAt ?? connectionInfo?.lastEventAt,
+        totalUnreadCount:
+            event.totalUnreadCount ?? connectionInfo?.totalUnreadCount,
+        ownUserId: event.me?.id ?? connectionInfo?.ownUserId,
+        unreadChannels: event.unreadChannels ?? connectionInfo?.unreadChannels,
+      ),
+      mode: InsertMode.insertOrReplace,
+    );
+  }
+
+  /// Update stored lastSyncAt with latest data
+  Future<int> updateLastSyncAt(DateTime lastSyncAt) async {
+    return (update(connectionEvents)..where((tbl) => tbl.id.equals(1))).write(
+      ConnectionEventsCompanion(
+        lastSyncAt: Value(lastSyncAt),
+      ),
+    );
+  }
+}

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

@@ -0,0 +1,8 @@
+export 'user_dao.dart';
+export 'channel_dao.dart';
+export 'message_dao.dart';
+export 'member_dao.dart';
+export 'connection_event_dao.dart';
+export 'reaction_dao.dart';
+export 'read_dao.dart';
+export 'channel_query_dao.dart';

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

@@ -0,0 +1,43 @@
+import 'package:moor/moor.dart';
+import 'package:stream_chat/stream_chat.dart';
+import 'package:stream_chat_persistence/src/db/moor_chat_database.dart';
+
+import 'package:stream_chat_persistence/src/entity/members.dart';
+import 'package:stream_chat_persistence/src/entity/users.dart';
+
+import '../mapper/mapper.dart';
+
+part 'member_dao.g.dart';
+
+///
+@UseDao(tables: [Members, Users])
+class MemberDao extends DatabaseAccessor<MoorChatDatabase>
+    with _$MemberDaoMixin {
+  ///
+  MemberDao(MoorChatDatabase db) : super(db);
+
+  /// Get all members where [members.channelCid] matches [cid]
+  Future<List<Member>> getMembersByCid(String cid) async {
+    return (select(members).join([
+      leftOuterJoin(users, members.userId.equalsExp(users.id)),
+    ])
+          ..where(members.channelCid.equals(cid))
+          ..orderBy([OrderingTerm.asc(members.createdAt)]))
+        .map((row) {
+      final userEntity = row.readTable(users);
+      final memberEntity = row.readTable(members);
+      return memberEntity.toMember(user: userEntity?.toUser());
+    }).get();
+  }
+
+  ///
+  Future<void> updateMembers(String cid, List<Member> memberList) async {
+    return batch(
+      (it) => it.insertAll(
+        members,
+        memberList.map((m) => m.toEntity(cid: cid)).toList(),
+        mode: InsertMode.insertOrReplace,
+      ),
+    );
+  }
+}

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

@@ -0,0 +1,143 @@
+import 'package:moor/moor.dart';
+import 'package:stream_chat/stream_chat.dart';
+import 'package:stream_chat_persistence/src/db/moor_chat_database.dart';
+import 'package:stream_chat_persistence/src/entity/messages.dart';
+import 'package:stream_chat_persistence/src/entity/users.dart';
+import '../mapper/mapper.dart';
+
+part 'message_dao.g.dart';
+
+///
+@UseDao(tables: [Messages, Users])
+class MessageDao extends DatabaseAccessor<MoorChatDatabase>
+    with _$MessageDaoMixin {
+  ///
+  MessageDao(this._db) : super(_db);
+
+  final MoorChatDatabase _db;
+
+  /// Removes all the messages by matching [messages.id] in [messageIds]
+  ///
+  /// This will automatically delete the following linked records
+  /// 1. Message Reactions
+  Future<void> deleteMessageByIds(List<String> messageIds) {
+    return (delete(messages)..where((tbl) => tbl.id.isIn(messageIds))).go();
+  }
+
+  /// Removes all the messages by matching [messages.channelCid] in [cids]
+  ///
+  /// This will automatically delete the following linked records
+  /// 1. Message Reactions
+  Future<void> deleteMessageByCids(List<String> cids) async {
+    return (delete(messages)..where((tbl) => tbl.channelCid.isIn(cids))).go();
+  }
+
+  Future<Message> _messageFromJoinRow(TypedResult rows) async {
+    final userEntity = rows.readTable(users);
+    final msgEntity = rows.readTable(messages);
+    final latestReactions = await _db.reactionDao.getReactions(msgEntity.id);
+    final ownReactions = await _db.reactionDao.getOwnReactions(
+      msgEntity.id,
+      userEntity.id,
+    );
+    Message quotedMessage;
+    if (msgEntity.quotedMessageId != null) {
+      quotedMessage = await getMessageById(msgEntity.quotedMessageId);
+    }
+    return msgEntity.toMessage(
+      user: userEntity?.toUser(),
+      latestReactions: latestReactions,
+      ownReactions: ownReactions,
+      quotedMessage: quotedMessage,
+    );
+  }
+
+  ///
+  Future<Message> getMessageById(String id) async {
+    return await (select(messages).join([
+      leftOuterJoin(users, messages.userId.equalsExp(users.id)),
+    ])
+          ..where(messages.id.equals(id)))
+        .map(_messageFromJoinRow)
+        .getSingle();
+  }
+
+  ///
+  Future<List<Message>> getThreadMessages(String cid) async {
+    return Future.wait(await (select(messages).join([
+      leftOuterJoin(users, messages.userId.equalsExp(users.id)),
+    ])
+          ..where(messages.channelCid.equals(cid))
+          ..where(isNotNull(messages.parentId))
+          ..orderBy([OrderingTerm.asc(messages.createdAt)]))
+        .map(_messageFromJoinRow)
+        .get());
+  }
+
+  ///
+  Future<List<Message>> getThreadMessagesByParentId(
+    String parentId, {
+    String lessThan,
+  }) async {
+    final msgList = await Future.wait(await (select(messages).join([
+      innerJoin(users, messages.userId.equalsExp(users.id)),
+    ])
+          ..where(messages.parentId.equals(parentId))
+          ..orderBy([OrderingTerm.asc(messages.createdAt)]))
+        .map(_messageFromJoinRow)
+        .get());
+
+    if (lessThan != null) {
+      final lessThanIndex = msgList.indexWhere((m) => m.id == lessThan);
+      msgList.removeRange(lessThanIndex, msgList.length);
+    }
+    return msgList;
+  }
+
+  ///
+  Future<List<Message>> getMessagesByCid(
+    String cid, {
+    int limit = 20,
+    String messageLessThan,
+    String messageGreaterThan,
+  }) async {
+    final msgList = await Future.wait(await (select(messages).join([
+      leftOuterJoin(users, messages.userId.equalsExp(users.id)),
+    ])
+          ..where(messages.channelCid.equals(cid))
+          ..where(
+              isNull(messages.parentId) | messages.showInChannel.equals(true))
+          ..orderBy([OrderingTerm.asc(messages.createdAt)]))
+        .map(_messageFromJoinRow)
+        .get());
+
+    if (messageLessThan != null) {
+      final lessThanIndex = msgList.indexWhere((m) => m.id == messageLessThan);
+      if (lessThanIndex != -1) {
+        msgList.removeRange(lessThanIndex, msgList.length);
+      }
+    }
+    if (messageGreaterThan != null) {
+      final greaterThanIndex =
+          msgList.indexWhere((m) => m.id == messageGreaterThan);
+      if (greaterThanIndex != -1) {
+        msgList.removeRange(0, greaterThanIndex);
+      }
+    }
+    if (limit != null) {
+      return msgList.take(limit).toList();
+    }
+    return msgList;
+  }
+
+  ///
+  Future<int> updateMessages(String cid, List<Message> messageList) {
+    return batch((batch) {
+      batch.insertAll(
+        messages,
+        messageList.map((it) => it.toEntity(cid: cid)).toList(),
+        mode: InsertMode.insertOrReplace,
+      );
+    });
+  }
+}

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

@@ -0,0 +1,50 @@
+import 'package:moor/moor.dart';
+import 'package:stream_chat/stream_chat.dart';
+import 'package:stream_chat_persistence/src/db/moor_chat_database.dart';
+import 'package:stream_chat_persistence/src/entity/reactions.dart';
+import 'package:stream_chat_persistence/src/entity/users.dart';
+import '../mapper/mapper.dart';
+
+part 'reaction_dao.g.dart';
+
+///
+@UseDao(tables: [Reactions, Users])
+class ReactionDao extends DatabaseAccessor<MoorChatDatabase>
+    with _$ReactionDaoMixin {
+  ///
+  ReactionDao(MoorChatDatabase db) : super(db);
+
+  ///
+  Future<List<Reaction>> getReactions(String messageId) {
+    return (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.readTable(users);
+      final reactionEntity = rows.readTable(reactions);
+      return reactionEntity.toReaction(user: userEntity?.toUser());
+    }).get();
+  }
+
+  ///
+  Future<List<Reaction>> getOwnReactions(
+    String messageId,
+    String userId,
+  ) async {
+    final reactions = await getReactions(messageId);
+    return reactions.where((it) => it.userId == userId).toList();
+  }
+
+  ///
+  Future<void> updateReactions(List<Reaction> reactionList) {
+    return batch((it) {
+      it.insertAll(
+        reactions,
+        reactionList.map((r) => r.toEntity()).toList(),
+        mode: InsertMode.insertOrReplace,
+      );
+    });
+  }
+}

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

@@ -0,0 +1,42 @@
+import 'package:moor/moor.dart';
+import 'package:stream_chat/stream_chat.dart';
+import 'package:stream_chat_persistence/src/db/moor_chat_database.dart';
+import 'package:stream_chat_persistence/src/entity/reads.dart';
+import 'package:stream_chat_persistence/src/entity/users.dart';
+import '../mapper/mapper.dart';
+
+part 'read_dao.g.dart';
+
+///
+@UseDao(tables: [Reads, Users])
+class ReadDao extends DatabaseAccessor<MoorChatDatabase> with _$ReadDaoMixin {
+  ///
+  ReadDao(MoorChatDatabase db) : super(db);
+
+  /// Get all reads where [reads.channelCid] matches [cid]
+  Future<List<Read>> getReadsByCid(String cid) async {
+    return (select(reads).join([
+      leftOuterJoin(users, reads.userId.equalsExp(users.id)),
+    ])
+          ..where(reads.channelCid.equals(cid))
+          ..orderBy([
+            OrderingTerm.asc(reads.lastRead),
+          ]))
+        .map((row) {
+      final userEntity = row.readTable(users);
+      final readEntity = row.readTable(reads);
+      return readEntity.toRead(user: userEntity?.toUser());
+    }).get();
+  }
+
+  ///
+  Future<void> updateReads(String cid, List<Read> readList) {
+    return batch(
+      (it) => it.insertAll(
+        reads,
+        readList.map((r) => r.toEntity(cid: cid)).toList(),
+        mode: InsertMode.insertOrReplace,
+      ),
+    );
+  }
+}

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

@@ -0,0 +1,25 @@
+import 'package:moor/moor.dart';
+import 'package:stream_chat/stream_chat.dart';
+import 'package:stream_chat_persistence/src/db/moor_chat_database.dart';
+import 'package:stream_chat_persistence/src/entity/users.dart';
+import '../mapper/user_mapper.dart';
+
+part 'user_dao.g.dart';
+
+///
+@UseDao(tables: [Users])
+class UserDao extends DatabaseAccessor<MoorChatDatabase> with _$UserDaoMixin {
+  ///
+  UserDao(MoorChatDatabase db) : super(db);
+
+  ///
+  Future<void> updateUsers(List<User> userList) {
+    return batch(
+      (it) => it.insertAll(
+        users,
+        userList.map((u) => u.toEntity()).toList(),
+        mode: InsertMode.insertOrReplace,
+      ),
+    );
+  }
+}