[pull] next from novuhq:next

apps/worker/package.json

@@ -43,6 +43,7 @@
     "@sentry/tracing": "^7.40.0",
     "@types/newrelic": "^9.14.6",
     "svix": "^1.64.1",
+    "lru-cache": "^11.2.4",
     "axios": "^1.9.0",
     "body-parser": "^2.2.0",
     "class-transformer": "0.5.1",

apps/worker/src/app/workflow/usecases/run-job/run-job.usecase.ts

@@ -34,6 +34,7 @@ import {
 import { setUser } from '@sentry/node';
 import { differenceInMilliseconds } from 'date-fns';
 import { formatInTimeZone } from 'date-fns-tz';
+import { LRUCache } from 'lru-cache';
 import { EXCEPTION_MESSAGE_ON_WEBHOOK_FILTER, PlatformException, shouldHaltOnStepFailure } from '../../../shared/utils';
 import { AddJob } from '../add-job';
 import { PartialNotificationEntity } from '../add-job/add-job.command';
@@ -48,6 +49,13 @@ import { calculateNextAvailableTime, isWithinSchedule } from './schedule-validat
 
 const nr = require('newrelic');
 
+const workflowCache = new LRUCache<string, NotificationTemplateEntity>({
+  max: 1000,
+  ttl: 1000 * 30,
+});
+
+const workflowInflightRequests = new Map<string, Promise<NotificationTemplateEntity | undefined>>();
+
 export type SelectedWorkflowFields = Pick<NotificationTemplateEntity, 'steps'>;
 
 /**
@@ -152,8 +160,12 @@ export class RunJob {
         throw new PlatformException(`Notification with id ${job._notificationId} not found`);
       }
 
-      const workflow =
-        (await this.notificationTemplateRepository.findById(job._templateId, job._environmentId)) ?? undefined;
+      const workflow = await this.getWorkflow(
+        job._templateId,
+        job._environmentId,
+        job._organizationId,
+        job.payload?.__source
+      );
 
       if (isSubscribersScheduleEnabled) {
         const schedule = await this.getSubscriberSchedule.execute(
@@ -375,6 +387,59 @@ export class RunJob {
     }
   }
 
+  @Instrument()
+  private async getWorkflow(
+    templateId: string,
+    environmentId: string,
+    organizationId: string,
+    source?: string
+  ): Promise<NotificationTemplateEntity | undefined> {
+    const cacheKey = `${environmentId}:${templateId}`;
+
+    const isFeatureFlagEnabled = await this.featureFlagsService.getFlag({
+      key: FeatureFlagsKeysEnum.IS_LRU_CACHE_ENABLED,
+      defaultValue: false,
+      environment: { _id: environmentId },
+      organization: { _id: organizationId },
+      component: 'worker-workflow',
+    });
+
+    const isCacheEnabled = isFeatureFlagEnabled && !source;
+
+    if (isCacheEnabled) {
+      const cached = workflowCache.get(cacheKey);
+      if (cached) {
+        return cached;
+      }
+
+      const inflightRequest = workflowInflightRequests.get(cacheKey);
+      if (inflightRequest) {
+        return inflightRequest;
+      }
+    }
+
+    const fetchPromise = this.notificationTemplateRepository
+      .findById(templateId, environmentId)
+      .then((workflow) => {
+        if (workflow && isCacheEnabled) {
+          workflowCache.set(cacheKey, workflow);
+        }
+
+        return workflow ?? undefined;
+      })
+      .finally(() => {
+        if (isCacheEnabled) {
+          workflowInflightRequests.delete(cacheKey);
+        }
+      });
+
+    if (isCacheEnabled) {
+      workflowInflightRequests.set(cacheKey, fetchPromise);
+    }
+
+    return fetchPromise;
+  }
+
   private isUnsnoozeJob(job: JobEntity) {
     return job.type === StepTypeEnum.IN_APP && job.delay && job.payload?.unsnooze;
   }

apps/worker/src/app/workflow/usecases/subscriber-job-bound/subscriber-job-bound.usecase.ts

@@ -6,6 +6,7 @@ import {
   CreateNotificationJobsCommand,
   CreateOrUpdateSubscriberCommand,
   CreateOrUpdateSubscriberUseCase,
+  FeatureFlagsService,
   GetPreferences,
   GetPreferencesCommand,
   Instrument,
@@ -19,18 +20,27 @@ import { IntegrationRepository, NotificationTemplateEntity, NotificationTemplate
 import {
   buildWorkflowPreferences,
   ChannelTypeEnum,
+  FeatureFlagsKeysEnum,
   InAppProviderIdEnum,
   ISubscribersDefine,
   ProvidersIdEnum,
   ResourceTypeEnum,
   SeverityLevelEnum,
   STEP_TYPE_TO_CHANNEL_TYPE,
 } from '@novu/shared';
+import { LRUCache } from 'lru-cache';
 import { StoreSubscriberJobs, StoreSubscriberJobsCommand } from '../store-subscriber-jobs';
 import { SubscriberJobBoundCommand } from './subscriber-job-bound.command';
 
 const LOG_CONTEXT = 'SubscriberJobBoundUseCase';
 
+const workflowCache = new LRUCache<string, NotificationTemplateEntity>({
+  max: 1000,
+  ttl: 1000 * 30,
+});
+
+const workflowInflightRequests = new Map<string, Promise<NotificationTemplateEntity | null>>();
+
 @Injectable()
 export class SubscriberJobBound {
   constructor(
@@ -42,7 +52,8 @@ export class SubscriberJobBound {
     private logger: PinoLogger,
     private analyticsService: AnalyticsService,
     private traceLogRepository: TraceLogRepository,
-    private getPreferences: GetPreferences
+    private getPreferences: GetPreferences,
+    private featureFlagsService: FeatureFlagsService
   ) {}
 
   @InstrumentUsecase()
@@ -74,6 +85,8 @@ export class SubscriberJobBound {
       : await this.getWorkflow({
           _id: templateId,
           environmentId,
+          organizationId,
+          source: command.payload?.__source,
         });
 
     if (!template) {
@@ -268,8 +281,62 @@ export class SubscriberJobBound {
     return true;
   }
 
-  private async getWorkflow({ _id, environmentId }: { _id: string; environmentId: string }) {
-    return await this.notificationTemplateRepository.findById(_id, environmentId);
+  @Instrument()
+  private async getWorkflow({
+    _id,
+    environmentId,
+    organizationId,
+    source,
+  }: {
+    _id: string;
+    environmentId: string;
+    organizationId: string;
+    source?: string;
+  }): Promise<NotificationTemplateEntity | null> {
+    const cacheKey = `${environmentId}:${_id}`;
+
+    const isFeatureFlagEnabled = await this.featureFlagsService.getFlag({
+      key: FeatureFlagsKeysEnum.IS_LRU_CACHE_ENABLED,
+      defaultValue: false,
+      environment: { _id: environmentId },
+      organization: { _id: organizationId },
+      component: 'worker-workflow',
+    });
+
+    const isCacheEnabled = isFeatureFlagEnabled && !source;
+
+    if (isCacheEnabled) {
+      const cached = workflowCache.get(cacheKey);
+      if (cached) {
+        return cached;
+      }
+
+      const inflightRequest = workflowInflightRequests.get(cacheKey);
+      if (inflightRequest) {
+        return inflightRequest;
+      }
+    }
+
+    const fetchPromise = this.notificationTemplateRepository
+      .findById(_id, environmentId)
+      .then((workflow) => {
+        if (workflow && isCacheEnabled) {
+          workflowCache.set(cacheKey, workflow);
+        }
+
+        return workflow;
+      })
+      .finally(() => {
+        if (isCacheEnabled) {
+          workflowInflightRequests.delete(cacheKey);
+        }
+      });
+
+    if (isCacheEnabled) {
+      workflowInflightRequests.set(cacheKey, fetchPromise);
+    }
+
+    return fetchPromise;
   }
 
   @InstrumentUsecase()