Merge branch 'novuhq:next' into configure-mongo-min-pool-size

.github/actions/run-api/action.yml

@@ -19,7 +19,6 @@ runs:
     - name: Start API
       shell: bash
       env:
-        IS_IN_MEMORY_CLUSTER_MODE_ENABLED: true
         LAUNCH_DARKLY_SDK_KEY: ${{ inputs.launch_darkly_sdk_key }}
       run: cd apps/api && pnpm start:test &
 

.github/actions/run-backend/action.yml

@@ -28,7 +28,6 @@ runs:
         TZ: "UTC"
         GITHUB_OAUTH_REDIRECT: "http://localhost:1336/v1/auth/github/callback"
         LAUNCH_DARKLY_SDK_KEY: ${{ inputs.launch_darkly_sdk_key }}
-        IN_MEMORY_CLUSTER_MODE_ENABLED: true
       run: cd apps/api && pnpm start:build &
 
     - name: Start Worker
@@ -38,7 +37,6 @@ runs:
         PORT: "1342"
         TZ: "UTC"
         LAUNCH_DARKLY_SDK_KEY: ${{ inputs.launch_darkly_sdk_key }}
-        IN_MEMORY_CLUSTER_MODE_ENABLED: true
       run: cd apps/worker && pnpm start:prod &
 
     - name: Wait on API and Worker

.github/actions/run-worker/action.yml

@@ -20,7 +20,6 @@ runs:
     - name: Start worker
       shell: bash
       env:
-        IN_MEMORY_CLUSTER_MODE_ENABLED: true
         LAUNCH_DARKLY_SDK_KEY: ${{ inputs.launch_darkly_sdk_key }}
       run: cd apps/worker && pnpm start:test &
 

.github/workflows/reusable-api-e2e.yml

@@ -64,8 +64,6 @@ jobs:
         run: CI='' pnpm build:api
 
       - name: Run E2E tests
-        env:
-          IN_MEMORY_CLUSTER_MODE_ENABLED: true
         run: |
           cd apps/api && pnpm test:e2e
 

.github/workflows/reusable-web-e2e.yml

@@ -44,6 +44,7 @@ jobs:
           else
             echo ::set-output has_token=false
           fi
+
       # checkout with submodules if token is provided
       - uses: actions/checkout@v3
         if: steps.setup.outputs.has_token == 'true'

.github/workflows/reusable-webhook-e2e.yml

@@ -26,7 +26,7 @@ jobs:
     - uses: ./.github/actions/start-localstack
 
       # Runs a single command using the runners shell
-    - name: Build API
+    - name: Build Webhook
       run: CI='' pnpm build:webhook
 
     # Runs a set of commands using the runners shell

.github/workflows/reusable-widget-e2e.yml

@@ -60,6 +60,7 @@ jobs:
           cypress_version: 12.9.0
           cypress: true
       - uses: ./.github/actions/setup-redis-cluster
+
       - uses: mansagroup/nrwl-nx-action@v3
         with:
           targets: lint

.github/workflows/test.yml

@@ -163,7 +163,6 @@ jobs:
       - uses: ./.github/actions/setup-redis-cluster
       - uses: mansagroup/nrwl-nx-action@v3
         env:
-          IN_MEMORY_CLUSTER_MODE_ENABLED: true
           LOGGING_LEVEL: 'info'
         with:
           targets: lint,build,test
@@ -213,14 +212,13 @@ jobs:
       - uses: ./.github/actions/start-localstack
 
       - uses: ./.github/actions/run-worker
-        if: ${{matrix.projectName == '@novu/api' }}
+        if: ${{matrix.projectName != '@novu/worker' }}
         with:
           launch_darkly_sdk_key: ${{ secrets.LAUNCH_DARKLY_SDK_KEY }}
 
       - uses: mansagroup/nrwl-nx-action@v3
         name: Running the E2E tests
         env:
-          IN_MEMORY_CLUSTER_MODE_ENABLED: true
           LAUNCH_DARKLY_SDK_KEY: ${{ secrets.LAUNCH_DARKLY_SDK_KEY }}
         with:
           targets: test:e2e
@@ -275,4 +273,3 @@ jobs:
           launch_darkly_sdk_key: ${{ secrets.LAUNCH_DARKLY_SDK_KEY }}
 
       - uses: ./.github/actions/validate-swagger
-

apps/api/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@novu/api",
-  "version": "0.18.0",
+  "version": "0.19.0",
   "description": "description",
   "author": "",
   "private": "true",
@@ -38,12 +38,12 @@
     "@nestjs/platform-express": "^10.2.2",
     "@nestjs/swagger": "^7.1.8",
     "@nestjs/terminus": "^10.0.1",
-    "@novu/application-generic": "^0.18.0",
-    "@novu/dal": "^0.18.0",
-    "@novu/node": "^0.18.0",
-    "@novu/shared": "^0.18.0",
-    "@novu/stateless": "^0.18.0",
-    "@novu/testing": "^0.18.0",
+    "@novu/application-generic": "^0.19.0",
+    "@novu/dal": "^0.19.0",
+    "@novu/node": "^0.19.0",
+    "@novu/shared": "^0.19.0",
+    "@novu/stateless": "^0.19.0",
+    "@novu/testing": "^0.19.0",
     "@sendgrid/mail": "^7.6.0",
     "@sentry/hub": "^7.40.0",
     "@sentry/node": "^7.40.0",
@@ -54,7 +54,6 @@
     "bcrypt": "^5.0.0",
     "body-parser": "^1.20.0",
     "bull": "^4.2.1",
-    "bullmq": "^3.10.2",
     "class-transformer": "^0.5.1",
     "class-validator": "^0.14.0",
     "compression": "^1.7.4",
@@ -70,7 +69,7 @@
     "lodash": "^4.17.15",
     "nanoid": "^3.1.20",
     "nest-raven": "^10.0.0",
-    "newrelic": "^11.0.0",
+    "newrelic": "^9.15.0",
     "passport": "0.6.0",
     "passport-github2": "^0.1.12",
     "passport-jwt": "^4.0.0",

apps/api/src/.env.development

@@ -25,7 +25,7 @@ REDIS_CACHE_FAMILY=
 REDIS_CACHE_KEY_PREFIX=
 REDIS_CACHE_ENABLE_AUTOPIPELINING=true
 
-IN_MEMORY_CLUSTER_MODE_ENABLED=true
+IS_IN_MEMORY_CLUSTER_MODE_ENABLED=false
 ELASTICACHE_CLUSTER_SERVICE_HOST=
 ELASTICACHE_CLUSTER_SERVICE_PORT=
 REDIS_CLUSTER_SERVICE_HOST=localhost

apps/api/src/.env.production

@@ -13,7 +13,7 @@ REDIS_PORT=6379
 REDIS_PREFIX=
 REDIS_DB_INDEX=2
 
-IN_MEMORY_CLUSTER_MODE_ENABLED=false
+IS_IN_MEMORY_CLUSTER_MODE_ENABLED=false
 ELASTICACHE_CLUSTER_SERVICE_HOST=
 ELASTICACHE_CLUSTER_SERVICE_PORT=
 REDIS_CLUSTER_SERVICE_HOST=

apps/api/src/.env.test

@@ -23,7 +23,7 @@ REDIS_CACHE_FAMILY=
 REDIS_CACHE_KEY_PREFIX=
 REDIS_CACHE_ENABLE_AUTOPIPELINING=false 
 
-IN_MEMORY_CLUSTER_MODE_ENABLED=false
+IS_IN_MEMORY_CLUSTER_MODE_ENABLED=false
 ELASTICACHE_CLUSTER_SERVICE_HOST=
 ELASTICACHE_CLUSTER_SERVICE_PORT=
 REDIS_CLUSTER_SERVICE_HOST=localhost
@@ -74,6 +74,7 @@ VERCEL_REDIRECT_URI=http://localhost:4200/auth/login
 VERCEL_BASE_URL=https://api.vercel.com
 
 FF_IS_TOPIC_NOTIFICATION_ENABLED=true
+IS_MULTI_PROVIDER_CONFIGURATION_ENABLED=true
 
 STORE_NOTIFICATION_CONTENT=true
 

apps/api/src/.example.env

@@ -23,7 +23,7 @@ REDIS_CACHE_FAMILY=
 REDIS_CACHE_KEY_PREFIX=
 REDIS_CACHE_ENABLE_AUTOPIPELINING=
 
-IN_MEMORY_CLUSTER_MODE_ENABLED=false
+IS_IN_MEMORY_CLUSTER_MODE_ENABLED=false
 REDIS_CLUSTER_SERVICE_HOST=
 REDIS_CLUSTER_SERVICE_PORT=
 REDIS_CLUSTER_DB_INDEX=

apps/api/src/app/blueprint/e2e/get-grouped-blueprints.e2e.ts

@@ -7,6 +7,7 @@ import { EmailBlockTypeEnum, FilterPartTypeEnum, INotificationTemplate, StepType
 import {
   buildGroupedBlueprintsKey,
   CacheService,
+  InMemoryProviderEnum,
   InMemoryProviderService,
   InvalidateCacheService,
 } from '@novu/application-generic';
@@ -21,14 +22,16 @@ describe('Get grouped notification template blueprints - /blueprints/group-by-ca
   const notificationTemplateRepository: NotificationTemplateRepository = new NotificationTemplateRepository();
   const environmentRepository: EnvironmentRepository = new EnvironmentRepository();
 
-  const inMemoryProviderService = new InMemoryProviderService();
-  inMemoryProviderService.initialize();
-  const invalidateCache = new InvalidateCacheService(new CacheService(inMemoryProviderService));
-
+  let invalidateCache: InvalidateCacheService;
   let getGroupedBlueprints: GetGroupedBlueprints;
   let indexModuleStub: sinon.SinonStub;
 
   before(async () => {
+    const inMemoryProviderService = new InMemoryProviderService(InMemoryProviderEnum.REDIS);
+    const cacheService = new CacheService(inMemoryProviderService);
+    await cacheService.initialize();
+    invalidateCache = new InvalidateCacheService(cacheService);
+
     session = new UserSession();
     await session.initialize();
 

apps/api/src/app/events/e2e/delay-events.e2e.ts

@@ -11,7 +11,7 @@ import {
 } from '@novu/dal';
 import { UserSession, SubscribersService } from '@novu/testing';
 import { StepTypeEnum, DelayTypeEnum, DigestUnitEnum, DigestTypeEnum } from '@novu/shared';
-import { QueueService } from '@novu/application-generic';
+import { StandardQueueService } from '@novu/application-generic';
 
 const axiosInstance = axios.create();
 
@@ -21,7 +21,7 @@ describe('Trigger event - Delay triggered events - /v1/events/trigger (POST)', f
   let subscriber: SubscriberEntity;
   let subscriberService: SubscribersService;
   const jobRepository = new JobRepository();
-  let queueService: QueueService;
+  let standardQueueService: StandardQueueService;
   const messageRepository = new MessageRepository();
 
   const triggerEvent = async (payload, transactionId?: string, overrides = {}) => {
@@ -48,7 +48,7 @@ describe('Trigger event - Delay triggered events - /v1/events/trigger (POST)', f
     template = await session.createTemplate();
     subscriberService = new SubscribersService(session.organization._id, session.environment._id);
     subscriber = await subscriberService.createSubscriber();
-    queueService = session?.testServer?.getService(QueueService);
+    standardQueueService = session?.testServer?.getService(StandardQueueService);
   });
 
   it('should delay event for time interval', async function () {
@@ -141,7 +141,7 @@ describe('Trigger event - Delay triggered events - /v1/events/trigger (POST)', f
         customVar: 'Testing of User Name',
       },
       id,
-      { delay: { amount: 3, unit: DigestUnitEnum.SECONDS } }
+      { delay: { amount: 1, unit: DigestUnitEnum.SECONDS } }
     );
     await session.awaitRunningJobs(template?._id, true, 0);
     const messages = await messageRepository.find({
@@ -190,7 +190,7 @@ describe('Trigger event - Delay triggered events - /v1/events/trigger (POST)', f
     const updatedAt = delayedJob?.updatedAt as string;
     const diff = differenceInMilliseconds(new Date(delayedJob.payload.sendAt), new Date(updatedAt));
 
-    const delay = await queueService.bullMqService.queue.getDelayed();
+    const delay = await standardQueueService.queue.getDelayed();
     expect(delay[0].opts.delay).to.approximately(diff, 1000);
   });
 

apps/api/src/app/events/e2e/digest-events.e2e.ts

@@ -12,7 +12,6 @@ import {
 } from '@novu/dal';
 import { StepTypeEnum, DigestTypeEnum, DigestUnitEnum, IDigestRegularMetadata } from '@novu/shared';
 import { UserSession, SubscribersService } from '@novu/testing';
-import { QueueService } from '@novu/application-generic';
 
 const axiosInstance = axios.create();
 
@@ -24,7 +23,6 @@ describe('Trigger event - Digest triggered events - /v1/events/trigger (POST)',
   let subscriber: SubscriberEntity;
   let subscriberService: SubscribersService;
   const jobRepository = new JobRepository();
-  let queueService: QueueService;
   const messageRepository = new MessageRepository();
 
   const triggerEvent = async (payload, transactionId?: string): Promise<void> => {
@@ -50,7 +48,6 @@ describe('Trigger event - Digest triggered events - /v1/events/trigger (POST)',
     template = await session.createTemplate();
     subscriberService = new SubscribersService(session.organization._id, session.environment._id);
     subscriber = await subscriberService.createSubscriber();
-    queueService = session.testServer?.getService(QueueService);
   });
 
   it('should digest events within time interval', async function () {

apps/api/src/app/events/e2e/process-subscriber.e2e.ts

@@ -8,13 +8,14 @@ import {
   NotificationTemplateRepository,
 } from '@novu/dal';
 import { UserSession, SubscribersService } from '@novu/testing';
-import { ChannelTypeEnum, ISubscribersDefine, StepTypeEnum } from '@novu/shared';
+import { ChannelTypeEnum, ISubscribersDefine, IUpdateNotificationTemplateDto, StepTypeEnum } from '@novu/shared';
 import {
-  InMemoryProviderService,
   buildNotificationTemplateIdentifierKey,
   buildNotificationTemplateKey,
-  InvalidateCacheService,
   CacheService,
+  InMemoryProviderEnum,
+  InMemoryProviderService,
+  InvalidateCacheService,
 } from '@novu/application-generic';
 
 import { UpdateSubscriberPreferenceRequestDto } from '../../widgets/dtos/update-subscriber-preference-request.dto';
@@ -26,18 +27,25 @@ describe('Trigger event - process subscriber /v1/events/trigger (POST)', functio
   let template: NotificationTemplateEntity;
   let subscriber: SubscriberEntity;
   let subscriberService: SubscribersService;
-
-  const inMemoryProviderService = new InMemoryProviderService();
-  inMemoryProviderService.initialize();
-  const invalidateCache = new InvalidateCacheService(new CacheService(inMemoryProviderService));
+  let cacheService: CacheService;
+  let invalidateCache: InvalidateCacheService;
+  let inMemoryProviderService: InMemoryProviderService;
 
   const subscriberRepository = new SubscriberRepository();
   const messageRepository = new MessageRepository();
   const notificationTemplateRepository = new NotificationTemplateRepository();
 
+  before(async () => {
+    inMemoryProviderService = new InMemoryProviderService(InMemoryProviderEnum.REDIS);
+    cacheService = new CacheService(inMemoryProviderService);
+    await cacheService.initialize();
+    invalidateCache = new InvalidateCacheService(cacheService);
+  });
+
   beforeEach(async () => {
     session = new UserSession();
     await session.initialize();
+
     template = await session.createTemplate();
     subscriberService = new SubscribersService(session.organization._id, session.environment._id);
     subscriber = await subscriberService.createSubscriber();
@@ -188,6 +196,14 @@ describe('Trigger event - process subscriber /v1/events/trigger (POST)', functio
 
     expect(message.length).to.equal(2);
 
+    const notificationTemplateKey = buildNotificationTemplateKey({
+      _id: template._id,
+      _environmentId: session.environment._id,
+    });
+    await invalidateCache.invalidateByKey({
+      key: notificationTemplateKey,
+    });
+
     const updateData = {
       channel: {
         type: ChannelTypeEnum.IN_APP,
@@ -197,21 +213,7 @@ describe('Trigger event - process subscriber /v1/events/trigger (POST)', functio
 
     await updateSubscriberPreference(updateData, session.subscriberToken, template._id);
 
-    await invalidateCache.invalidateByKey({
-      key: buildNotificationTemplateKey({
-        _id: template._id,
-        _environmentId: session.environment._id,
-      }),
-    });
-
-    await invalidateCache.invalidateByKey({
-      key: buildNotificationTemplateIdentifierKey({
-        templateIdentifier: template.triggers[0].identifier,
-        _environmentId: session.environment._id,
-      }),
-    });
-
-    await notificationTemplateRepository.update(
+    const rest = await notificationTemplateRepository.update(
       {
         _id: template._id,
         _environmentId: session.environment._id,