diff --git a/lib/app/features/chat/providers/conversation_management_provider.c.dart b/lib/app/features/chat/providers/conversation_management_provider.c.dart index 9a2708fd9..33fb30664 100644 --- a/lib/app/features/chat/providers/conversation_management_provider.c.dart +++ b/lib/app/features/chat/providers/conversation_management_provider.c.dart @@ -4,13 +4,13 @@ import 'package:hooks_riverpod/hooks_riverpod.dart'; import 'package:ion/app/exceptions/exceptions.dart'; import 'package:ion/app/features/chat/model/entities/private_direct_message_data.c.dart'; import 'package:ion/app/features/chat/providers/conversation_message_management_provider.c.dart'; -import 'package:ion/app/services/database/ion_database.c.dart'; +import 'package:ion/app/services/database/conversation_db_service.c.dart'; import 'package:riverpod_annotation/riverpod_annotation.dart'; part 'conversation_management_provider.c.g.dart'; @Riverpod(keepAlive: true) -Future conversationManagementService( +Future> conversationManagementService( Ref ref, ) async { final databaseService = ref.watch(conversationsDBServiceProvider); diff --git a/lib/app/features/chat/providers/conversation_message_actions_provider.c.dart b/lib/app/features/chat/providers/conversation_message_actions_provider.c.dart new file mode 100644 index 000000000..d2e674dbd --- /dev/null +++ b/lib/app/features/chat/providers/conversation_message_actions_provider.c.dart @@ -0,0 +1,246 @@ +// SPDX-License-Identifier: ice License 1.0 + +import 'dart:convert'; + +import 'package:hooks_riverpod/hooks_riverpod.dart'; +import 'package:ion/app/exceptions/exceptions.dart'; +import 'package:ion/app/features/auth/providers/auth_provider.c.dart'; +import 'package:ion/app/features/chat/model/entities/private_direct_message_data.c.dart'; +import 'package:ion/app/features/chat/model/entities/private_message_reaction_data.c.dart'; +import 'package:ion/app/features/chat/providers/conversation_message_management_provider.c.dart'; +import 'package:ion/app/features/core/providers/env_provider.c.dart'; +import 'package:ion/app/features/feed/data/models/bookmarks/bookmarks.c.dart'; +import 'package:ion/app/features/nostr/model/entity_expiration.c.dart'; +import 'package:ion/app/features/nostr/model/nostr_entity.dart'; +import 'package:ion/app/features/nostr/providers/nostr_event_signer_provider.c.dart'; +import 'package:ion/app/features/nostr/providers/nostr_notifier.c.dart'; +import 'package:ion/app/services/database/conversation_db_service.c.dart'; +import 'package:ion/app/services/logger/logger.dart'; +import 'package:ion/app/services/nostr/ion_connect_gift_wrap_service.c.dart'; +import 'package:ion/app/services/nostr/ion_connect_seal_service.c.dart'; +import 'package:nip44/nip44.dart'; +import 'package:nostr_dart/nostr_dart.dart'; +import 'package:riverpod_annotation/riverpod_annotation.dart'; + +part 'conversation_message_actions_provider.c.g.dart'; + +@Riverpod(keepAlive: true) +Future> conversationMessageActionsService( + Ref ref, +) async { + final databaseService = ref.watch(conversationsDBServiceProvider); + final conversationMessageManagementService = + ref.watch(conversationMessageManagementServiceProvider).requireValue; + + final eventSigner = await ref.watch(currentUserNostrEventSignerProvider.future); + + return ConversationMessageActionsService( + eventSigner: eventSigner, + databaseService: databaseService, + env: ref.watch(envProvider.notifier), + userPubkey: ref.watch(currentPubkeySelectorProvider), + sealService: ref.watch(ionConnectSealServiceProvider), + nostrNotifier: ref.watch(nostrNotifierProvider.notifier), + wrapService: ref.watch(ionConnectGiftWrapServiceProvider), + conversationMessageManagementService: conversationMessageManagementService, + ); +} + +class ConversationMessageActionsService { + ConversationMessageActionsService({ + required this.env, + required this.userPubkey, + required this.wrapService, + required this.sealService, + required this.eventSigner, + required this.nostrNotifier, + required this.databaseService, + required this.conversationMessageManagementService, + }); + + final Env env; + final String? userPubkey; + final EventSigner? eventSigner; + final NostrNotifier nostrNotifier; + final IonConnectSealService sealService; + final IonConnectGiftWrapService wrapService; + final ConversationsDBService databaseService; + final ConversationMessageManagementService conversationMessageManagementService; + + Future deleteMessage(String id) async { + await databaseService.markConversationMessageAsDeleted(id); + } + + Future bookmarkMessage(List ids, String receiverPubkey) async { + if (eventSigner == null) { + throw EventSignerNotFoundException(); + } + + final createdAt = DateTime.now().toUtc(); + + final encodedRumor = jsonEncode([ + ids.map((id) => ['e', id]).toList(), + ]); + + final tags = [ + ['d', 'chat_messages'], + ['encrypted'], + ]; + + Logger.log('Encoded rumor $encodedRumor'); + + final encryptedRumor = await Nip44.encryptMessage( + encodedRumor, + eventSigner!.privateKey, + receiverPubkey, + ); + + Logger.log('Encrypted rumor $encryptedRumor'); + + final id = EventMessage.calculateEventId( + tags: tags, + createdAt: createdAt, + content: encryptedRumor, + publicKey: receiverPubkey, + kind: BookmarksEntity.kind, + ); + + final bookmarkMessage = EventMessage( + id: id, + tags: tags, + createdAt: createdAt, + content: encryptedRumor, + pubkey: receiverPubkey, + kind: BookmarksEntity.kind, + sig: null, + ); + + Logger.log('Bookmark message $bookmarkMessage'); + + await nostrNotifier.sendEvent(bookmarkMessage, cache: false); + } + + Future sendMessageReaction({ + required String eventId, + required String reaction, + required String receiverPubkey, + }) async { + if (eventSigner == null) { + throw EventSignerNotFoundException(); + } + + await _createSealWrapSendReaction( + content: reaction, + signer: eventSigner!, + receiverPubkey: receiverPubkey, + kind: PrivateMessageReactionEntity.kind, + tags: [ + ['k', PrivateDirectMessageEntity.kind.toString()], + ['p', receiverPubkey], + ['e', eventId], + ], + ); + } + + Future sendMessageReceivedStatus({ + required String eventId, + required String receiverPubkey, + }) async { + if (eventSigner == null) { + throw EventSignerNotFoundException(); + } + + await _createSealWrapSendReaction( + signer: eventSigner!, + content: 'received', + receiverPubkey: receiverPubkey, + kind: PrivateMessageReactionEntity.kind, + tags: [ + ['k', PrivateDirectMessageEntity.kind.toString()], + ['p', receiverPubkey], + ['e', eventId], + ], + ); + } + + Future sendMessageReadStatus({ + required String lastMessageId, + required String receiverPubkey, + }) async { + if (eventSigner == null) { + throw EventSignerNotFoundException(); + } + + await _createSealWrapSendReaction( + signer: eventSigner!, + content: 'read', + receiverPubkey: receiverPubkey, + kind: PrivateMessageReactionEntity.kind, + tags: [ + ['k', PrivateDirectMessageEntity.kind.toString()], + ['p', receiverPubkey], + ['e', lastMessageId], + ], + ); + } + + Future _createSealWrapSendReaction({ + required String content, + required EventSigner signer, + required String receiverPubkey, + required List> tags, + int? kind, + }) async { + final createdAt = DateTime.now().toUtc(); + + final id = EventMessage.calculateEventId( + tags: tags, + content: content, + createdAt: createdAt, + publicKey: signer.publicKey, + kind: PrivateMessageReactionEntity.kind, + ); + + final eventMessage = EventMessage( + id: id, + tags: tags, + content: content, + createdAt: createdAt, + pubkey: signer.publicKey, + kind: PrivateMessageReactionEntity.kind, + sig: await signer.sign(message: id), + ); + + Logger.log('Event message $eventMessage'); + + final seal = await sealService.createSeal( + eventMessage, + signer, + receiverPubkey, + ); + + Logger.log('Seal message $seal'); + + final expirationTag = EntityExpiration( + value: DateTime.now().add( + Duration(hours: env.get(EnvVariable.STORY_EXPIRATION_HOURS)), + ), + ).toTag(); + + final wrap = await wrapService.createWrap( + seal, + receiverPubkey, + signer, + PrivateMessageReactionEntity.kind, + expirationTag: expirationTag, + ); + + Logger.log('Wrap message $wrap'); + + final result = await nostrNotifier.sendEvent(wrap, cache: false); + + Logger.log('Sent message $result'); + + return result; + } +} diff --git a/lib/app/features/chat/providers/conversation_message_management_provider.c.dart b/lib/app/features/chat/providers/conversation_message_management_provider.c.dart index 2988a0154..006a09673 100644 --- a/lib/app/features/chat/providers/conversation_message_management_provider.c.dart +++ b/lib/app/features/chat/providers/conversation_message_management_provider.c.dart @@ -9,7 +9,6 @@ import 'package:cryptography/cryptography.dart'; import 'package:hooks_riverpod/hooks_riverpod.dart'; import 'package:ion/app/exceptions/exceptions.dart'; import 'package:ion/app/features/chat/model/entities/private_direct_message_data.c.dart'; -import 'package:ion/app/features/chat/model/entities/private_message_reaction_data.c.dart'; import 'package:ion/app/features/core/model/media_type.dart'; import 'package:ion/app/features/core/providers/env_provider.c.dart'; import 'package:ion/app/features/nostr/model/entity_expiration.c.dart'; @@ -30,7 +29,7 @@ import 'package:riverpod_annotation/riverpod_annotation.dart'; part 'conversation_message_management_provider.c.g.dart'; @Riverpod(keepAlive: true) -Future conversationMessageManagementService( +Future> conversationMessageManagementService( Ref ref, ) async { final eventSigner = await ref.watch(currentUserNostrEventSignerProvider.future); @@ -141,27 +140,6 @@ class ConversationMessageManagementService { } } - Future sendMessageReceivedStatus({ - required String eventId, - required String receiverPubkey, - }) async { - if (eventSigner == null) { - throw EventSignerNotFoundException(); - } - - await _createSealWrapSendMessage( - signer: eventSigner!, - content: 'received', - receiverPubkey: receiverPubkey, - kind: PrivateMessageReactionEntity.kind, - tags: [ - ['k', PrivateDirectMessageEntity.kind.toString()], - ['p', receiverPubkey], - ['e', eventId], - ], - ); - } - // Works in progress with https://pub.dev/packages/flutter_cache_manager Future> downloadDecryptDecompressMedia( PrivateDirectMessageEntity privateDirectMessageEntity, @@ -239,14 +217,13 @@ class ConversationMessageManagementService { required String receiverPubkey, required EventSigner signer, required List> tags, - int? kind, }) async { final createdAt = DateTime.now().toUtc(); final id = EventMessage.calculateEventId( publicKey: signer.publicKey, createdAt: createdAt, - kind: kind ?? PrivateDirectMessageEntity.kind, + kind: PrivateDirectMessageEntity.kind, tags: tags, content: content, ); @@ -257,7 +234,7 @@ class ConversationMessageManagementService { content: content, createdAt: createdAt, pubkey: signer.publicKey, - kind: kind ?? PrivateDirectMessageEntity.kind, + kind: PrivateDirectMessageEntity.kind, sig: null, ); @@ -271,20 +248,11 @@ class ConversationMessageManagementService { Logger.log('Seal message $seal'); - final expirationTag = EntityExpiration( - value: DateTime.now().add( - Duration( - hours: env.get(EnvVariable.STORY_EXPIRATION_HOURS), - ), - ), - ).toTag(); - final wrap = await wrapService.createWrap( seal, receiverPubkey, signer, - kind ?? PrivateDirectMessageEntity.kind, - expirationTag: kind == PrivateDirectMessageEntity.kind ? expirationTag : null, + PrivateDirectMessageEntity.kind, ); Logger.log('Wrap message $wrap'); diff --git a/lib/app/features/chat/providers/fetch_conversation_provider.c.dart b/lib/app/features/chat/providers/fetch_conversation_provider.c.dart index dc96c5344..803e866e9 100644 --- a/lib/app/features/chat/providers/fetch_conversation_provider.c.dart +++ b/lib/app/features/chat/providers/fetch_conversation_provider.c.dart @@ -7,7 +7,7 @@ import 'package:ion/app/features/chat/model/entities/private_message_reaction_da import 'package:ion/app/features/nostr/model/action_source.dart'; import 'package:ion/app/features/nostr/providers/nostr_event_signer_provider.c.dart'; import 'package:ion/app/features/nostr/providers/nostr_notifier.c.dart'; -import 'package:ion/app/services/database/ion_database.c.dart'; +import 'package:ion/app/services/database/conversation_db_service.c.dart'; import 'package:ion/app/services/nostr/ion_connect_gift_wrap_service.c.dart'; import 'package:ion/app/services/nostr/ion_connect_seal_service.c.dart'; import 'package:nostr_dart/nostr_dart.dart'; diff --git a/lib/app/features/chat/recent_chats/providers/conversations_provider.c.dart b/lib/app/features/chat/recent_chats/providers/conversations_provider.c.dart index 2b56fca81..125177930 100644 --- a/lib/app/features/chat/recent_chats/providers/conversations_provider.c.dart +++ b/lib/app/features/chat/recent_chats/providers/conversations_provider.c.dart @@ -6,7 +6,7 @@ import 'package:ion/app/features/chat/model/entities/private_direct_message_data import 'package:ion/app/features/chat/model/group.c.dart'; import 'package:ion/app/features/chat/model/message_author.c.dart'; import 'package:ion/app/features/chat/providers/mock.dart'; -import 'package:ion/app/services/database/ion_database.c.dart'; +import 'package:ion/app/services/database/conversation_db_service.c.dart'; import 'package:riverpod_annotation/riverpod_annotation.dart'; part 'conversations_provider.c.g.dart'; diff --git a/lib/app/services/database/conversation_db_service.c.dart b/lib/app/services/database/conversation_db_service.c.dart new file mode 100644 index 000000000..41cd0a586 --- /dev/null +++ b/lib/app/services/database/conversation_db_service.c.dart @@ -0,0 +1,352 @@ +// SPDX-License-Identifier: ice License 1.0 + +import 'package:drift/drift.dart'; +import 'package:hooks_riverpod/hooks_riverpod.dart'; +import 'package:ion/app/exceptions/exceptions.dart'; +import 'package:ion/app/features/chat/model/entities/private_direct_message_data.c.dart'; +import 'package:ion/app/features/chat/model/entities/private_message_reaction_data.c.dart'; +import 'package:ion/app/services/database/ion_database.c.dart'; +import 'package:nostr_dart/nostr_dart.dart'; +import 'package:riverpod_annotation/riverpod_annotation.dart'; +import 'package:uuid/uuid.dart'; + +part 'conversation_db_service.c.g.dart'; + +@Riverpod(keepAlive: true) +ConversationsDBService conversationsDBService(Ref ref) => + ConversationsDBService(ref.watch(ionDatabaseProvider)); + +class ConversationsDBService { + ConversationsDBService(this._db); + + final IONDatabase _db; + + Future _insertConversationData({ + required String conversationId, + required EventMessage eventMessage, + bool isDeleted = false, + }) { + final conversationMessage = PrivateDirectMessageEntity.fromEventMessage(eventMessage); + return _db.into(_db.conversationMessagesTable).insert( + ConversationMessagesTableData( + isDeleted: isDeleted, + conversationId: conversationId, + eventMessageId: conversationMessage.id, + createdAt: conversationMessage.createdAt, + pubKeys: conversationMessage.allPubkeysMask, + subject: conversationMessage.data.relatedSubject?.value, + status: DeliveryStatus.none, + ), + mode: InsertMode.insertOrReplace, + ); + } + + Future _insertConversationReactionsTableData(EventMessage eventMessage) { + final reactionEntity = PrivateMessageReactionEntity.fromEventMessage(eventMessage); + + return _db.into(_db.conversationReactionsTable).insert( + ConversationReactionsTableData( + reactionEventId: reactionEntity.id, + createdAt: reactionEntity.createdAt, + content: reactionEntity.data.content, + messageId: reactionEntity.data.eventId, + ), + mode: InsertMode.insertOrReplace, + ); + } + + // TODO: Try to implement in a single transaction + //https://drift.simonbinder.eu/dart_api/transactions/?h=transa + Future insertEventMessage(EventMessage eventMessage) async { + await _db.into(_db.eventMessagesTable).insert( + EventMessageTableData.fromEventMessage(eventMessage), + mode: InsertMode.insertOrReplace, + ); + + await _db.transaction(() async { + return true; + }); + + if (eventMessage.kind == PrivateDirectMessageEntity.kind) { + await _updateConversationMessagesTable(eventMessage); + } else if (eventMessage.kind == PrivateMessageReactionEntity.kind) { + switch (eventMessage.content) { + case 'received': + await _updateConversationMessageAsReceived(eventMessage); + case 'read': + await _updateConversationMessagesAsRead(eventMessage); + default: + await _insertConversationReactionsTableData(eventMessage); + } + } + } + + Future _updateConversationMessagesTable( + EventMessage eventMessage, + ) async { + if (eventMessage.kind != PrivateDirectMessageEntity.kind) return; + + final conversationMessage = PrivateDirectMessageEntity.fromEventMessage(eventMessage); + final conversationIdByPubkeys = await _lookupConversationByPubkeys(conversationMessage); + + if (conversationIdByPubkeys != null) { + // Existing conversation (one-to-one or group) + await _insertConversationData( + eventMessage: eventMessage, + conversationId: conversationIdByPubkeys, + ); + } else { + // Existing group conversation (change of participants) + final conversationIdBySubject = await _lookupConversationBySubject(conversationMessage); + + if (conversationIdBySubject != null) { + await _insertConversationData( + eventMessage: eventMessage, + conversationId: conversationIdBySubject, + ); + } else if (eventMessage.content.isEmpty) { + // New conversation + final uuid = const Uuid().v1(); + await _insertConversationData( + eventMessage: eventMessage, + conversationId: uuid, + ); + } else { + // Invalid message (doesn't belong to any conversation) + throw ConversationIsNotFoundException(); + } + } + } + + // Check if there are conversations with the same pubkeys + Future _lookupConversationByPubkeys( + PrivateDirectMessageEntity conversationMessage, + ) async { + final conversationsWithSameParticipants = await (_db.select(_db.conversationMessagesTable) + ..where( + (table) => table.pubKeys.equals(conversationMessage.allPubkeysMask), + ) + ..limit(1)) + .get(); + + if (conversationsWithSameParticipants.isNotEmpty) { + return conversationsWithSameParticipants.first.conversationId; + } + + return null; + } + + // Check if there are conversations with the same subject but different pubkeys + // (this means that amount of participants in conversation was changed) + Future _lookupConversationBySubject( + PrivateDirectMessageEntity conversationMessage, + ) async { + final subject = conversationMessage.data.relatedSubject?.value; + + if (subject == null) return null; + + final conversationWithChangedParticipants = await (_db.select(_db.conversationMessagesTable) + ..where((table) => table.subject.equals(subject)) + ..limit(1)) + .get(); + + if (conversationWithChangedParticipants.isNotEmpty) { + return conversationWithChangedParticipants.first.conversationId; + } + + return null; + } + + // Call when "OK" is received from relay to mark message as sent (one tick) + Future markConversationMessageAsSent(String id) async { + final conversationMessagesTableData = await (_db.select(_db.conversationMessagesTable) + ..where((table) => table.eventMessageId.equals(id))) + .getSingle(); + + final sentConversationMessagesTableData = + conversationMessagesTableData.copyWith(status: DeliveryStatus.isSent); + + await _db.update(_db.conversationMessagesTable).replace(sentConversationMessagesTableData); + } + + Future markConversationMessageAsDeleted(String id) async { + final conversationMessagesTableData = await (_db.select(_db.conversationMessagesTable) + ..where((table) => table.eventMessageId.equals(id))) + .getSingle(); + + final deleteConversationMessagesTableData = + conversationMessagesTableData.copyWith(isDeleted: true); + + await _db.update(_db.conversationMessagesTable).replace(deleteConversationMessagesTableData); + } + + // Call when kind 7 is received from relay with "received" content + Future _updateConversationMessageAsReceived( + EventMessage eventMessage, + ) async { + final reactionEntity = PrivateMessageReactionEntity.fromEventMessage(eventMessage); + + final conversationMessagesTableData = await (_db.select(_db.conversationMessagesTable) + ..where( + (table) => table.eventMessageId.equals(reactionEntity.data.eventId), + )) + .getSingle(); + + final receivedConversationMessagesTableData = + conversationMessagesTableData.copyWith(status: DeliveryStatus.isReceived); + + await _db.update(_db.conversationMessagesTable).replace(receivedConversationMessagesTableData); + } + + // Call when kind 7 is received from relay with "read" content + Future _updateConversationMessagesAsRead( + EventMessage eventMessage, + ) async { + final reactionEntity = PrivateMessageReactionEntity.fromEventMessage(eventMessage); + + final latestConversationMessageTableData = await (_db.select(_db.conversationMessagesTable) + ..where( + (table) => table.eventMessageId.equals(reactionEntity.data.eventId), + )) + .getSingle(); + + final allPreviousReceivedMessages = await (_db.select(_db.conversationMessagesTable) + ..where( + (table) => + table.conversationId.equals(latestConversationMessageTableData.conversationId), + ) + ..where( + (table) => table.status.equals(DeliveryStatus.isReceived.index), + ) + ..where( + (table) => table.createdAt.isSmallerOrEqualValue( + latestConversationMessageTableData.createdAt, + ), + )) + .get(); + + await _db.batch( + (b) { + b.replaceAll( + _db.conversationMessagesTable, + allPreviousReceivedMessages + .map( + (previousMessage) => previousMessage.copyWith(status: DeliveryStatus.isRead), + ) + .toList(), + ); + }, + ); + } + + final _allConversationsLatestMessageQuery = + 'SELECT * FROM (SELECT * FROM conversation_messages_table ORDER BY created_at DESC) AS sub GROUP BY conversation_id'; + + Future> getAllConversations() async { + // Select last message of each conversation + final uniqueConversationRows = await _db.customSelect( + _allConversationsLatestMessageQuery, + readsFrom: {_db.conversationMessagesTable}, + ).get(); + + final lastConversationEventMessages = + await _selectLastMessageOfEachConversation(uniqueConversationRows); + + return lastConversationEventMessages; + } + + Stream> watchConversations() { + return _db + .customSelect( + _allConversationsLatestMessageQuery, + readsFrom: {_db.conversationMessagesTable}, + ) + .watch() + .asyncMap((uniqueConversationRows) async { + final lastConversationEventMessages = await _selectLastMessageOfEachConversation( + uniqueConversationRows, + ); + + return lastConversationEventMessages; + }); + } + + Future> _selectLastMessageOfEachConversation( + List uniqueConversationRows, + ) async { + final lastConversationMessagesIds = + uniqueConversationRows.map((row) => row.data['event_message_id'] as String).toList(); + + final lastConversationEventMessages = (await (_db.select(_db.eventMessagesTable) + ..where((table) => table.id.isIn(lastConversationMessagesIds))) + .get()) + .map((e) => e.toEventMessage()) + .toList(); + + return lastConversationEventMessages; + } + + Future> getMessageReactions( + String messageId, + ) async { + final reactionsEventMessagesIds = (await (_db.select(_db.conversationReactionsTable) + ..where( + (table) => table.messageId.equals(messageId), + )) + .get()) + .map((reactionsTableData) => reactionsTableData.reactionEventId) + .toList(); + + final reactionsEventMessages = await (_db.select(_db.eventMessagesTable) + ..where((table) => table.id.isIn(reactionsEventMessagesIds))) + .get(); + + final reactions = reactionsEventMessages + .map( + (reactionEventMessageData) => PrivateMessageReactionEntity.fromEventMessage( + reactionEventMessageData.toEventMessage(), + ), + ) + .toList(); + + return reactions; + } + + //get last createdAt date from conversation_messages_table + Future getLastConversationMessageCreatedAt() async { + final lastConversationMessage = await (_db.select(_db.conversationMessagesTable) + ..orderBy([(table) => OrderingTerm.desc(table.createdAt)]) + ..limit(1)) + .getSingleOrNull(); + + return lastConversationMessage?.createdAt; + } + + // Mark conversation as removed and all its messages prior to last message as + // deleted + Future deleteConversation(String conversationId) async { + await _db.into(_db.deletedConversationTable).insert( + DeletedConversationTableCompanion( + conversationId: Value(conversationId), + deletedAt: Value(DateTime.now().toUtc()), + ), + mode: InsertMode.insertOrReplace, + ); + + final conversationMessagesTableData = await (_db.select(_db.conversationMessagesTable) + ..where((table) => table.conversationId.equals(conversationId))) + .get(); + + final deleteConversationMessagesTableData = + conversationMessagesTableData.map((e) => e.copyWith(isDeleted: true)).toList(); + + await _db.batch( + (b) { + b.replaceAll( + _db.conversationMessagesTable, + deleteConversationMessagesTableData, + ); + }, + ); + } +} diff --git a/lib/app/services/database/ion_database.c.dart b/lib/app/services/database/ion_database.c.dart index 48039ac10..46bdd8efe 100644 --- a/lib/app/services/database/ion_database.c.dart +++ b/lib/app/services/database/ion_database.c.dart @@ -5,311 +5,14 @@ import 'dart:convert'; import 'package:drift/drift.dart'; import 'package:drift_flutter/drift_flutter.dart'; import 'package:hooks_riverpod/hooks_riverpod.dart'; -import 'package:ion/app/exceptions/exceptions.dart'; -import 'package:ion/app/features/chat/model/entities/private_direct_message_data.c.dart'; -import 'package:ion/app/features/chat/model/entities/private_message_reaction_data.c.dart'; import 'package:nostr_dart/nostr_dart.dart'; import 'package:riverpod_annotation/riverpod_annotation.dart'; -import 'package:uuid/uuid.dart'; part 'ion_database.c.g.dart'; @Riverpod(keepAlive: true) IONDatabase ionDatabase(Ref ref) => IONDatabase(); -@Riverpod(keepAlive: true) -ConversationsDBService conversationsDBService(Ref ref) => - ConversationsDBService(ref.watch(ionDatabaseProvider)); - -class ConversationsDBService { - ConversationsDBService(this._db); - - final IONDatabase _db; - - Future _insertConversationData({ - required String conversationId, - required EventMessage eventMessage, - bool isDeleted = false, - }) { - final conversationMessage = PrivateDirectMessageEntity.fromEventMessage(eventMessage); - return _db.into(_db.conversationMessagesTable).insert( - ConversationMessagesTableData( - isDeleted: isDeleted, - conversationId: conversationId, - eventMessageId: conversationMessage.id, - createdAt: conversationMessage.createdAt, - pubKeys: conversationMessage.allPubkeysMask, - subject: conversationMessage.data.relatedSubject?.value, - status: DeliveryStatus.none, - ), - mode: InsertMode.insertOrReplace, - ); - } - - Future _insertConversationReactionsTableData(EventMessage eventMessage) { - final reactionEntity = PrivateMessageReactionEntity.fromEventMessage(eventMessage); - - return _db.into(_db.conversationReactionsTable).insert( - ConversationReactionsTableData( - reactionEventId: reactionEntity.id, - createdAt: reactionEntity.createdAt, - content: reactionEntity.data.content, - messageId: reactionEntity.data.eventId, - ), - mode: InsertMode.insertOrReplace, - ); - } - - Future insertEventMessage(EventMessage eventMessage) async { - await _db.into(_db.eventMessagesTable).insert( - EventMessageTableData.fromEventMessage(eventMessage), - mode: InsertMode.insertOrReplace, - ); - - if (eventMessage.kind == PrivateDirectMessageEntity.kind) { - await _updateConversationMessagesTable(eventMessage); - } else if (eventMessage.kind == PrivateMessageReactionEntity.kind) { - switch (eventMessage.content) { - case 'received': - await _updateConversationMessageAsReceived(eventMessage); - case 'read': - await _updateConversationMessagesAsRead(eventMessage); - default: - await _insertConversationReactionsTableData(eventMessage); - } - } - } - - Future _updateConversationMessagesTable( - EventMessage eventMessage, - ) async { - if (eventMessage.kind == PrivateDirectMessageEntity.kind) { - final conversationMessage = PrivateDirectMessageEntity.fromEventMessage(eventMessage); - final conversationIdByPubkeys = await _lookupConversationByPubkeys(conversationMessage); - - if (conversationIdByPubkeys != null) { - // Existing conversation (one-to-one or group) - await _insertConversationData( - eventMessage: eventMessage, - conversationId: conversationIdByPubkeys, - ); - } else { - // Existing group conversation (change of participants) - final conversationIdBySubject = await _lookupConversationBySubject(conversationMessage); - - if (conversationIdBySubject != null) { - await _insertConversationData( - eventMessage: eventMessage, - conversationId: conversationIdBySubject, - ); - } else if (eventMessage.content.isEmpty) { - // New conversation - final uuid = const Uuid().v1(); - await _insertConversationData( - eventMessage: eventMessage, - conversationId: uuid, - ); - } else { - // Invalid message (doesn't belong to any conversation) - throw ConversationIsNotFoundException(); - } - } - } - } - - // Check if there are conversations with the same pubkeys - Future _lookupConversationByPubkeys( - PrivateDirectMessageEntity conversationMessage, - ) async { - final conversationsWithSameParticipants = await (_db.select(_db.conversationMessagesTable) - ..where( - (table) => table.pubKeys.equals(conversationMessage.allPubkeysMask), - ) - ..limit(1)) - .get(); - - if (conversationsWithSameParticipants.isNotEmpty) { - return conversationsWithSameParticipants.first.conversationId; - } - - return null; - } - - // Check if there are conversations with the same subject but different pubkeys - // (this means that amount of participants in conversation was changed) - Future _lookupConversationBySubject( - PrivateDirectMessageEntity conversationMessage, - ) async { - final subject = conversationMessage.data.relatedSubject?.value; - - if (subject != null) { - final conversationWithChangedParticipants = await (_db.select(_db.conversationMessagesTable) - ..where((table) => table.subject.equals(subject)) - ..limit(1)) - .get(); - - if (conversationWithChangedParticipants.isNotEmpty) { - return conversationWithChangedParticipants.first.conversationId; - } - } - return null; - } - - // Call when "OK" is received from relay to mark message as sent (one tick) - Future markConversationMessageAsSent(String id) async { - final conversationMessagesTableData = await (_db.select(_db.conversationMessagesTable) - ..where((table) => table.eventMessageId.equals(id))) - .getSingle(); - - final sentConversationMessagesTableData = - conversationMessagesTableData.copyWith(status: DeliveryStatus.isSent); - - await _db.update(_db.conversationMessagesTable).replace(sentConversationMessagesTableData); - } - - // Call when kind 7 is received from relay with "received" content - Future _updateConversationMessageAsReceived( - EventMessage eventMessage, - ) async { - final reactionEntity = PrivateMessageReactionEntity.fromEventMessage(eventMessage); - - final conversationMessagesTableData = await (_db.select(_db.conversationMessagesTable) - ..where( - (table) => table.eventMessageId.equals(reactionEntity.data.eventId), - )) - .getSingle(); - - final receivedConversationMessagesTableData = - conversationMessagesTableData.copyWith(status: DeliveryStatus.isReceived); - - await _db.update(_db.conversationMessagesTable).replace(receivedConversationMessagesTableData); - } - - // Call when kind 7 is received from relay with "read" content - Future _updateConversationMessagesAsRead( - EventMessage eventMessage, - ) async { - final reactionEntity = PrivateMessageReactionEntity.fromEventMessage(eventMessage); - - final latestConversationMessageTableData = await (_db.select(_db.conversationMessagesTable) - ..where( - (table) => table.eventMessageId.equals(reactionEntity.data.eventId), - )) - .getSingle(); - - final allPreviousReceivedMessages = await (_db.select(_db.conversationMessagesTable) - ..where( - (table) => - table.conversationId.equals(latestConversationMessageTableData.conversationId), - ) - ..where( - (table) => table.status.equals(DeliveryStatus.isReceived.index), - ) - ..where( - (table) => table.createdAt.isSmallerOrEqualValue( - latestConversationMessageTableData.createdAt, - ), - )) - .get(); - - await _db.batch( - (b) { - b.replaceAll( - _db.conversationMessagesTable, - allPreviousReceivedMessages - .map( - (previousMessage) => previousMessage.copyWith(status: DeliveryStatus.isRead), - ) - .toList(), - ); - }, - ); - } - - final _allConversationsLatestMessageQuery = - 'SELECT * FROM (SELECT * FROM conversation_messages_table ORDER BY created_at DESC) AS sub GROUP BY conversation_id'; - - Future> getAllConversations() async { - // Select last message of each conversation - final uniqueConversationRows = await _db.customSelect( - _allConversationsLatestMessageQuery, - readsFrom: {_db.conversationMessagesTable}, - ).get(); - - final lastConversationEventMessages = - await _selectLastMessageOfEachConversation(uniqueConversationRows); - - return lastConversationEventMessages; - } - - Stream> watchConversations() { - return _db - .customSelect( - _allConversationsLatestMessageQuery, - readsFrom: {_db.conversationMessagesTable}, - ) - .watch() - .asyncMap((uniqueConversationRows) async { - final lastConversationEventMessages = await _selectLastMessageOfEachConversation( - uniqueConversationRows, - ); - - return lastConversationEventMessages; - }); - } - - Future> _selectLastMessageOfEachConversation( - List uniqueConversationRows, - ) async { - final lastConversationMessagesIds = - uniqueConversationRows.map((row) => row.data['event_message_id'] as String).toList(); - - final lastConversationEventMessages = (await (_db.select(_db.eventMessagesTable) - ..where((table) => table.id.isIn(lastConversationMessagesIds))) - .get()) - .map((e) => e.toEventMessage()) - .toList(); - - return lastConversationEventMessages; - } - - Future> getMessageReactions( - String messageId, - ) async { - final reactionsEventMessagesIds = (await (_db.select(_db.conversationReactionsTable) - ..where( - (table) => table.messageId.equals(messageId), - )) - .get()) - .map((reactionsTableData) => reactionsTableData.reactionEventId) - .toList(); - - final reactionsEventMessages = await (_db.select(_db.eventMessagesTable) - ..where((table) => table.id.isIn(reactionsEventMessagesIds))) - .get(); - - final reactions = reactionsEventMessages - .map( - (reactionEventMessageData) => PrivateMessageReactionEntity.fromEventMessage( - reactionEventMessageData.toEventMessage(), - ), - ) - .toList(); - - return reactions; - } - - //get last createdAt date from conversation_messages_table - Future getLastConversationMessageCreatedAt() async { - final lastConversationMessage = await (_db.select(_db.conversationMessagesTable) - ..orderBy([(table) => OrderingTerm.desc(table.createdAt)]) - ..limit(1)) - .getSingleOrNull(); - - return lastConversationMessage?.createdAt; - } -} - // DO NOT create or use database directly, use proxy notifier // [IONDatabaseNotifier] methods instead @DriftDatabase( @@ -317,6 +20,7 @@ class ConversationsDBService { EventMessagesTable, ConversationMessagesTable, ConversationReactionsTable, + DeletedConversationTable, ], ) class IONDatabase extends _$IONDatabase { @@ -377,6 +81,15 @@ class ConversationReactionsTable extends Table { Set> get primaryKey => {reactionEventId}; } +// Table for deleted conversations and time of deletion +class DeletedConversationTable extends Table { + TextColumn get conversationId => text()(); + DateTimeColumn get deletedAt => dateTime()(); + + @override + Set> get primaryKey => {conversationId}; +} + // As it is the only custom model needed better to keep it in the same file class EventMessageTableData implements Insertable { EventMessageTableData({ diff --git a/test/services/database/database_test.dart b/test/services/database/database_test.dart index 3c3f6d0db..cfece987e 100644 --- a/test/services/database/database_test.dart +++ b/test/services/database/database_test.dart @@ -5,6 +5,7 @@ import 'package:drift/native.dart'; import 'package:flutter_test/flutter_test.dart'; import 'package:ion/app/features/chat/model/entities/private_direct_message_data.c.dart'; import 'package:ion/app/features/feed/data/models/entities/reaction_data.c.dart'; +import 'package:ion/app/services/database/conversation_db_service.c.dart'; import 'package:ion/app/services/database/ion_database.c.dart'; import 'package:nostr_dart/nostr_dart.dart';