add dart client to packages
This commit is contained in:
Neevash Ramdial
@@ -0,0 +1,258 @@
|
||||
part of 'offline_storage.dart';
|
||||
|
||||
@DataClassName('ChannelQuery')
|
||||
class _ChannelQueries extends Table {
|
||||
TextColumn get queryHash => text()();
|
||||
|
||||
TextColumn get channelCid => text()();
|
||||
|
||||
@override
|
||||
Set<Column> get primaryKey => {
|
||||
queryHash,
|
||||
channelCid,
|
||||
};
|
||||
}
|
||||
|
||||
class _Channels extends Table {
|
||||
TextColumn get id => text()();
|
||||
|
||||
TextColumn get type => text()();
|
||||
|
||||
TextColumn get cid => text()();
|
||||
|
||||
TextColumn get config => text()();
|
||||
|
||||
BoolColumn get frozen => boolean().withDefault(Constant(false))();
|
||||
|
||||
DateTimeColumn get lastMessageAt => dateTime().nullable()();
|
||||
|
||||
DateTimeColumn get createdAt => dateTime().nullable()();
|
||||
|
||||
DateTimeColumn get updatedAt => dateTime().nullable()();
|
||||
|
||||
DateTimeColumn get deletedAt => dateTime().nullable()();
|
||||
|
||||
IntColumn get memberCount => integer().nullable()();
|
||||
|
||||
TextColumn get extraData => text().nullable().map(_ExtraDataConverter())();
|
||||
|
||||
TextColumn get createdBy => text().nullable()();
|
||||
|
||||
@override
|
||||
Set<Column> get primaryKey => {cid};
|
||||
}
|
||||
|
||||
class _ConnectionEvent extends Table {
|
||||
IntColumn get id => integer()();
|
||||
|
||||
TextColumn get ownUser => text().nullable().map(_ExtraDataConverter())();
|
||||
|
||||
IntColumn get totalUnreadCount => integer().nullable()();
|
||||
|
||||
IntColumn get unreadChannels => integer().nullable()();
|
||||
|
||||
DateTimeColumn get lastEventAt => dateTime().nullable()();
|
||||
|
||||
DateTimeColumn get lastSyncAt => dateTime().nullable()();
|
||||
|
||||
@override
|
||||
Set<Column> get primaryKey => {id};
|
||||
}
|
||||
|
||||
class _Users extends Table {
|
||||
TextColumn get id => text()();
|
||||
|
||||
TextColumn get role => text().nullable()();
|
||||
|
||||
DateTimeColumn get createdAt => dateTime().nullable()();
|
||||
|
||||
DateTimeColumn get updatedAt => dateTime().nullable()();
|
||||
|
||||
DateTimeColumn get lastActive => dateTime().nullable()();
|
||||
|
||||
BoolColumn get online => boolean().nullable()();
|
||||
|
||||
BoolColumn get banned => boolean().nullable()();
|
||||
|
||||
TextColumn get extraData => text().nullable().map(_ExtraDataConverter())();
|
||||
|
||||
@override
|
||||
Set<Column> get primaryKey => {id};
|
||||
}
|
||||
|
||||
class _Reads extends Table {
|
||||
DateTimeColumn get lastRead => dateTime()();
|
||||
|
||||
TextColumn get userId => text()();
|
||||
|
||||
TextColumn get channelCid => text()();
|
||||
|
||||
IntColumn get unreadMessages => integer().nullable()();
|
||||
|
||||
@override
|
||||
Set<Column> get primaryKey => {
|
||||
userId,
|
||||
channelCid,
|
||||
};
|
||||
}
|
||||
|
||||
class _Reactions extends Table {
|
||||
TextColumn get messageId => text()();
|
||||
|
||||
TextColumn get type => text()();
|
||||
|
||||
DateTimeColumn get createdAt => dateTime()();
|
||||
|
||||
IntColumn get score => integer().nullable()();
|
||||
|
||||
TextColumn get userId => text()();
|
||||
|
||||
TextColumn get extraData => text().nullable().map(_ExtraDataConverter())();
|
||||
|
||||
@override
|
||||
Set<Column> get primaryKey => {
|
||||
messageId,
|
||||
type,
|
||||
userId,
|
||||
};
|
||||
}
|
||||
|
||||
class _Messages extends Table {
|
||||
TextColumn get id => text()();
|
||||
|
||||
TextColumn get messageText => text().nullable()();
|
||||
|
||||
TextColumn get attachmentJson => text().nullable()();
|
||||
|
||||
IntColumn get status =>
|
||||
integer().map(_MessageSendingStatusConverter()).nullable()();
|
||||
|
||||
TextColumn get type => text().nullable()();
|
||||
|
||||
List<User> mentionedUsers;
|
||||
|
||||
TextColumn get reactionCounts =>
|
||||
text().nullable().map(_ExtraDataConverter<int>())();
|
||||
|
||||
TextColumn get reactionScores =>
|
||||
text().nullable().map(_ExtraDataConverter<int>())();
|
||||
|
||||
TextColumn get parentId => text().nullable()();
|
||||
|
||||
TextColumn get quotedMessageId => text().nullable()();
|
||||
|
||||
IntColumn get replyCount => integer().nullable()();
|
||||
|
||||
BoolColumn get showInChannel => boolean().nullable()();
|
||||
|
||||
BoolColumn get shadowed => boolean().nullable()();
|
||||
|
||||
TextColumn get command => text().nullable()();
|
||||
|
||||
DateTimeColumn get createdAt => dateTime()();
|
||||
|
||||
DateTimeColumn get updatedAt => dateTime().nullable()();
|
||||
|
||||
DateTimeColumn get deletedAt => dateTime().nullable()();
|
||||
|
||||
TextColumn get userId => text().nullable()();
|
||||
|
||||
TextColumn get channelCid => text().nullable()();
|
||||
|
||||
TextColumn get extraData => text().nullable().map(_ExtraDataConverter())();
|
||||
|
||||
@override
|
||||
Set<Column> get primaryKey => {id};
|
||||
}
|
||||
|
||||
class _Members extends Table {
|
||||
TextColumn get userId => text()();
|
||||
|
||||
TextColumn get channelCid => text()();
|
||||
|
||||
TextColumn get role => text().nullable()();
|
||||
|
||||
DateTimeColumn get inviteAcceptedAt => dateTime().nullable()();
|
||||
|
||||
DateTimeColumn get inviteRejectedAt => dateTime().nullable()();
|
||||
|
||||
BoolColumn get invited => boolean().nullable()();
|
||||
|
||||
BoolColumn get banned => boolean().nullable()();
|
||||
|
||||
BoolColumn get shadowBanned => boolean().nullable()();
|
||||
|
||||
BoolColumn get isModerator => boolean().nullable()();
|
||||
|
||||
DateTimeColumn get createdAt => dateTime()();
|
||||
|
||||
DateTimeColumn get updatedAt => dateTime().nullable()();
|
||||
|
||||
@override
|
||||
Set<Column> get primaryKey => {
|
||||
userId,
|
||||
channelCid,
|
||||
};
|
||||
}
|
||||
|
||||
class _ExtraDataConverter<T> extends TypeConverter<Map<String, T>, String> {
|
||||
@override
|
||||
Map<String, T> mapToDart(fromDb) {
|
||||
if (fromDb == null) {
|
||||
return null;
|
||||
}
|
||||
return Map<String, T>.from(jsonDecode(fromDb) ?? {});
|
||||
}
|
||||
|
||||
@override
|
||||
String mapToSql(value) {
|
||||
return jsonEncode(value);
|
||||
}
|
||||
}
|
||||
|
||||
class _MessageSendingStatusConverter
|
||||
extends TypeConverter<MessageSendingStatus, int> {
|
||||
@override
|
||||
MessageSendingStatus mapToDart(int fromDb) {
|
||||
switch (fromDb) {
|
||||
case 0:
|
||||
return MessageSendingStatus.SENDING;
|
||||
case 1:
|
||||
return MessageSendingStatus.SENT;
|
||||
case 2:
|
||||
return MessageSendingStatus.FAILED;
|
||||
case 3:
|
||||
return MessageSendingStatus.UPDATING;
|
||||
case 4:
|
||||
return MessageSendingStatus.FAILED_UPDATE;
|
||||
case 5:
|
||||
return MessageSendingStatus.DELETING;
|
||||
case 6:
|
||||
return MessageSendingStatus.FAILED_DELETE;
|
||||
default:
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
@override
|
||||
int mapToSql(MessageSendingStatus value) {
|
||||
switch (value) {
|
||||
case MessageSendingStatus.SENDING:
|
||||
return 0;
|
||||
case MessageSendingStatus.SENT:
|
||||
return 1;
|
||||
case MessageSendingStatus.FAILED:
|
||||
return 2;
|
||||
case MessageSendingStatus.UPDATING:
|
||||
return 3;
|
||||
case MessageSendingStatus.FAILED_UPDATE:
|
||||
return 4;
|
||||
case MessageSendingStatus.DELETING:
|
||||
return 5;
|
||||
case MessageSendingStatus.FAILED_DELETE:
|
||||
return 6;
|
||||
default:
|
||||
return null;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,841 @@
|
||||
import 'dart:convert';
|
||||
|
||||
import 'package:flutter/material.dart' show WidgetsFlutterBinding;
|
||||
import 'package:logging/logging.dart';
|
||||
import 'package:moor/isolate.dart';
|
||||
import 'package:moor/moor.dart';
|
||||
import 'package:stream_chat/src/db/shared/shared_db.dart';
|
||||
import 'package:stream_chat/src/models/event.dart';
|
||||
import 'package:stream_chat/src/models/own_user.dart';
|
||||
|
||||
import '../api/requests.dart';
|
||||
import '../models/attachment.dart';
|
||||
import '../models/channel_config.dart';
|
||||
import '../models/channel_model.dart';
|
||||
import '../models/channel_state.dart';
|
||||
import '../models/member.dart';
|
||||
import '../models/message.dart';
|
||||
import '../models/reaction.dart';
|
||||
import '../models/read.dart';
|
||||
import '../models/user.dart';
|
||||
|
||||
part 'models.part.dart';
|
||||
|
||||
part 'offline_storage.g.dart';
|
||||
|
||||
/// Gets a new instance of the database running on a background isolate
|
||||
Future<OfflineStorage> connectDatabase(User user, Logger logger) async {
|
||||
logger.info('Connecting on background isolate');
|
||||
WidgetsFlutterBinding.ensureInitialized();
|
||||
return SharedDB.constructOfflineStorage(
|
||||
userId: user.id,
|
||||
logger: logger,
|
||||
);
|
||||
}
|
||||
|
||||
LazyDatabase _openConnection(String userId) {
|
||||
moorRuntimeOptions.dontWarnAboutMultipleDatabases = true;
|
||||
return LazyDatabase(() async {
|
||||
return await SharedDB.constructDatabase('db_$userId.sqlite');
|
||||
});
|
||||
}
|
||||
|
||||
/// Offline database used for caching channel queries and state
|
||||
@UseMoor(tables: [
|
||||
_ConnectionEvent,
|
||||
_Channels,
|
||||
_Users,
|
||||
_Messages,
|
||||
_Reads,
|
||||
_Members,
|
||||
_ChannelQueries,
|
||||
_Reactions,
|
||||
])
|
||||
class OfflineStorage extends _$OfflineStorage {
|
||||
/// Creates a new database instance
|
||||
OfflineStorage.connect(
|
||||
DatabaseConnection connection,
|
||||
this._userId,
|
||||
this._isolate,
|
||||
this._logger,
|
||||
) : super.connect(connection);
|
||||
|
||||
/// Instantiate a new OfflineStorage
|
||||
OfflineStorage(
|
||||
this._userId,
|
||||
this._logger,
|
||||
) : _isolate = null,
|
||||
super(_openConnection(_userId)) {
|
||||
_logger.info('Connecting on standard isolate');
|
||||
}
|
||||
|
||||
final String _userId;
|
||||
final MoorIsolate _isolate;
|
||||
final Logger _logger;
|
||||
|
||||
// you should bump this number whenever you change or add a table definition. Migrations
|
||||
// are covered later in this readme.
|
||||
@override
|
||||
int get schemaVersion => 8;
|
||||
|
||||
@override
|
||||
MigrationStrategy get migration => MigrationStrategy(
|
||||
onUpgrade: (openingDetails, before, after) async {
|
||||
if (before != after) {
|
||||
final m = createMigrator();
|
||||
for (final table in allTables) {
|
||||
await m.deleteTable(table.actualTableName);
|
||||
await m.createTable(table);
|
||||
}
|
||||
}
|
||||
},
|
||||
);
|
||||
|
||||
/// Closes the database instance
|
||||
/// If [flush] is true, the database data will be deleted
|
||||
Future<void> disconnect({bool flush = false}) async {
|
||||
_logger.info('Disconnecting');
|
||||
if (flush) {
|
||||
_logger.info('Flushing');
|
||||
await batch((batch) {
|
||||
allTables.forEach((table) {
|
||||
delete(table).go();
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
await _isolate?.shutdownAll();
|
||||
|
||||
await close();
|
||||
}
|
||||
|
||||
/// Get stored replies by messageId
|
||||
Future<List<Message>> getReplies(
|
||||
String parentId, {
|
||||
String lessThan,
|
||||
}) async {
|
||||
final offlineList = 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 = offlineList.indexWhere((m) => m.id == lessThan);
|
||||
offlineList.removeRange(lessThanIndex, offlineList.length);
|
||||
}
|
||||
|
||||
return offlineList;
|
||||
}
|
||||
|
||||
/// Get stored connection event
|
||||
Future<Event> getConnectionInfo() async {
|
||||
return select(connectionEvent).map((row) {
|
||||
return Event(
|
||||
me: row.ownUser != null ? OwnUser.fromJson(row.ownUser) : null,
|
||||
totalUnreadCount: row.totalUnreadCount,
|
||||
unreadChannels: row.unreadChannels,
|
||||
);
|
||||
}).getSingle();
|
||||
}
|
||||
|
||||
/// Get stored lastSyncAt
|
||||
Future<DateTime> getLastSyncAt() async {
|
||||
return select(connectionEvent).getSingle().then((r) => r?.lastSyncAt);
|
||||
}
|
||||
|
||||
/// Update stored connection event
|
||||
Future<void> updateConnectionInfo(Event event) async {
|
||||
final connectionInfo = await select(connectionEvent).getSingle();
|
||||
|
||||
return into(connectionEvent).insert(
|
||||
_ConnectionEventData(
|
||||
id: 1,
|
||||
lastSyncAt: connectionInfo?.lastSyncAt,
|
||||
lastEventAt: event.createdAt ?? connectionInfo?.lastEventAt,
|
||||
totalUnreadCount:
|
||||
event.totalUnreadCount ?? connectionInfo?.totalUnreadCount,
|
||||
ownUser: event.me?.toJson() ?? connectionInfo?.ownUser,
|
||||
unreadChannels: event.unreadChannels ?? connectionInfo?.unreadChannels,
|
||||
),
|
||||
mode: InsertMode.insertOrReplace,
|
||||
);
|
||||
}
|
||||
|
||||
/// Update stored lastSyncAt
|
||||
Future<void> updateLastSyncAt(DateTime lastSyncAt) async {
|
||||
return await (update(connectionEvent)..where((r) => r.id.equals(1))).write(
|
||||
_ConnectionEventCompanion(
|
||||
id: Value(1),
|
||||
lastSyncAt: Value(lastSyncAt),
|
||||
),
|
||||
);
|
||||
}
|
||||
|
||||
/// Get the channel cids saved in the offline storage
|
||||
Future<List<String>> getChannelCids() async {
|
||||
return (select(channels)
|
||||
..orderBy([(c) => OrderingTerm.desc(c.lastMessageAt)])
|
||||
..limit(250))
|
||||
.map((c) => c.cid)
|
||||
.get();
|
||||
}
|
||||
|
||||
/// Get channel data by cid
|
||||
Future<ChannelState> getChannel(
|
||||
String cid, {
|
||||
int limit,
|
||||
String messageLessThan,
|
||||
String messageGreaterThan,
|
||||
}) async {
|
||||
return await (select(channels)..where((c) => c.cid.equals(cid))).join([
|
||||
leftOuterJoin(users, channels.createdBy.equalsExp(users.id)),
|
||||
]).map((row) {
|
||||
return _channelFromRow(
|
||||
row.readTable(channels),
|
||||
row.readTable(users),
|
||||
limit: limit,
|
||||
messageLessThan: messageLessThan,
|
||||
messageGreaterThan: messageGreaterThan,
|
||||
);
|
||||
}).getSingle();
|
||||
}
|
||||
|
||||
/// Get list of channels by filter, sort and paginationParams
|
||||
Future<List<ChannelState>> getChannelStates({
|
||||
Map<String, dynamic> filter,
|
||||
List<SortOption> sort = const [],
|
||||
PaginationParams paginationParams,
|
||||
}) async {
|
||||
_logger.info('Get channel states');
|
||||
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.createdBy.equalsExp(users.id)),
|
||||
]).map((row) async {
|
||||
final userRow = row.readTable(users);
|
||||
final channelRow = row.readTable(channels);
|
||||
|
||||
return _channelFromRow(channelRow, userRow);
|
||||
}).get();
|
||||
}));
|
||||
|
||||
_logger.info('Got ${cachedChannels.length} channels');
|
||||
|
||||
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;
|
||||
}
|
||||
|
||||
/// 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(
|
||||
(_ChannelQueries query) => query.queryHash.equals(hash),
|
||||
))
|
||||
.go();
|
||||
}
|
||||
|
||||
return batch((batch) {
|
||||
batch.insertAll(
|
||||
channelQueries,
|
||||
cids.map((cid) {
|
||||
return ChannelQuery(
|
||||
queryHash: hash,
|
||||
channelCid: cid,
|
||||
);
|
||||
}).toList(),
|
||||
mode: InsertMode.insertOrReplace,
|
||||
);
|
||||
});
|
||||
}
|
||||
|
||||
/// Remove a message by message id
|
||||
Future<void> deleteMessages(List<String> messageIds) {
|
||||
return batch((batch) {
|
||||
batch.deleteWhere<_Reactions, _Reaction>(
|
||||
reactions,
|
||||
(r) => r.messageId.isIn(messageIds),
|
||||
);
|
||||
batch.deleteWhere<_Messages, _Message>(
|
||||
messages,
|
||||
(m) => m.id.isIn(messageIds),
|
||||
);
|
||||
});
|
||||
}
|
||||
|
||||
/// Remove a message by message id
|
||||
Future<void> deleteChannelsMessages(List<String> cids) async {
|
||||
final messageIds = await (select(messages)
|
||||
..where((m) => m.channelCid.isIn(cids)))
|
||||
.map((m) => m.id)
|
||||
.get();
|
||||
return batch((batch) {
|
||||
batch.deleteWhere<_Reactions, _Reaction>(
|
||||
reactions,
|
||||
(r) => r.messageId.isIn(messageIds),
|
||||
);
|
||||
batch.deleteWhere<_Messages, _Message>(
|
||||
messages,
|
||||
(m) => m.id.isIn(messageIds),
|
||||
);
|
||||
});
|
||||
}
|
||||
|
||||
/// Remove a channel by cid
|
||||
Future<void> deleteChannels(List<String> cids) async {
|
||||
await deleteChannelsMessages(cids);
|
||||
return batch((batch) {
|
||||
batch.deleteWhere<_Members, _Member>(
|
||||
members,
|
||||
(m) => m.channelCid.isIn(cids),
|
||||
);
|
||||
batch.deleteWhere<_Reads, _Read>(
|
||||
reads,
|
||||
(r) => r.channelCid.isIn(cids),
|
||||
);
|
||||
batch.deleteWhere<_Channels, _Channel>(
|
||||
channels,
|
||||
(c) => c.cid.isIn(cids),
|
||||
);
|
||||
});
|
||||
}
|
||||
|
||||
/// Update messages data from a list
|
||||
Future<void> updateMessages(
|
||||
List<Message> newMessages,
|
||||
String cid,
|
||||
) {
|
||||
return batch((batch) {
|
||||
batch.insertAll(
|
||||
messages,
|
||||
newMessages.map(
|
||||
(m) {
|
||||
return _Message(
|
||||
id: m.id,
|
||||
attachmentJson: m.attachments != null
|
||||
? jsonEncode(m.attachments.map((a) => a.toJson()).toList())
|
||||
: null,
|
||||
channelCid: cid,
|
||||
type: m.type,
|
||||
parentId: m.parentId,
|
||||
quotedMessageId: m.quotedMessageId,
|
||||
command: m.command,
|
||||
createdAt: m.createdAt,
|
||||
shadowed: m.shadowed,
|
||||
showInChannel: m.showInChannel,
|
||||
replyCount: m.replyCount,
|
||||
reactionScores: m.reactionScores,
|
||||
reactionCounts: m.reactionCounts,
|
||||
status: m.status,
|
||||
updatedAt: m.updatedAt,
|
||||
extraData: m.extraData,
|
||||
userId: m.user.id,
|
||||
deletedAt: m.deletedAt,
|
||||
messageText: m.text,
|
||||
);
|
||||
},
|
||||
).toList(),
|
||||
mode: InsertMode.insertOrReplace,
|
||||
);
|
||||
});
|
||||
}
|
||||
|
||||
/// Update single channel state
|
||||
Future<void> updateChannelState(ChannelState channelState) async {
|
||||
await updateChannelStates([channelState]);
|
||||
}
|
||||
|
||||
/// Update list of channel states
|
||||
Future<void> updateChannelStates(List<ChannelState> channelStates) async {
|
||||
channelStates.forEach((cs) {
|
||||
updateMessages(
|
||||
cs.messages,
|
||||
cs.channel.cid,
|
||||
);
|
||||
});
|
||||
|
||||
await batch((batch) {
|
||||
_updateReactions(batch, channelStates);
|
||||
|
||||
_updateUsers(batch, channelStates);
|
||||
|
||||
_updateReads(channelStates, batch);
|
||||
|
||||
_updateMembers(channelStates, batch);
|
||||
|
||||
_updateChannels(batch, channelStates);
|
||||
});
|
||||
}
|
||||
|
||||
/// Get the info about channel threads
|
||||
Future<Map<String, List<Message>>> getChannelThreads(String cid) async {
|
||||
final rowMessages = await 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());
|
||||
|
||||
final threads = <String, List<Message>>{};
|
||||
rowMessages.forEach((message) {
|
||||
if (threads.containsKey(message.parentId)) {
|
||||
threads[message.parentId].add(message);
|
||||
} else {
|
||||
threads[message.parentId] = [message];
|
||||
}
|
||||
});
|
||||
|
||||
return threads;
|
||||
}
|
||||
|
||||
Future<ChannelState> _channelFromRow(
|
||||
_Channel channelRow,
|
||||
_User userRow, {
|
||||
int limit,
|
||||
String messageLessThan,
|
||||
String messageGreaterThan,
|
||||
}) async {
|
||||
final rowMessages = await _getChannelMessages(
|
||||
channelRow,
|
||||
limit: limit,
|
||||
lessThan: messageLessThan,
|
||||
greaterThan: messageGreaterThan,
|
||||
);
|
||||
final rowReads = await _getChannelReads(channelRow);
|
||||
final rowMembers = await _getChannelMembers(channelRow);
|
||||
|
||||
return ChannelState(
|
||||
members: rowMembers,
|
||||
read: rowReads,
|
||||
messages: rowMessages,
|
||||
channel: ChannelModel(
|
||||
id: channelRow.id,
|
||||
type: channelRow.type,
|
||||
frozen: channelRow.frozen,
|
||||
createdAt: channelRow.createdAt,
|
||||
updatedAt: channelRow.updatedAt,
|
||||
memberCount: channelRow.memberCount,
|
||||
cid: channelRow.cid,
|
||||
lastMessageAt: channelRow.lastMessageAt,
|
||||
deletedAt: channelRow.deletedAt,
|
||||
extraData: channelRow.extraData,
|
||||
config: ChannelConfig.fromJson(jsonDecode(channelRow.config)),
|
||||
createdBy: userRow != null ? _userFromUserRow(userRow) : null,
|
||||
),
|
||||
);
|
||||
}
|
||||
|
||||
String _computeHash(Map<String, dynamic> filter) {
|
||||
if (filter == null) {
|
||||
return 'allchannels';
|
||||
}
|
||||
final hash = base64Encode(utf8.encode('filter: ${jsonEncode(filter)}'));
|
||||
return hash;
|
||||
}
|
||||
|
||||
Future<Message> _getMessageById(String id) async {
|
||||
if (id == null || id.isEmpty) return null;
|
||||
final message = await Future.wait(await (select(messages).join([
|
||||
leftOuterJoin(users, messages.userId.equalsExp(users.id)),
|
||||
])
|
||||
..where(messages.id.equals(id)))
|
||||
.map(_messageFromJoinRow)
|
||||
.get());
|
||||
return message.first;
|
||||
}
|
||||
|
||||
Future<List<Message>> _getChannelMessages(
|
||||
_Channel channelRow, {
|
||||
int limit,
|
||||
String lessThan,
|
||||
String greaterThan,
|
||||
}) async {
|
||||
final rowMessages = await Future.wait(await (select(messages).join([
|
||||
leftOuterJoin(users, messages.userId.equalsExp(users.id)),
|
||||
])
|
||||
..where(messages.channelCid.equals(channelRow.cid))
|
||||
..where(
|
||||
isNull(messages.parentId) | messages.showInChannel.equals(true))
|
||||
..orderBy([
|
||||
OrderingTerm.asc(messages.createdAt),
|
||||
]))
|
||||
.map(_messageFromJoinRow)
|
||||
.get());
|
||||
|
||||
if (lessThan != null) {
|
||||
final lessThanIndex = rowMessages.indexWhere((m) => m.id == lessThan);
|
||||
if (lessThanIndex != -1) {
|
||||
rowMessages.removeRange(lessThanIndex, rowMessages.length);
|
||||
}
|
||||
}
|
||||
if (greaterThan != null) {
|
||||
final greaterThanIndex =
|
||||
rowMessages.indexWhere((m) => m.id == greaterThan);
|
||||
if (greaterThanIndex != -1) {
|
||||
rowMessages.removeRange(0, greaterThanIndex);
|
||||
}
|
||||
}
|
||||
if (limit != null) {
|
||||
return rowMessages.take(limit).toList();
|
||||
}
|
||||
return rowMessages;
|
||||
}
|
||||
|
||||
Future<Message> _messageFromJoinRow(row) async {
|
||||
final messageRow = row.readTable(messages);
|
||||
final userRow = row.readTable(users);
|
||||
|
||||
final latestReactions = await _getLatestReactions(messageRow);
|
||||
final ownReactions = await _getOwnReactions(messageRow);
|
||||
final quotedMessage = await _getMessageById(messageRow.quotedMessageId);
|
||||
|
||||
return Message(
|
||||
shadowed: messageRow.shadowed,
|
||||
latestReactions: latestReactions,
|
||||
ownReactions: ownReactions,
|
||||
attachments: messageRow.attachmentJson != null
|
||||
? List<Map<String, dynamic>>.from(
|
||||
jsonDecode(messageRow.attachmentJson))
|
||||
.map((j) => Attachment.fromJson(j))
|
||||
.toList()
|
||||
: null,
|
||||
createdAt: messageRow.createdAt,
|
||||
extraData: messageRow.extraData,
|
||||
updatedAt: messageRow.updatedAt,
|
||||
id: messageRow.id,
|
||||
type: messageRow.type,
|
||||
status: messageRow.status,
|
||||
command: messageRow.command,
|
||||
parentId: messageRow.parentId,
|
||||
quotedMessageId: messageRow.quotedMessageId,
|
||||
quotedMessage: quotedMessage,
|
||||
reactionCounts: messageRow.reactionCounts,
|
||||
reactionScores: messageRow.reactionScores,
|
||||
replyCount: messageRow.replyCount,
|
||||
showInChannel: messageRow.showInChannel,
|
||||
text: messageRow.messageText,
|
||||
user: _userFromUserRow(userRow),
|
||||
deletedAt: messageRow.deletedAt,
|
||||
);
|
||||
}
|
||||
|
||||
Future<List<Reaction>> _getLatestReactions(_Message messageRow) async {
|
||||
return await (select(reactions).join([
|
||||
leftOuterJoin(users, reactions.userId.equalsExp(users.id)),
|
||||
])
|
||||
..where(reactions.messageId.equals(messageRow.id))
|
||||
..orderBy([OrderingTerm.asc(reactions.createdAt)]))
|
||||
.map((row) {
|
||||
final r = row.readTable(reactions);
|
||||
final u = row.readTable(users);
|
||||
return _reactionFromRow(r, u);
|
||||
}).get();
|
||||
}
|
||||
|
||||
Reaction _reactionFromRow(_Reaction r, _User u) {
|
||||
return Reaction(
|
||||
extraData: r.extraData,
|
||||
type: r.type,
|
||||
createdAt: r.createdAt,
|
||||
userId: r.userId,
|
||||
user: _userFromUserRow(u),
|
||||
messageId: r.messageId,
|
||||
score: r.score,
|
||||
);
|
||||
}
|
||||
|
||||
Future<List<Reaction>> _getOwnReactions(_Message messageRow) async {
|
||||
return await (select(reactions).join([
|
||||
leftOuterJoin(users, reactions.userId.equalsExp(users.id)),
|
||||
])
|
||||
..where(reactions.userId.equals(_userId))
|
||||
..where(reactions.messageId.equals(messageRow.id))
|
||||
..orderBy([OrderingTerm.asc(reactions.createdAt)]))
|
||||
.map((row) {
|
||||
final r = row.readTable(reactions);
|
||||
final u = row.readTable(users);
|
||||
return _reactionFromRow(r, u);
|
||||
}).get();
|
||||
}
|
||||
|
||||
Future<List<Read>> _getChannelReads(_Channel channelRow) async {
|
||||
final rowReads = await (select(reads).join([
|
||||
leftOuterJoin(users, reads.userId.equalsExp(users.id)),
|
||||
])
|
||||
..where(reads.channelCid.equals(channelRow.cid))
|
||||
..orderBy([
|
||||
OrderingTerm.asc(reads.lastRead),
|
||||
]))
|
||||
.map((row) {
|
||||
final userRow = row.readTable(users);
|
||||
final readRow = row.readTable(reads);
|
||||
return Read(
|
||||
user: _userFromUserRow(userRow),
|
||||
lastRead: readRow.lastRead,
|
||||
unreadMessages: readRow.unreadMessages,
|
||||
);
|
||||
}).get();
|
||||
return rowReads;
|
||||
}
|
||||
|
||||
Future<List<Member>> _getChannelMembers(_Channel channelRow) async {
|
||||
final rowMembers = await (select(members).join([
|
||||
leftOuterJoin(users, members.userId.equalsExp(users.id)),
|
||||
])
|
||||
..where(members.channelCid.equals(channelRow.cid))
|
||||
..orderBy([
|
||||
OrderingTerm.asc(members.createdAt),
|
||||
]))
|
||||
.map((row) {
|
||||
final userRow = row.readTable(users);
|
||||
final memberRow = row.readTable(members);
|
||||
return Member(
|
||||
user: _userFromUserRow(userRow),
|
||||
userId: userRow.id,
|
||||
banned: memberRow.banned,
|
||||
shadowBanned: memberRow.shadowBanned,
|
||||
updatedAt: memberRow.updatedAt,
|
||||
createdAt: memberRow.createdAt,
|
||||
role: memberRow.role,
|
||||
inviteAcceptedAt: memberRow.inviteAcceptedAt,
|
||||
invited: memberRow.invited,
|
||||
inviteRejectedAt: memberRow.inviteRejectedAt,
|
||||
isModerator: memberRow.isModerator,
|
||||
);
|
||||
}).get();
|
||||
return rowMembers;
|
||||
}
|
||||
|
||||
User _userFromUserRow(_User userRow) {
|
||||
return User(
|
||||
updatedAt: userRow.updatedAt,
|
||||
role: userRow.role,
|
||||
online: userRow.online,
|
||||
lastActive: userRow.lastActive,
|
||||
extraData: userRow.extraData,
|
||||
banned: userRow.banned,
|
||||
createdAt: userRow.createdAt,
|
||||
id: userRow.id,
|
||||
);
|
||||
}
|
||||
|
||||
void _updateChannels(Batch batch, List<ChannelState> channelStates) {
|
||||
batch.insertAll(
|
||||
channels,
|
||||
channelStates.map((cs) {
|
||||
final channel = cs.channel;
|
||||
return _channelDataFromChannelModel(channel);
|
||||
}).toList(),
|
||||
mode: InsertMode.insertOrReplace,
|
||||
);
|
||||
}
|
||||
|
||||
void _updateMembers(List<ChannelState> channelStates, Batch batch) async {
|
||||
await (delete(members)
|
||||
..where((tbl) =>
|
||||
tbl.channelCid.isIn(channelStates.map((e) => e.channel.cid))))
|
||||
.go();
|
||||
final newMembers = channelStates
|
||||
.map((cs) => cs.members.map((m) => _Member(
|
||||
userId: m.user.id,
|
||||
banned: m.banned,
|
||||
shadowBanned: m.shadowBanned,
|
||||
channelCid: cs.channel.cid,
|
||||
createdAt: m.createdAt,
|
||||
isModerator: m.isModerator,
|
||||
inviteRejectedAt: m.inviteRejectedAt,
|
||||
invited: m.invited,
|
||||
inviteAcceptedAt: m.inviteAcceptedAt,
|
||||
role: m.role,
|
||||
updatedAt: m.updatedAt,
|
||||
)))
|
||||
.where((v) => v != null)
|
||||
.expand((v) => v);
|
||||
if (newMembers != null && newMembers.isNotEmpty) {
|
||||
batch.insertAll(
|
||||
members,
|
||||
newMembers.toList(),
|
||||
mode: InsertMode.insertOrReplace,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
void _updateReads(List<ChannelState> channelStates, Batch batch) {
|
||||
final newReads = channelStates
|
||||
.map((cs) => cs.read?.map((r) => _Read(
|
||||
lastRead: r.lastRead,
|
||||
userId: r.user.id,
|
||||
channelCid: cs.channel.cid,
|
||||
unreadMessages: r.unreadMessages,
|
||||
)))
|
||||
.where((v) => v != null)
|
||||
.expand((v) => v);
|
||||
|
||||
if (newReads != null && newReads.isNotEmpty) {
|
||||
batch.insertAll(
|
||||
reads,
|
||||
newReads.toList(),
|
||||
mode: InsertMode.insertOrReplace,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
void _updateUsers(Batch batch, List<ChannelState> channelStates) {
|
||||
batch.insertAll(
|
||||
users,
|
||||
channelStates
|
||||
.map((cs) => [
|
||||
if (cs.channel.createdBy != null)
|
||||
_userDataFromUser(cs.channel.createdBy),
|
||||
if (cs.messages != null)
|
||||
...cs.messages
|
||||
.map((m) => [
|
||||
_userDataFromUser(m.user),
|
||||
if (m.latestReactions != null)
|
||||
...m.latestReactions
|
||||
.where((r) => r.user != null)
|
||||
.map((r) => _userDataFromUser(r.user)),
|
||||
if (m.ownReactions != null)
|
||||
...m.ownReactions
|
||||
.where((r) => r.user != null)
|
||||
.map((r) => _userDataFromUser(r.user)),
|
||||
])
|
||||
.expand((v) => v),
|
||||
if (cs.read != null)
|
||||
...cs.read.map((r) => _userDataFromUser(r.user)),
|
||||
if (cs.members != null)
|
||||
...cs.members.map((m) => _userDataFromUser(m.user)),
|
||||
])
|
||||
.expand((v) => v)
|
||||
.toList(),
|
||||
mode: InsertMode.insertOrReplace,
|
||||
);
|
||||
}
|
||||
|
||||
void _updateReactions(Batch batch, List<ChannelState> channelStates) {
|
||||
batch.deleteWhere<_Reactions, _Reaction>(
|
||||
reactions,
|
||||
(r) => r.messageId.isIn(channelStates
|
||||
.map((cs) => cs.messages.map((m) => m.id))
|
||||
.expand((v) => v)),
|
||||
);
|
||||
final newReactions = channelStates
|
||||
.map((cs) => cs.messages.map((m) {
|
||||
final ownReactions =
|
||||
m.ownReactions?.where((e) => e.userId != null)?.map(
|
||||
(r) => _reactionDataFromReaction(m, r),
|
||||
) ??
|
||||
[];
|
||||
final latestReactions =
|
||||
m.latestReactions?.where((e) => e.userId != null)?.map(
|
||||
(r) => _reactionDataFromReaction(m, r),
|
||||
) ??
|
||||
[];
|
||||
return [
|
||||
...ownReactions,
|
||||
...latestReactions,
|
||||
];
|
||||
}).expand((v) => v))
|
||||
.expand((v) => v);
|
||||
|
||||
if (newReactions.isNotEmpty) {
|
||||
batch.insertAll(
|
||||
reactions,
|
||||
newReactions.toList(),
|
||||
mode: InsertMode.insertOrReplace,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
_Reaction _reactionDataFromReaction(Message m, Reaction r) {
|
||||
return _Reaction(
|
||||
messageId: m.id,
|
||||
type: r.type,
|
||||
extraData: r.extraData,
|
||||
score: r.score,
|
||||
createdAt: r.createdAt,
|
||||
userId: r.userId,
|
||||
);
|
||||
}
|
||||
|
||||
_User _userDataFromUser(User user) {
|
||||
return _User(
|
||||
id: user.id,
|
||||
createdAt: user.createdAt,
|
||||
banned: user.banned,
|
||||
extraData: user.extraData,
|
||||
lastActive: user.lastActive,
|
||||
online: user.online,
|
||||
role: user.role,
|
||||
updatedAt: user.updatedAt,
|
||||
);
|
||||
}
|
||||
|
||||
_Channel _channelDataFromChannelModel(ChannelModel channel) {
|
||||
return _Channel(
|
||||
id: channel.id,
|
||||
config: jsonEncode(channel.config?.toJson() ?? {}),
|
||||
type: channel.type,
|
||||
frozen: channel.frozen,
|
||||
createdAt: channel.createdAt,
|
||||
updatedAt: channel.updatedAt,
|
||||
memberCount: channel.memberCount,
|
||||
cid: channel.cid,
|
||||
lastMessageAt: channel.lastMessageAt,
|
||||
deletedAt: channel.deletedAt,
|
||||
extraData: channel.extraData,
|
||||
createdBy: channel.createdBy?.id,
|
||||
);
|
||||
}
|
||||
}
|
||||
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,64 @@
|
||||
import 'dart:io';
|
||||
import 'dart:isolate';
|
||||
import 'package:flutter/material.dart';
|
||||
import 'package:moor/ffi.dart';
|
||||
import 'package:moor/isolate.dart';
|
||||
import 'package:moor/moor.dart';
|
||||
import 'package:path/path.dart';
|
||||
import 'package:path_provider/path_provider.dart';
|
||||
import 'package:stream_chat/src/db/offline_storage.dart';
|
||||
|
||||
class SharedDB {
|
||||
static constructDatabase(dbName) async {
|
||||
final dir = await getApplicationDocumentsDirectory();
|
||||
final path = join(dir.path, dbName);
|
||||
final file = File(path);
|
||||
return VmDatabase(file);
|
||||
}
|
||||
|
||||
static Future<MoorIsolate> createMoorIsolate(String userId) async {
|
||||
WidgetsFlutterBinding.ensureInitialized();
|
||||
final dir = await getApplicationDocumentsDirectory();
|
||||
final path = join(dir.path, 'db_$userId.sqlite');
|
||||
|
||||
final receivePort = ReceivePort();
|
||||
await Isolate.spawn(
|
||||
startBackground,
|
||||
_IsolateStartRequest(receivePort.sendPort, path),
|
||||
);
|
||||
|
||||
return (await receivePort.first as MoorIsolate);
|
||||
}
|
||||
|
||||
static void startBackground(_IsolateStartRequest request) {
|
||||
final executor = LazyDatabase(() async {
|
||||
return VmDatabase(File(request.targetPath));
|
||||
});
|
||||
final moorIsolate = MoorIsolate.inCurrent(
|
||||
() => DatabaseConnection.fromExecutor(executor),
|
||||
);
|
||||
request.sendMoorIsolate.send(moorIsolate);
|
||||
}
|
||||
|
||||
static Future<OfflineStorage> constructOfflineStorage({
|
||||
userId,
|
||||
logger,
|
||||
}) async {
|
||||
logger.info('Connecting on background isolate');
|
||||
final isolate = await createMoorIsolate(userId);
|
||||
final connection = await isolate.connect();
|
||||
return OfflineStorage.connect(
|
||||
connection,
|
||||
userId,
|
||||
isolate,
|
||||
logger,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
class _IsolateStartRequest {
|
||||
final SendPort sendMoorIsolate;
|
||||
final String targetPath;
|
||||
|
||||
_IsolateStartRequest(this.sendMoorIsolate, this.targetPath);
|
||||
}
|
||||
@@ -0,0 +1,3 @@
|
||||
export 'unsupported_db.dart'
|
||||
if (dart.library.io) 'native_db.dart' // implementation using dart:io
|
||||
if (dart.library.html) 'web_db.dart';
|
||||
@@ -0,0 +1,12 @@
|
||||
class SharedDB {
|
||||
static constructDatabase(dbName) async {
|
||||
print('Unsupported Platform');
|
||||
return null;
|
||||
}
|
||||
|
||||
static createMoorIsolate(userId) {}
|
||||
|
||||
static startBackground(request) {}
|
||||
|
||||
static constructOfflineStorage({userId, logger}) {}
|
||||
}
|
||||
@@ -0,0 +1,18 @@
|
||||
import 'package:moor/moor_web.dart';
|
||||
import 'package:stream_chat/src/db/offline_storage.dart';
|
||||
|
||||
class SharedDB {
|
||||
static constructDatabase(dbName) async {
|
||||
return WebDatabase(dbName);
|
||||
}
|
||||
|
||||
static Future<OfflineStorage> constructOfflineStorage({
|
||||
userId,
|
||||
logger,
|
||||
}) async {
|
||||
return OfflineStorage(
|
||||
userId,
|
||||
logger,
|
||||
);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user