Merge pull request oxen-io#2532 from Bilb/fix-deleted-messages-all-at-once

To merge once theming is done: handle deleted messages & deleted reacts all at once

Author: Bilb

ts/components/conversation/message/message-item/GenericReadableMessage.tsx

@@ -8,7 +8,7 @@ import _ from 'lodash';
 import { Data } from '../../../../data/data';
 import { MessageRenderingProps } from '../../../../models/messageType';
 import { getConversationController } from '../../../../session/conversations';
-import { messageExpired } from '../../../../state/ducks/conversations';
+import { messagesExpired } from '../../../../state/ducks/conversations';
 import {
   getGenericReadableMessageSelectorProps,
   getIsMessageSelected,
@@ -68,10 +68,12 @@ function useIsExpired(props: ExpiringProps) {
       await Data.removeMessage(messageId);
       if (convoId) {
         dispatch(
-          messageExpired({
-            conversationKey: convoId,
-            messageId,
-          })
+          messagesExpired([
+            {
+              conversationKey: convoId,
+              messageId,
+            },
+          ])
         );
         const convo = getConversationController().get(convoId);
         convo?.updateLastMessage();

ts/data/data.ts

@@ -139,7 +139,7 @@ export const Data = {
   saveMessage,
   saveMessages,
   removeMessage,
-  _removeMessages,
+  removeMessagesByIds,
   getMessageIdsFromServerIds,
   getMessageById,
   getMessageBySenderAndSentAt,
@@ -396,9 +396,13 @@ async function removeMessage(id: string): Promise<void> {
   }
 }
 
-// Note: this method will not clean up external files, just delete from SQL
-async function _removeMessages(ids: Array<string>): Promise<void> {
-  await channels.removeMessage(ids);
+/**
+ * Note: this method will not clean up external files, just delete from SQL.
+ * Files are cleaned up on app start if they are not linked to any messages
+ *
+ */
+async function removeMessagesByIds(ids: Array<string>): Promise<void> {
+  await channels.removeMessagesByIds(ids);
 }
 
 async function getMessageIdsFromServerIds(
@@ -644,7 +648,7 @@ async function removeAllMessagesInConversation(conversationId: string): Promise<
     await Promise.all(messages.map(message => message.cleanup()));
 
     // eslint-disable-next-line no-await-in-loop
-    await channels.removeMessage(ids);
+    await channels.removeMessagesByIds(ids);
   } while (messages.length > 0);
 }
 

ts/data/dataInit.ts

@@ -38,7 +38,7 @@ const channelsToMake = new Set([
   'saveSeenMessageHashes',
   'saveMessages',
   'removeMessage',
-  '_removeMessages',
+  'removeMessagesByIds',
   'getUnreadByConversation',
   'markAllAsReadByConversationNoExpiration',
   'getUnreadCountByConversation',

ts/mains/main_renderer.tsx

@@ -217,18 +217,23 @@ async function start() {
   );
 
   window.log.info(`Cleanup: Found ${messagesForCleanup.length} messages for cleanup`);
+
+  const idsToCleanUp: Array<string> = [];
   await Promise.all(
-    messagesForCleanup.map(async (message: MessageModel) => {
+    messagesForCleanup.map((message: MessageModel) => {
       const sentAt = message.get('sent_at');
 
       if (message.hasErrors()) {
         return;
       }
 
       window.log.info(`Cleanup: Deleting unsent message ${sentAt}`);
-      await Data.removeMessage(message.id);
+      idsToCleanUp.push(message.id);
     })
   );
+  if (idsToCleanUp.length) {
+    await Data.removeMessagesByIds(idsToCleanUp);
+  }
   window.log.info('Cleanup: complete');
 
   window.log.info('listening for registration events');

ts/models/conversation.ts

@@ -249,12 +249,6 @@ export class ConversationModel extends Backbone.Model<ConversationAttributes> {
     await deleteExternalFilesOfConversation(this.attributes);
   }
 
-  public async onExpired(_message: MessageModel) {
-    await this.updateLastMessage();
-
-    // removeMessage();
-  }
-
   public getGroupAdmins(): Array<string> {
     const groupAdmins = this.get('groupAdmins');
 
@@ -478,26 +472,6 @@ export class ConversationModel extends Backbone.Model<ConversationAttributes> {
     return true;
   }
 
-  public async onReadMessage(message: MessageModel, readAt: number) {
-    // We mark as read everything older than this message - to clean up old stuff
-    //   still marked unread in the database. If the user generally doesn't read in
-    //   the desktop app, so the desktop app only gets read syncs, we can very
-    //   easily end up with messages never marked as read (our previous early read
-    //   sync handling, read syncs never sent because app was offline)
-
-    // We queue it because we often get a whole lot of read syncs at once, and
-    //   their markRead calls could very easily overlap given the async pull from DB.
-
-    // Lastly, we don't send read syncs for any message marked read due to a read
-    //   sync. That's a notification explosion we don't need.
-    return this.queueJob(() =>
-      this.markReadBouncy(message.get('received_at') as any, {
-        sendReadReceipts: false,
-        readAt,
-      })
-    );
-  }
-
   public async getUnreadCount() {
     const unreadCount = await Data.getUnreadCountByConversation(this.id);
 
@@ -1693,15 +1667,17 @@ export class ConversationModel extends Backbone.Model<ConversationAttributes> {
     return this.get('type') === ConversationTypeEnum.GROUP;
   }
 
-  public async removeMessage(messageId: any) {
+  public async removeMessage(messageId: string) {
     await Data.removeMessage(messageId);
     this.updateLastMessage();
 
     window.inboxStore?.dispatch(
-      conversationActions.messageDeleted({
-        conversationKey: this.id,
-        messageId,
-      })
+      conversationActions.messagesDeleted([
+        {
+          conversationKey: this.id,
+          messageId,
+        },
+      ])
     );
   }
 

ts/models/message.ts

@@ -125,7 +125,6 @@ export class MessageModel extends Backbone.Model<MessageAttributes> {
       throw new Error('A message always needs to have an conversationId.');
     }
 
-    // this.on('expired', this.onExpired);
     if (!attributes.skipTimerInit) {
       void this.setToExpire();
     }

ts/node/sql.ts

@@ -945,21 +945,44 @@ function saveMessages(arrayOfMessages: Array<any>) {
 }
 
 function removeMessage(id: string, instance?: BetterSqlite3.Database) {
-  if (!Array.isArray(id)) {
-    assertGlobalInstanceOrInstance(instance)
-      .prepare(`DELETE FROM ${MESSAGES_TABLE} WHERE id = $id;`)
-      .run({ id });
+  if (!isString(id)) {
+    throw new Error('removeMessage: only takes single message to delete!');
+
     return;
   }
 
-  if (!id.length) {
-    throw new Error('removeMessages: No ids to delete!');
+  assertGlobalInstanceOrInstance(instance)
+    .prepare(`DELETE FROM ${MESSAGES_TABLE} WHERE id = $id;`)
+    .run({ id });
+}
+
+function removeMessagesByIds(ids: Array<string>, instance?: BetterSqlite3.Database) {
+  if (!Array.isArray(ids)) {
+    throw new Error('removeMessagesByIds only allowed an array of strings');
+  }
+
+  if (!ids.length) {
+    throw new Error('removeMessagesByIds: No ids to delete!');
   }
 
   // Our node interface doesn't seem to allow you to replace one single ? with an array
   assertGlobalInstanceOrInstance(instance)
-    .prepare(`DELETE FROM ${MESSAGES_TABLE} WHERE id IN ( ${id.map(() => '?').join(', ')} );`)
-    .run(id);
+    .prepare(`DELETE FROM ${MESSAGES_TABLE} WHERE id IN ( ${ids.map(() => '?').join(', ')} );`)
+    .run(ids);
+}
+
+function removeAllMessagesInConversation(
+  conversationId: string,
+  instance?: BetterSqlite3.Database
+) {
+  if (!conversationId) {
+    return;
+  }
+
+  // Our node interface doesn't seem to allow you to replace one single ? with an array
+  assertGlobalInstanceOrInstance(instance)
+    .prepare(`DELETE FROM ${MESSAGES_TABLE} WHERE conversationId = $conversationId`)
+    .run({ conversationId });
 }
 
 function getMessageIdsFromServerIds(serverIds: Array<string | number>, conversationId: string) {
@@ -2440,6 +2463,8 @@ export const sqlNode = {
   updateLastHash,
   saveMessages,
   removeMessage,
+  removeMessagesByIds,
+  removeAllMessagesInConversation,
   getUnreadByConversation,
   markAllAsReadByConversationNoExpiration,
   getUnreadCountByConversation,

ts/session/apis/open_group_api/sogsv3/sogsApiV3.ts

@@ -1,4 +1,4 @@
-import _, { compact, isArray, isNumber, isObject, pick } from 'lodash';
+import _, { compact, isArray, isEmpty, isNumber, isObject, pick } from 'lodash';
 import { OpenGroupData } from '../../../../data/opengroups';
 import { handleOpenGroupV4Message } from '../../../../receiver/opengroup';
 import { OpenGroupRequestCommonType } from '../opengroupV2/ApiUtil';
@@ -35,6 +35,8 @@ import { ConversationTypeEnum } from '../../../../models/conversationAttributes'
 import { createSwarmMessageSentFromUs } from '../../../../models/messageFactory';
 import { Data } from '../../../../data/data';
 import { processMessagesUsingCache } from './sogsV3MutationCache';
+import { destroyMessagesAndUpdateRedux } from '../../../../util/expiringMessages';
+import { sogsRollingDeletions } from './sogsRollingDeletions';
 
 /**
  * Get the convo matching those criteria and make sure it is an opengroup convo, or return null.
@@ -159,34 +161,38 @@ const handleSogsV3DeletedMessages = async (
   serverUrl: string,
   roomId: string
 ) => {
-  const deletions = messages.filter(m => Boolean(m.deleted));
-  const exceptDeletion = messages.filter(m => !m.deleted);
-  if (!deletions.length) {
-    return messages;
+  const messagesDeleted = messages.filter(m => Boolean(m.deleted));
+  const messagesWithoutDeleted = messages.filter(m => !m.deleted);
+  if (!messagesDeleted.length) {
+    return messagesWithoutDeleted;
   }
-  const allIdsRemoved = deletions.map(m => m.id);
+
+  const allIdsRemoved = messagesDeleted.map(m => m.id);
+
   try {
     const convoId = getOpenGroupV2ConversationId(serverUrl, roomId);
     const convo = getConversationController().get(convoId);
     const messageIds = await Data.getMessageIdsFromServerIds(allIdsRemoved, convo.id);
 
-    // we shouldn't get too many messages to delete at a time, so no need to add a function to remove multiple messages for now
+    allIdsRemoved.forEach(removedId => {
+      sogsRollingDeletions.addMessageDeletedId(convoId, removedId);
+    });
 
-    await Promise.all(
-      (messageIds || []).map(async id => {
-        if (convo) {
-          await convo.removeMessage(id);
-        }
-        await Data.removeMessage(id);
-      })
-    );
+    if (messageIds && messageIds.length) {
+      await destroyMessagesAndUpdateRedux(
+        messageIds.map(messageId => ({
+          conversationKey: convoId,
+          messageId,
+        }))
+      );
+    }
   } catch (e) {
     window?.log?.warn('handleDeletions failed:', e);
   }
-  return exceptDeletion;
+  return messagesWithoutDeleted;
 };
 
-// tslint:disable-next-line: cyclomatic-complexity
+// tslint:disable-next-line: max-func-body-length cyclomatic-complexity
 const handleMessagesResponseV4 = async (
   messages: Array<OpenGroupMessageV4>,
   serverUrl: string,
@@ -284,6 +290,7 @@ const handleMessagesResponseV4 = async (
 
     const incomingMessageSeqNo = compact(messages.map(n => n.seqno));
     const maxNewMessageSeqNo = Math.max(...incomingMessageSeqNo);
+
     for (let index = 0; index < messagesWithResolvedBlindedIdsIfFound.length; index++) {
       const msgToHandle = messagesWithResolvedBlindedIdsIfFound[index];
       try {
@@ -309,13 +316,24 @@ const handleMessagesResponseV4 = async (
     await OpenGroupData.saveV2OpenGroupRoom(roomInfosRefreshed);
 
     const messagesWithReactions = messages.filter(m => m.reactions !== undefined);
+
     if (messagesWithReactions.length > 0) {
       const conversationId = getOpenGroupV2ConversationId(serverUrl, roomId);
       const groupConvo = getConversationController().get(conversationId);
       if (groupConvo && groupConvo.isOpenGroupV2()) {
-        for (const message of messagesWithReactions) {
+        for (const messageWithReaction of messagesWithReactions) {
+          if (isEmpty(messageWithReaction.reactions)) {
+            /*
+             * When a message is deleted from the server, we get the deleted event as a data: null on the message itself
+             * and an update on its reactions.
+             * But, because we just deleted that message, we can skip trying to update its reactions: it's not in the DB anymore.
+             */
+            if (sogsRollingDeletions.hasMessageDeletedId(conversationId, messageWithReaction.id)) {
+              continue;
+            }
+          }
           void groupConvo.queueJob(async () => {
-            await processMessagesUsingCache(serverUrl, roomId, message);
+            await processMessagesUsingCache(serverUrl, roomId, messageWithReaction);
           });
         }
       }
@@ -526,6 +544,7 @@ export const handleBatchPollResults = async (
           break;
         case 'pollInfo':
           await handlePollInfoResponse(subResponse.code, subResponse.body, serverUrl);
+
           break;
         case 'inbox':
           await handleInboxOutboxMessages(subResponse.body, serverUrl, false);

ts/session/apis/open_group_api/sogsv3/sogsRollingDeletions.ts

@@ -0,0 +1,48 @@
+import { RingBuffer } from '../../../utils/RingBuffer';
+
+const rollingDeletedMessageIds: Map<string, RingBuffer<number>> = new Map();
+
+const addMessageDeletedId = (conversationId: string, messageDeletedId: number) => {
+  if (!rollingDeletedMessageIds.has(conversationId)) {
+    rollingDeletedMessageIds.set(
+      conversationId,
+      new RingBuffer<number>(sogsRollingDeletions.getPerRoomCount())
+    );
+  }
+  const ringBuffer = rollingDeletedMessageIds.get(conversationId);
+  if (!ringBuffer) {
+    return;
+  }
+  ringBuffer.insert(messageDeletedId);
+};
+
+const hasMessageDeletedId = (conversationId: string, messageDeletedId: number) => {
+  if (!rollingDeletedMessageIds.has(conversationId)) {
+    return false;
+  }
+
+  const messageIdWasDeletedRecently = rollingDeletedMessageIds
+    ?.get(conversationId)
+    ?.has(messageDeletedId);
+
+  return messageIdWasDeletedRecently;
+};
+
+/**
+ * emptyMessageDeleteIds should only be used for testing purposes.
+ */
+const emptyMessageDeleteIds = () => {
+  rollingDeletedMessageIds.clear();
+};
+
+export const sogsRollingDeletions = {
+  addMessageDeletedId,
+  hasMessageDeletedId,
+  emptyMessageDeleteIds,
+  getPerRoomCount,
+};
+
+// keep 2000 deleted message ids in memory
+function getPerRoomCount() {
+  return 2000;
+}