refactor(dal): remove soft delete in message collection (#8073)

djabarovgeorge · web-flow · commit e6e4faa82962 · 2025-04-15T16:12:38.000+03:00
diff --git a/apps/api/src/app/messages/usecases/remove-messages-by-transactionId/remove-messages-by-transactionId.usecase.ts b/apps/api/src/app/messages/usecases/remove-messages-by-transactionId/remove-messages-by-transactionId.usecase.ts
@@ -1,6 +1,8 @@
 import { Injectable, NotFoundException } from '@nestjs/common';
+
 import { buildFeedKey, buildMessageCountKey, InvalidateCacheService } from '@novu/application-generic';
-import { MessageEntity, MessageRepository } from '@novu/dal';
+import { EnforceEnvId, MessageEntity, MessageRepository } from '@novu/dal';
+
 import { RemoveMessagesByTransactionIdCommand } from './remove-messages-by-transactionId.command';
 
 @Injectable()
@@ -41,7 +43,7 @@ export class RemoveMessagesByTransactionId {
       }
     }
 
-    const deleteQuery: Partial<MessageEntity> = {
+    const deleteQuery: Partial<MessageEntity> & EnforceEnvId = {
       transactionId: command.transactionId,
       _environmentId: command.environmentId,
       _organizationId: command.organizationId,
@@ -51,6 +53,6 @@ export class RemoveMessagesByTransactionId {
       deleteQuery.channel = command.channel;
     }
 
-    await this.messageRepository.deleteMany(deleteQuery);
+    await this.messageRepository.delete(deleteQuery);
   }
 }
diff --git a/apps/api/src/app/widgets/dtos/feeds-response.dto.ts b/apps/api/src/app/widgets/dtos/feeds-response.dto.ts
@@ -177,13 +177,6 @@ export class NotificationFeedItemDto implements INotificationDto {
   })
   seen: boolean;
 
-  @ApiProperty({
-    description: 'Indicates whether the notification has been deleted.',
-    example: false,
-    type: Boolean,
-  })
-  deleted: boolean;
-
   @ApiPropertyOptional({
     description: 'Device tokens for push notifications, if applicable.',
     type: [String],
diff --git a/apps/api/src/app/widgets/usecases/remove-message/remove-message.usecase.ts b/apps/api/src/app/widgets/usecases/remove-message/remove-message.usecase.ts
@@ -1,5 +1,5 @@
+import { DalException, MessageRepository, SubscriberRepository, SubscriberEntity } from '@novu/dal';
 import { Injectable, NotFoundException, BadRequestException } from '@nestjs/common';
-import { MessageEntity, DalException, MessageRepository, SubscriberRepository, SubscriberEntity } from '@novu/dal';
 import {
   WebSocketsQueueService,
   AnalyticsService,
@@ -22,7 +22,7 @@ export class RemoveMessage {
     private subscriberRepository: SubscriberRepository
   ) {}
 
-  async execute(command: RemoveMessageCommand): Promise<MessageEntity> {
+  async execute(command: RemoveMessageCommand): Promise<void> {
     await this.invalidateCache.invalidateQuery({
       key: buildFeedKey().invalidate({
         subscriberId: command.subscriberId,
@@ -40,41 +40,30 @@ export class RemoveMessage {
     const subscriber = await this.subscriberRepository.findBySubscriberId(command.environmentId, command.subscriberId);
     if (!subscriber) throw new NotFoundException(`Subscriber ${command.subscriberId} not found`);
 
-    let deletedMessage;
     try {
-      await this.messageRepository.delete({
+      const deletedMessage = await this.messageRepository.delete({
         _environmentId: command.environmentId,
         _organizationId: command.organizationId,
         _id: command.messageId,
         _subscriberId: command.subscriberId,
       });
-      const item = await this.messageRepository.findDeleted({
-        _environmentId: command.environmentId,
-        _organizationId: command.organizationId,
-        _id: command.messageId,
-      });
 
-      // eslint-disable-next-line prefer-destructuring
-      deletedMessage = item[0];
-
-      if (!deletedMessage.read) {
-        await this.updateServices(command, subscriber, deletedMessage, MarkEnum.READ);
-      }
-      if (!deletedMessage.seen) {
-        await this.updateServices(command, subscriber, deletedMessage, MarkEnum.SEEN);
+      if (deletedMessage.deletedCount) {
+        await Promise.all([
+          this.updateServices(command, subscriber, command.messageId, MarkEnum.READ),
+          this.updateServices(command, subscriber, command.messageId, MarkEnum.SEEN),
+        ]);
       }
     } catch (e) {
       if (e instanceof DalException) {
         throw new BadRequestException(e.message);
       }
       throw e;
     }
-
-    return deletedMessage;
   }
 
   private async updateServices(command: RemoveMessageCommand, subscriber, message, marked: MarkEnum) {
-    this.updateSocketCount(subscriber, marked);
+    await this.updateSocketCount(subscriber, marked);
 
     this.analyticsService.track(`Removed Message - [Notification Center]`, command.organizationId, {
       _subscriber: message._subscriberId,
@@ -83,10 +72,10 @@ export class RemoveMessage {
     });
   }
 
-  private updateSocketCount(subscriber: SubscriberEntity, mark: MarkEnum) {
+  private async updateSocketCount(subscriber: SubscriberEntity, mark: MarkEnum) {
     const eventMessage = mark === MarkEnum.READ ? WebSocketEventEnum.UNREAD : WebSocketEventEnum.UNSEEN;
 
-    this.webSocketsQueueService.add({
+    await this.webSocketsQueueService.add({
       name: 'sendMessage',
       data: {
         event: eventMessage,
diff --git a/apps/api/src/app/widgets/usecases/remove-messages-bulk/remove-messages-bulk.usecase.ts b/apps/api/src/app/widgets/usecases/remove-messages-bulk/remove-messages-bulk.usecase.ts
@@ -27,30 +27,32 @@ export class RemoveMessagesBulk {
     if (!subscriber) throw new NotFoundException(`Subscriber ${command.subscriberId} not found`);
 
     try {
-      await this.messageRepository.deleteMany({
+      const deletedMessages = await this.messageRepository.delete({
         _environmentId: command.environmentId,
         _organizationId: command.organizationId,
         _subscriberId: subscriber._id,
         channel: ChannelTypeEnum.IN_APP,
         _id: { $in: command.messageIds },
       });
 
-      await this.updateServices(subscriber, MarkEnum.SEEN);
-      await this.updateServices(subscriber, MarkEnum.READ);
-
-      await this.invalidateCache.invalidateQuery({
-        key: buildFeedKey().invalidate({
-          subscriberId: command.subscriberId,
-          _environmentId: command.environmentId,
-        }),
-      });
-
-      await this.invalidateCache.invalidateQuery({
-        key: buildMessageCountKey().invalidate({
-          subscriberId: command.subscriberId,
-          _environmentId: command.environmentId,
-        }),
-      });
+      if (deletedMessages.deletedCount > 0) {
+        await Promise.all([
+          this.updateServices(subscriber, MarkEnum.SEEN),
+          this.updateServices(subscriber, MarkEnum.READ),
+          this.invalidateCache.invalidateQuery({
+            key: buildFeedKey().invalidate({
+              subscriberId: command.subscriberId,
+              _environmentId: command.environmentId,
+            }),
+          }),
+          this.invalidateCache.invalidateQuery({
+            key: buildMessageCountKey().invalidate({
+              subscriberId: command.subscriberId,
+              _environmentId: command.environmentId,
+            }),
+          }),
+        ]);
+      }
     } catch (e) {
       if (e instanceof DalException) {
         throw new BadRequestException(e.message);
@@ -59,10 +61,10 @@ export class RemoveMessagesBulk {
     }
   }
 
-  private async updateServices(subscriber, marked: string) {
+  private async updateServices(subscriber, marked: string): Promise<void> {
     const eventMessage = marked === MarkEnum.READ ? WebSocketEventEnum.UNREAD : WebSocketEventEnum.UNSEEN;
 
-    this.webSocketsQueueService.add({
+    await this.webSocketsQueueService.add({
       name: 'sendMessage',
       data: {
         event: eventMessage,
diff --git a/apps/api/src/app/widgets/usecases/remove-messages/remove-all-messages.usecase.ts b/apps/api/src/app/widgets/usecases/remove-messages/remove-all-messages.usecase.ts
@@ -6,6 +6,7 @@ import {
   SubscriberRepository,
   SubscriberEntity,
   FeedRepository,
+  EnforceEnvId,
 } from '@novu/dal';
 import { ChannelTypeEnum, WebSocketEventEnum } from '@novu/shared';
 import {
@@ -43,7 +44,7 @@ export class RemoveAllMessages {
         }
       }
 
-      const deleteMessageQuery: Partial<MessageEntity> = {
+      const deleteMessageQuery: Partial<MessageEntity> & EnforceEnvId = {
         _environmentId: command.environmentId,
         _organizationId: command.organizationId,
         _subscriberId: subscriber._id,
@@ -53,32 +54,33 @@ export class RemoveAllMessages {
       if (feed) {
         deleteMessageQuery._feedId = feed._id;
       }
+      const deletedMessages = await this.messageRepository.delete(deleteMessageQuery);
 
-      await this.messageRepository.deleteMany(deleteMessageQuery);
-
-      await this.updateServices(command, subscriber, MarkEnum.SEEN);
-      await this.updateServices(command, subscriber, MarkEnum.READ);
+      if (deletedMessages.deletedCount > 0) {
+        await Promise.all([
+          this.updateServices(command, subscriber, MarkEnum.SEEN),
+          this.updateServices(command, subscriber, MarkEnum.READ),
+          this.invalidateCache.invalidateQuery({
+            key: buildFeedKey().invalidate({
+              subscriberId: command.subscriberId,
+              _environmentId: command.environmentId,
+            }),
+          }),
+          this.invalidateCache.invalidateQuery({
+            key: buildMessageCountKey().invalidate({
+              subscriberId: command.subscriberId,
+              _environmentId: command.environmentId,
+            }),
+          }),
+        ]);
+      }
 
       this.analyticsService.track(`Removed All Feed Messages - [Notification Center]`, command.organizationId, {
         _subscriber: subscriber._id,
         _organization: command.organizationId,
         _environment: command.environmentId,
         _feedId: command.feedId,
       });
-
-      await this.invalidateCache.invalidateQuery({
-        key: buildFeedKey().invalidate({
-          subscriberId: command.subscriberId,
-          _environmentId: command.environmentId,
-        }),
-      });
-
-      await this.invalidateCache.invalidateQuery({
-        key: buildMessageCountKey().invalidate({
-          subscriberId: command.subscriberId,
-          _environmentId: command.environmentId,
-        }),
-      });
     } catch (e) {
       if (e instanceof DalException) {
         throw new BadRequestException(e.message);
@@ -87,14 +89,14 @@ export class RemoveAllMessages {
     }
   }
 
-  private async updateServices(command: RemoveAllMessagesCommand, subscriber, marked: string) {
-    this.updateSocketCount(subscriber, marked);
+  private async updateServices(command: RemoveAllMessagesCommand, subscriber, marked: string): Promise<void> {
+    await this.updateSocketCount(subscriber, marked);
   }
 
-  private updateSocketCount(subscriber: SubscriberEntity, mark: string) {
+  private async updateSocketCount(subscriber: SubscriberEntity, mark: string): Promise<void> {
     const eventMessage = mark === MarkEnum.READ ? WebSocketEventEnum.UNREAD : WebSocketEventEnum.UNSEEN;
 
-    this.webSocketsQueueService.add({
+    await this.webSocketsQueueService.add({
       name: 'sendMessage',
       data: {
         event: eventMessage,
diff --git a/apps/api/src/app/widgets/widgets.controller.ts b/apps/api/src/app/widgets/widgets.controller.ts
@@ -282,7 +282,7 @@ export class WidgetsController {
   async removeMessage(
     @SubscriberSession() subscriberSession: SubscriberEntity,
     @Param('messageId') messageId: string
-  ): Promise<MessageEntity> {
+  ): Promise<void> {
     if (!messageId) throw new BadRequestException('messageId is required');
 
     const command = RemoveMessageCommand.create({
diff --git a/libs/dal/src/repositories/message/message.entity.ts b/libs/dal/src/repositories/message/message.entity.ts
@@ -53,6 +53,9 @@ export class MessageEntity {
 
   archived: boolean;
 
+  /**
+   * todo: remove deleted field after all the soft deletes are removed task nv-5688
+   */
   deleted: boolean;
 
   email?: string;
diff --git a/libs/dal/src/repositories/message/message.repository.ts b/libs/dal/src/repositories/message/message.repository.ts
@@ -1,5 +1,4 @@
-import { SoftDeleteModel } from 'mongoose-delete';
-import { FilterQuery, Types } from 'mongoose';
+import { FilterQuery, QueryWithHelpers, Types, UpdateQuery } from 'mongoose';
 import {
   ActorTypeEnum,
   ButtonTypeEnum,
@@ -27,11 +26,9 @@ const getFlatObject = (obj: object) => {
 };
 
 export class MessageRepository extends BaseRepository<MessageDBModel, MessageEntity, EnforceEnvId> {
-  private message: SoftDeleteModel;
   private feedRepository = new FeedRepository();
   constructor() {
     super(Message, MessageEntity);
-    this.message = Message;
   }
 
   private async getFilterQueryForMessage(
@@ -111,6 +108,14 @@ export class MessageRepository extends BaseRepository<MessageDBModel, MessageEnt
     return requestQuery;
   }
 
+  /**
+   * if aggregation is needed, make sure to filter with {deleted: { $ne: true }}.
+   * todo: aggregate method should be implemented after all the soft deletes are removed task nv-5688
+   */
+  async aggregate(query: any[], options: { readPreference?: 'secondaryPreferred' | 'primary' } = {}): Promise<any> {
+    throw new Error('Not implemented');
+  }
+
   async findBySubscriberChannel(
     environmentId: string,
     subscriberId: string,
@@ -656,28 +661,6 @@ export class MessageRepository extends BaseRepository<MessageDBModel, MessageEnt
     );
   }
 
-  async delete(query: MessageQuery) {
-    return await this.message.delete({ _id: query._id, _environmentId: query._environmentId });
-  }
-
-  async deleteMany(query: MessageQuery) {
-    try {
-      return await this.message.delete({ ...query, deleted: false });
-    } catch (e: unknown) {
-      if (e instanceof Error) {
-        throw new DalException(e.message);
-      } else {
-        throw new DalException('An unknown error occurred');
-      }
-    }
-  }
-
-  async findDeleted(query: MessageQuery): Promise<MessageEntity> {
-    const res: MessageEntity = await this.message.findDeleted(query);
-
-    return this.mapEntity(res);
-  }
-
   async findMessageById(query: { _id: string; _environmentId: string }): Promise<MessageEntity | null> {
     const res = await this.MongooseModel.findOne({ _id: query._id, _environmentId: query._environmentId })
       .populate('subscriber')
@@ -734,7 +717,7 @@ export class MessageRepository extends BaseRepository<MessageDBModel, MessageEnt
     if (query.transactionId) {
       filterQuery.transactionId = { $in: query.transactionId };
     }
-    const data = await this.MongooseModel.find(query, select, {
+    const data = await this.MongooseModel.find(filterQuery, select, {
       sort: options?.sort,
       limit: options?.limit,
       skip: options?.skip,
diff --git a/libs/dal/src/repositories/message/message.schema.ts b/libs/dal/src/repositories/message/message.schema.ts
diff --git a/packages/shared/src/dto/widget/notification.dto.ts b/packages/shared/src/dto/widget/notification.dto.ts

Original file line number	Diff line number	Diff line change
`@@ -1,6 +1,8 @@`
`1`	`1`	`import { Injectable, NotFoundException } from '@nestjs/common';`
	`2`	`+`
`2`	`3`	`import { buildFeedKey, buildMessageCountKey, InvalidateCacheService } from '@novu/application-generic';`
`3`		`-import { MessageEntity, MessageRepository } from '@novu/dal';`
	`4`	`+import { EnforceEnvId, MessageEntity, MessageRepository } from '@novu/dal';`
	`5`	`+`
`4`	`6`	`import { RemoveMessagesByTransactionIdCommand } from './remove-messages-by-transactionId.command';`
`5`	`7`
`6`	`8`	`@Injectable()`
`@@ -41,7 +43,7 @@ export class RemoveMessagesByTransactionId {`
`41`	`43`	`}`
`42`	`44`	`}`
`43`	`45`
`44`		`- const deleteQuery: Partial<MessageEntity> = {`
	`46`	`+ const deleteQuery: Partial<MessageEntity> & EnforceEnvId = {`
`45`	`47`	`transactionId: command.transactionId,`
`46`	`48`	`_environmentId: command.environmentId,`
`47`	`49`	`_organizationId: command.organizationId,`
`@@ -51,6 +53,6 @@ export class RemoveMessagesByTransactionId {`
`51`	`53`	`deleteQuery.channel = command.channel;`
`52`	`54`	`}`
`53`	`55`
`54`		`- await this.messageRepository.deleteMany(deleteQuery);`
	`56`	`+ await this.messageRepository.delete(deleteQuery);`
`55`	`57`	`}`
`56`	`58`	`}`