fix(api): Remove subscribers along with their related entities

apps/api/src/app/subscribers/e2e/remove-subscriber.e2e.ts

@@ -45,18 +45,8 @@ describe('Delete Subscriber - /subscribers/:subscriberId (DELETE)', function ()
       },
     });
 
-    const isDeleted = !(await subscriberRepository.findBySubscriberId(session.environment._id, '123'));
-
-    expect(isDeleted).to.equal(true);
-
-    const deletedSubscriber = (
-      await subscriberRepository.findDeleted({
-        _environmentId: session.environment._id,
-        subscriberId: '123',
-      })
-    )?.[0];
-
-    expect(deletedSubscriber.deleted).to.equal(true);
+    const subscriber = await subscriberRepository.findBySubscriberId(session.environment._id, '123');
+    expect(subscriber).to.be.null;
   });
 
   it('should dispose subscriber relations to topic once he was removed', async () => {

apps/api/src/app/subscribers/usecases/remove-subscriber/remove-subscriber.spec.ts

@@ -1,15 +1,13 @@
-import { Test } from '@nestjs/testing';
-import { SubscribersService, UserSession } from '@novu/testing';
-import { NotFoundException } from '@nestjs/common';
 import { expect } from 'chai';
-
+import { NotFoundException } from '@nestjs/common';
+import { SubscribersService, UserSession } from '@novu/testing';
+import { Test } from '@nestjs/testing';
 import { RemoveSubscriber } from './remove-subscriber.usecase';
 import { RemoveSubscriberCommand } from './remove-subscriber.command';
-
 import { SharedModule } from '../../../shared/shared.module';
 import { SubscribersModule } from '../../subscribers.module';
 
-describe('Remove Subscriber', function () {
+describe.only('Remove Subscriber', function () {
   let useCase: RemoveSubscriber;
   let session: UserSession;
 
@@ -41,8 +39,6 @@ describe('Remove Subscriber', function () {
   });
 
   it('should throw a not found exception if subscriber to remove does not exist', async () => {
-    const subscriberService = new SubscribersService(session.organization._id, session.environment._id);
-
     try {
       await useCase.execute(
         RemoveSubscriberCommand.create({
@@ -51,10 +47,9 @@ describe('Remove Subscriber', function () {
           organizationId: session.organization._id,
         })
       );
-      throw new Error('Should not reach here');
+      expect(true, 'Should never reach this statement').to.be.false;
     } catch (e) {
       expect(e).to.be.instanceOf(NotFoundException);
-      expect(e.message).to.eql("Subscriber 'invalid-subscriber-id' was not found");
     }
   });
 });

apps/api/src/app/subscribers/usecases/remove-subscriber/remove-subscriber.usecase.ts

@@ -1,5 +1,10 @@
-import { Injectable } from '@nestjs/common';
-import { SubscriberRepository, DalException, TopicSubscribersRepository } from '@novu/dal';
+import { Injectable, NotFoundException } from '@nestjs/common';
+import {
+  SubscriberRepository,
+  TopicSubscribersRepository,
+  SubscriberPreferenceRepository,
+  PreferencesRepository,
+} from '@novu/dal';
 import {
   buildSubscriberKey,
   buildFeedKey,
@@ -8,64 +13,77 @@ import {
 } from '@novu/application-generic';
 
 import { RemoveSubscriberCommand } from './remove-subscriber.command';
-import { GetSubscriber } from '../get-subscriber';
-import { ApiException } from '../../../shared/exceptions/api.exception';
 
 @Injectable()
 export class RemoveSubscriber {
   constructor(
     private invalidateCache: InvalidateCacheService,
     private subscriberRepository: SubscriberRepository,
-    private getSubscriber: GetSubscriber,
-    private topicSubscribersRepository: TopicSubscribersRepository
+    private topicSubscribersRepository: TopicSubscribersRepository,
+    private subscriberPreferenceRepository: SubscriberPreferenceRepository,
+    private preferenceRepository: PreferencesRepository
   ) {}
 
-  async execute(command: RemoveSubscriberCommand) {
-    try {
-      const { environmentId: _environmentId, organizationId, subscriberId } = command;
-      const subscriber = await this.getSubscriber.execute({
-        environmentId: _environmentId,
-        organizationId,
-        subscriberId,
-      });
-
-      await Promise.all([
-        this.invalidateCache.invalidateByKey({
-          key: buildSubscriberKey({
-            subscriberId: command.subscriberId,
-            _environmentId: command.environmentId,
-          }),
+  async execute({ environmentId: _environmentId, subscriberId }: RemoveSubscriberCommand) {
+    await Promise.all([
+      this.invalidateCache.invalidateByKey({
+        key: buildSubscriberKey({
+          subscriberId,
+          _environmentId,
         }),
-        this.invalidateCache.invalidateQuery({
-          key: buildFeedKey().invalidate({
-            subscriberId,
-            _environmentId: command.environmentId,
-          }),
+      }),
+      this.invalidateCache.invalidateQuery({
+        key: buildFeedKey().invalidate({
+          subscriberId,
+          _environmentId,
         }),
-        this.invalidateCache.invalidateQuery({
-          key: buildMessageCountKey().invalidate({
-            subscriberId,
-            _environmentId: command.environmentId,
-          }),
+      }),
+      this.invalidateCache.invalidateQuery({
+        key: buildMessageCountKey().invalidate({
+          subscriberId,
+          _environmentId,
         }),
-      ]);
+      }),
+    ]);
+
+    let deletedSubscriberCount = 0;
 
-      await this.subscriberRepository.delete({
-        _environmentId: subscriber._environmentId,
-        _organizationId: subscriber._organizationId,
-        subscriberId: subscriber.subscriberId,
+    await this.subscriberRepository.withTransaction(async () => {
+      /*
+       * Note about parallelism in transactions
+       *
+       * Running operations in parallel is not supported during a transaction.
+       * The use of Promise.all, Promise.allSettled, Promise.race, etc. to parallelize operations
+       * inside a transaction is undefined behaviour and should be avoided.
+       *
+       * Refer to https://mongoosejs.com/docs/transactions.html#note-about-parallelism-in-transactions
+       */
+      const { deletedCount } = await this.subscriberRepository.delete({
+        subscriberId,
+        _environmentId,
       });
 
+      deletedSubscriberCount = deletedCount;
+      if (deletedSubscriberCount === 0) {
+        return;
+      }
+
       await this.topicSubscribersRepository.delete({
-        _environmentId: subscriber._environmentId,
-        _organizationId: subscriber._organizationId,
-        externalSubscriberId: subscriber.subscriberId,
+        _environmentId,
+        externalSubscriberId: subscriberId,
       });
-    } catch (e) {
-      if (e instanceof DalException) {
-        throw new ApiException(e.message);
-      }
-      throw e;
+      await this.subscriberPreferenceRepository.delete({
+        _environmentId,
+        _subscriberId: subscriberId,
+      });
+      await this.preferenceRepository.delete({
+        _environmentId,
+        _subscriberId: subscriberId,
+      });
+    });
+
+    if (deletedSubscriberCount === 0) {
+      throw new NotFoundException(`Subscriber '${subscriberId}' was not found`);
     }
 
     return {

libs/dal/src/repositories/preferences/preferences.repository.ts

@@ -28,13 +28,6 @@ export class PreferencesRepository extends BaseRepository<PreferencesDBModel, Pr
     return this.mapEntity(item);
   }
 
-  async delete(query: PreferencesQuery) {
-    const item = await this.findOne({ _id: query._id, _environmentId: query._environmentId });
-    if (!item) throw new DalException(`Could not find preferences with id ${query._id}`);
-
-    return await this.preferences.delete({ _id: item._id, _environmentId: item._environmentId });
-  }
-
   async findDeleted(query: PreferencesQuery): Promise<PreferencesEntity> {
     const res: PreferencesEntity = await this.preferences.findDeleted(query);