Merge pull request #4733 from novuhq/nv-3136-create-dedicated-service…

…-for-workflow-and-cache-in-memory

feat(infra): create dedicated services for in-memory for wf and cache

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

@@ -13,9 +13,8 @@ import {
 } from '@novu/shared';
 import {
   buildGroupedBlueprintsKey,
+  CacheInMemoryProviderService,
   CacheService,
-  InMemoryProviderEnum,
-  InMemoryProviderService,
   InvalidateCacheService,
 } from '@novu/application-generic';
 
@@ -34,8 +33,8 @@ describe('Get grouped notification template blueprints - /blueprints/group-by-ca
   let indexModuleStub: sinon.SinonStub;
 
   before(async () => {
-    const inMemoryProviderService = new InMemoryProviderService(InMemoryProviderEnum.REDIS);
-    const cacheService = new CacheService(inMemoryProviderService);
+    const cacheInMemoryProviderService = new CacheInMemoryProviderService();
+    const cacheService = new CacheService(cacheInMemoryProviderService);
     await cacheService.initialize();
     invalidateCache = new InvalidateCacheService(cacheService);
 

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

@@ -12,9 +12,8 @@ import { ChannelTypeEnum, ISubscribersDefine, IUpdateNotificationTemplateDto, St
 import {
   buildNotificationTemplateIdentifierKey,
   buildNotificationTemplateKey,
+  CacheInMemoryProviderService,
   CacheService,
-  InMemoryProviderEnum,
-  InMemoryProviderService,
   InvalidateCacheService,
 } from '@novu/application-generic';
 
@@ -29,15 +28,15 @@ describe('Trigger event - process subscriber /v1/events/trigger (POST)', functio
   let subscriberService: SubscribersService;
   let cacheService: CacheService;
   let invalidateCache: InvalidateCacheService;
-  let inMemoryProviderService: InMemoryProviderService;
+  let cacheInMemoryProviderService: CacheInMemoryProviderService;
 
   const subscriberRepository = new SubscriberRepository();
   const messageRepository = new MessageRepository();
   const notificationTemplateRepository = new NotificationTemplateRepository();
 
   before(async () => {
-    inMemoryProviderService = new InMemoryProviderService(InMemoryProviderEnum.REDIS);
-    cacheService = new CacheService(inMemoryProviderService);
+    cacheInMemoryProviderService = new CacheInMemoryProviderService();
+    cacheService = new CacheService(cacheInMemoryProviderService);
     await cacheService.initialize();
     invalidateCache = new InvalidateCacheService(cacheService);
   });

apps/api/src/app/health/e2e/health-check.e2e.ts

@@ -1,13 +1,10 @@
 import { UserSession } from '@novu/testing';
-import { InMemoryProviderEnum, InMemoryProviderService } from '@novu/application-generic';
 import { expect } from 'chai';
 
 describe('Health-check', () => {
   const session = new UserSession();
 
   before(async () => {
-    const inMemoryProviderService = new InMemoryProviderService(InMemoryProviderEnum.REDIS);
-
     await session.initialize();
   });
 

apps/api/src/app/widgets/e2e/get-count.e2e.ts

@@ -6,9 +6,8 @@ import { ChannelTypeEnum, InAppProviderIdEnum } from '@novu/shared';
 import {
   buildFeedKey,
   buildMessageCountKey,
+  CacheInMemoryProviderService,
   CacheService,
-  InMemoryProviderEnum,
-  InMemoryProviderService,
   InvalidateCacheService,
 } from '@novu/application-generic';
 
@@ -23,11 +22,11 @@ describe('Count - GET /widget/notifications/count', function () {
   } | null = null;
 
   let invalidateCache: InvalidateCacheService;
-  let inMemoryProviderService: InMemoryProviderService;
+  let cacheInMemoryProviderService: CacheInMemoryProviderService;
 
   before(async () => {
-    inMemoryProviderService = new InMemoryProviderService(InMemoryProviderEnum.REDIS);
-    const cacheService = new CacheService(inMemoryProviderService);
+    cacheInMemoryProviderService = new CacheInMemoryProviderService();
+    const cacheService = new CacheService(cacheInMemoryProviderService);
     await cacheService.initialize();
     invalidateCache = new InvalidateCacheService(cacheService);
   });
@@ -256,7 +255,7 @@ describe('Count - GET /widget/notifications/count', function () {
 
     const messages = await messageRepository.findBySubscriberChannel(
       session.environment._id,
-      subscriberProfile!._id,
+      String(subscriberProfile?._id),
       ChannelTypeEnum.IN_APP
     );
     const messageId = messages[0]._id;
@@ -319,7 +318,7 @@ describe('Count - GET /widget/notifications/count', function () {
 
     const messages = await messageRepository.findBySubscriberChannel(
       session.environment._id,
-      subscriberProfile!._id,
+      String(subscriberProfile?._id),
       ChannelTypeEnum.IN_APP
     );
     const messageId = messages[0]._id;

apps/api/src/app/widgets/e2e/get-unseen-count.e2e.ts

@@ -6,9 +6,8 @@ import { ChannelTypeEnum } from '@novu/shared';
 import {
   buildFeedKey,
   buildMessageCountKey,
+  CacheInMemoryProviderService,
   CacheService,
-  InMemoryProviderEnum,
-  InMemoryProviderService,
   InvalidateCacheService,
 } from '@novu/application-generic';
 
@@ -22,12 +21,12 @@ describe('Unseen Count - GET /widget/notifications/unseen', function () {
     _id: string;
   } | null = null;
 
-  let inMemoryProviderService: InMemoryProviderService;
+  let cacheInMemoryProviderService: CacheInMemoryProviderService;
   let invalidateCache: InvalidateCacheService;
 
   before(async () => {
-    inMemoryProviderService = new InMemoryProviderService(InMemoryProviderEnum.REDIS);
-    const cacheService = new CacheService(inMemoryProviderService);
+    cacheInMemoryProviderService = new CacheInMemoryProviderService();
+    const cacheService = new CacheService(cacheInMemoryProviderService);
     await cacheService.initialize();
     invalidateCache = new InvalidateCacheService(cacheService);
   });
@@ -68,7 +67,7 @@ describe('Unseen Count - GET /widget/notifications/unseen', function () {
 
     const messages = await messageRepository.findBySubscriberChannel(
       session.environment._id,
-      subscriberProfile!._id,
+      String(subscriberProfile?._id),
       ChannelTypeEnum.IN_APP
     );
     const messageId = messages[0]._id;
@@ -97,7 +96,7 @@ describe('Unseen Count - GET /widget/notifications/unseen', function () {
 
     const messages = await messageRepository.findBySubscriberChannel(
       session.environment._id,
-      subscriberProfile!._id,
+      String(subscriberProfile?._id),
       ChannelTypeEnum.IN_APP
     );
     const messageId = messages[0]._id;
@@ -126,7 +125,7 @@ describe('Unseen Count - GET /widget/notifications/unseen', function () {
 
     const messages = await messageRepository.findBySubscriberChannel(
       session.environment._id,
-      subscriberProfile!._id,
+      String(subscriberProfile?._id),
       ChannelTypeEnum.IN_APP
     );
     const messageId = messages[0]._id;
@@ -155,7 +154,7 @@ describe('Unseen Count - GET /widget/notifications/unseen', function () {
 
     const messages = await messageRepository.findBySubscriberChannel(
       session.environment._id,
-      subscriberProfile!._id,
+      String(subscriberProfile?._id),
       ChannelTypeEnum.IN_APP
     );
     const messageId = messages[0]._id;

apps/api/src/app/widgets/e2e/initialize-widget-session.e2e.ts

@@ -5,10 +5,9 @@ import { expect } from 'chai';
 import { createHash } from '../../shared/helpers/hmac.service';
 import {
   buildIntegrationKey,
+  CacheInMemoryProviderService,
   CacheService,
-  InMemoryProviderService,
   InvalidateCacheService,
-  InMemoryProviderEnum,
 } from '@novu/application-generic';
 
 const integrationRepository = new IntegrationRepository();
@@ -19,8 +18,8 @@ describe('Initialize Session - /widgets/session/initialize (POST)', async () =>
   let invalidateCache: InvalidateCacheService;
 
   before(async () => {
-    const inMemoryProviderService = new InMemoryProviderService(InMemoryProviderEnum.REDIS);
-    const cacheService = new CacheService(inMemoryProviderService);
+    const cacheInMemoryProviderService = new CacheInMemoryProviderService();
+    const cacheService = new CacheService(cacheInMemoryProviderService);
     await cacheService.initialize();
     invalidateCache = new InvalidateCacheService(cacheService);
   });

apps/worker/src/app/health/e2e/health-check.e2e.ts

@@ -1,4 +1,3 @@
-import { InMemoryProviderEnum, InMemoryProviderService } from '@novu/application-generic';
 import { expect } from 'chai';
 import * as request from 'supertest';
 import * as defaults from 'superagent-defaults';
@@ -7,9 +6,6 @@ describe('Health-check', () => {
   let testAgent;
 
   before(async () => {
-    const inMemoryProviderService = new InMemoryProviderService(InMemoryProviderEnum.REDIS);
-    await inMemoryProviderService.delayUntilReadiness();
-
     testAgent = defaults(request(`http://localhost:${process.env.PORT}`));
   });
 

packages/application-generic/src/custom-providers/index.ts

@@ -1,22 +1,21 @@
 import {
   AnalyticsService,
   BullMqService,
+  CacheInMemoryProviderService,
   CacheService,
   DistributedLockService,
   FeatureFlagsService,
-  InMemoryProviderEnum,
-  InMemoryProviderService,
   ReadinessService,
   OldInstanceBullMqService,
   StandardQueueService,
+  SubscriberProcessQueueService,
   WebSocketsQueueService,
   WorkflowQueueService,
 } from '../services';
 import {
   GetIsTopicNotificationEnabled,
   GetUseMergedDigestId,
 } from '../usecases';
-import { SubscriberProcessQueueService } from '../services/queues/subscriber-process-queue.service';
 
 export const featureFlagsService = {
   provide: FeatureFlagsService,
@@ -52,13 +51,10 @@ export const getIsTopicNotificationEnabled = {
   inject: [FeatureFlagsService],
 };
 
-export const inMemoryProviderService = {
-  provide: InMemoryProviderService,
-  useFactory: (
-    provider: InMemoryProviderEnum,
-    enableAutoPipelining?: boolean
-  ): InMemoryProviderService => {
-    return new InMemoryProviderService(provider, enableAutoPipelining);
+export const cacheInMemoryProviderService = {
+  provide: CacheInMemoryProviderService,
+  useFactory: (): CacheInMemoryProviderService => {
+    return new CacheInMemoryProviderService();
   },
 };
 
@@ -87,14 +83,10 @@ export const oldInstanceBullMqService = {
 export const cacheService = {
   provide: CacheService,
   useFactory: async (): Promise<CacheService> => {
-    const enableAutoPipelining =
-      process.env.REDIS_CACHE_ENABLE_AUTOPIPELINING === 'false';
-    const factoryInMemoryProviderService = inMemoryProviderService.useFactory(
-      InMemoryProviderEnum.ELASTICACHE,
-      enableAutoPipelining
-    );
+    const factoryCacheInMemoryProviderService =
+      cacheInMemoryProviderService.useFactory();
 
-    const service = new CacheService(factoryInMemoryProviderService);
+    const service = new CacheService(factoryCacheInMemoryProviderService);
 
     await service.initialize();
 
@@ -115,11 +107,12 @@ export const analyticsService = {
 export const distributedLockService = {
   provide: DistributedLockService,
   useFactory: async (): Promise<DistributedLockService> => {
-    const factoryInMemoryProviderService = inMemoryProviderService.useFactory(
-      InMemoryProviderEnum.ELASTICACHE
-    );
+    const factoryCacheInMemoryProviderService =
+      cacheInMemoryProviderService.useFactory();
 
-    const service = new DistributedLockService(factoryInMemoryProviderService);
+    const service = new DistributedLockService(
+      factoryCacheInMemoryProviderService
+    );
 
     await service.initialize();
 

packages/application-generic/src/health/active-jobs-metric-queue.health-indicator.ts

@@ -1,6 +1,6 @@
 import { Injectable } from '@nestjs/common';
 
-import { ActiveJobsMetricQueueService } from '../services';
+import { ActiveJobsMetricQueueService } from '../services/queues';
 import { QueueHealthIndicator } from './queue-health-indicator.service';
 
 const LOG_CONTEXT = 'ActiveJobsMetricQueueServiceHealthIndicator';

packages/application-generic/src/health/cache.health-indicator.ts

@@ -5,7 +5,7 @@ import {
 } from '@nestjs/terminus';
 import { Injectable, Logger } from '@nestjs/common';
 
-import { CacheService } from '../services';
+import { CacheService } from '../services/cache';
 
 const LOG_CONTEXT = 'CacheServiceHealthIndicator';
 

packages/application-generic/src/health/completed-jobs-metric-queue.health-indicator.ts

@@ -1,6 +1,6 @@
 import { Injectable } from '@nestjs/common';
 
-import { CompletedJobsMetricQueueService } from '../services';
+import { CompletedJobsMetricQueueService } from '../services/queues';
 import { QueueHealthIndicator } from './queue-health-indicator.service';
 
 const LOG_CONTEXT = 'CompletedJobsMetricQueueServiceHealthIndicator';

packages/application-generic/src/health/inbound-parse-queue.health-indicator.ts

@@ -1,6 +1,6 @@
 import { Injectable } from '@nestjs/common';
 
-import { InboundParseQueue } from '../services';
+import { InboundParseQueue } from '../services/queues';
 import { QueueHealthIndicator } from './queue-health-indicator.service';
 
 const LOG_CONTEXT = 'InboundParseQueueServiceHealthIndicator';

packages/application-generic/src/health/queue-health-indicator.service.ts

@@ -5,7 +5,7 @@ import {
 } from '@nestjs/terminus';
 import { Injectable, Logger } from '@nestjs/common';
 
-import { QueueBaseService } from '../services';
+import { QueueBaseService } from '../services/queues/queue-base.service';
 import { IHealthIndicator } from './health-indicator.interface';
 
 @Injectable()

packages/application-generic/src/health/standard-queue.health-indicator.ts

@@ -1,6 +1,6 @@
 import { Injectable } from '@nestjs/common';
 
-import { StandardQueueService } from '../services';
+import { StandardQueueService } from '../services/queues';
 import { QueueHealthIndicator } from './queue-health-indicator.service';
 
 const LOG_CONTEXT = 'StandardQueueServiceHealthIndicator';

packages/application-generic/src/health/subscriber-process-queue.health-indicator.ts

@@ -1,6 +1,6 @@
 import { Injectable } from '@nestjs/common';
 
-import { SubscriberProcessQueueService } from '../services';
+import { SubscriberProcessQueueService } from '../services/queues';
 import { ObservabilityBackgroundTransactionEnum } from '@novu/shared';
 import { QueueHealthIndicator } from './queue-health-indicator.service';
 

packages/application-generic/src/health/web-sockets-queue.health-indicator.ts

@@ -1,6 +1,6 @@
 import { Injectable } from '@nestjs/common';
 
-import { WebSocketsQueueService } from '../services';
+import { WebSocketsQueueService } from '../services/queues';
 import { QueueHealthIndicator } from './queue-health-indicator.service';
 
 const LOG_CONTEXT = 'WebSocketsQueueServiceHealthIndicator';

packages/application-generic/src/health/workflow-queue.health-indicator.ts

@@ -1,6 +1,6 @@
 import { Injectable } from '@nestjs/common';
 
-import { WorkflowQueueService } from '../services';
+import { WorkflowQueueService } from '../services/queues';
 import { QueueHealthIndicator } from './queue-health-indicator.service';
 
 const LOG_CONTEXT = 'WorkflowQueueServiceHealthIndicator';

packages/application-generic/src/modules/queues.module.ts

@@ -6,6 +6,7 @@ import {
   CompletedJobsMetricQueueServiceHealthIndicator,
   InboundParseQueueServiceHealthIndicator,
   StandardQueueServiceHealthIndicator,
+  SubscriberProcessQueueHealthIndicator,
   WebSocketsQueueServiceHealthIndicator,
   WorkflowQueueServiceHealthIndicator,
 } from '../health';
@@ -15,6 +16,7 @@ import {
   CompletedJobsMetricQueueService,
   InboundParseQueue,
   StandardQueueService,
+  SubscriberProcessQueueService,
   WebSocketsQueueService,
   WorkflowQueueService,
 } from '../services/queues';
@@ -23,16 +25,13 @@ import {
   CompletedJobsMetricWorkerService,
   InboundParseWorker,
   StandardWorkerService,
+  SubscriberProcessWorkerService,
   WebSocketsWorkerService,
   WorkflowWorkerService,
   OldInstanceStandardWorkerService,
   OldInstanceWorkflowWorkerService,
 } from '../services/workers';
 
-import { SubscriberProcessQueueService } from '../services/queues/subscriber-process-queue.service';
-import { SubscriberProcessWorkerService } from '../services/workers/subscriber-process-worker.service';
-import { SubscriberProcessQueueHealthIndicator } from '../health/subscriber-process-queue.health-indicator';
-
 const PROVIDERS: Provider[] = [
   ActiveJobsMetricQueueService,
   ActiveJobsMetricQueueServiceHealthIndicator,