fix(worker): refactor template fetch to reduce db calls (#9609)

scopsy · web-flow · commit db45bbde3972 · 2025-12-03T19:15:20.000+02:00
diff --git a/apps/worker/src/app/workflow/usecases/add-job/add-job.usecase.ts b/apps/worker/src/app/workflow/usecases/add-job/add-job.usecase.ts
@@ -28,7 +28,7 @@ import {
   TierRestrictionsValidateUsecase,
   WorkflowRunStatusEnum,
 } from '@novu/application-generic';
-import { JobEntity, JobRepository, JobStatusEnum, SubscriberRepository } from '@novu/dal';
+import { JobEntity, JobRepository, JobStatusEnum, NotificationTemplateEntity, SubscriberRepository } from '@novu/dal';
 import { DelayOutput, DigestOutput, ExecuteOutput } from '@novu/framework/internal';
 import {
   castUnitToDigestUnitEnum,
@@ -515,13 +515,15 @@ export class AddJob {
 
   private async fetchBridgeData(
     command: AddJobCommand,
-    filterVariables: IFilterVariables
+    filterVariables: IFilterVariables,
+    workflow?: NotificationTemplateEntity
   ): Promise<ExecuteOutput | null> {
     const response = await this.executeBridgeJob.execute(
       ExecuteBridgeJobCommand.create({
         identifier: command.job.identifier,
         ...command,
         variables: filterVariables,
+        workflow,
       })
     );
 
diff --git a/apps/worker/src/app/workflow/usecases/execute-bridge-job/execute-bridge-job.command.ts b/apps/worker/src/app/workflow/usecases/execute-bridge-job/execute-bridge-job.command.ts
@@ -1,7 +1,7 @@
 import { EnvironmentWithUserCommand, IFilterVariables } from '@novu/application-generic';
 
-import { JobEntity } from '@novu/dal';
-import { IsDefined, IsString } from 'class-validator';
+import { JobEntity, NotificationTemplateEntity } from '@novu/dal';
+import { IsDefined, IsOptional, IsString } from 'class-validator';
 
 export class ExecuteBridgeJobCommand extends EnvironmentWithUserCommand {
   @IsDefined()
@@ -28,4 +28,7 @@ export class ExecuteBridgeJobCommand extends EnvironmentWithUserCommand {
 
   @IsDefined()
   variables?: IFilterVariables;
+
+  @IsOptional()
+  workflow?: NotificationTemplateEntity;
 }
diff --git a/apps/worker/src/app/workflow/usecases/execute-bridge-job/execute-bridge-job.usecase.ts b/apps/worker/src/app/workflow/usecases/execute-bridge-job/execute-bridge-job.usecase.ts
@@ -63,16 +63,23 @@ export class ExecuteBridgeJob {
 
     let workflow: NotificationTemplateEntity | null = null;
     if (isStateful) {
-      workflow = await this.notificationTemplateRepository.findOne(
-        {
-          _id: command.job._templateId,
-          _environmentId: command.environmentId,
-          type: {
-            $in: [ResourceTypeEnum.ECHO, ResourceTypeEnum.BRIDGE],
+      if (
+        command.workflow &&
+        (command.workflow.type === ResourceTypeEnum.ECHO || command.workflow.type === ResourceTypeEnum.BRIDGE)
+      ) {
+        workflow = command.workflow;
+      } else {
+        workflow = await this.notificationTemplateRepository.findOne(
+          {
+            _id: command.job._templateId,
+            _environmentId: command.environmentId,
+            type: {
+              $in: [ResourceTypeEnum.ECHO, ResourceTypeEnum.BRIDGE],
+            },
           },
-        },
-        '_id triggers type origin'
-      );
+          '_id triggers type origin'
+        );
+      }
     }
 
     if (!workflow && isStateful) {
diff --git a/apps/worker/src/app/workflow/usecases/run-job/run-job.usecase.ts b/apps/worker/src/app/workflow/usecases/run-job/run-job.usecase.ts
@@ -109,7 +109,7 @@ export class RunJob {
       });
 
       // Update workflow run delivery lifecycle after job cancellation
-      await this.conditionallyUpdateDeliveryLifecycle(job, WorkflowRunStatusEnum.COMPLETED);
+      await this.conditionallyUpdateDeliveryLifecycle(job, WorkflowRunStatusEnum.COMPLETED, undefined);
 
       return;
     }
@@ -152,6 +152,9 @@ export class RunJob {
         throw new PlatformException(`Notification with id ${job._notificationId} not found`);
       }
 
+      const workflow =
+        (await this.notificationTemplateRepository.findById(job._templateId, job._environmentId)) ?? undefined;
+
       if (isSubscribersScheduleEnabled) {
         const schedule = await this.getSubscriberSchedule.execute(
           GetSubscriberScheduleCommand.create({
@@ -177,7 +180,7 @@ export class RunJob {
 
         if (
           isOutsideSubscriberSchedule &&
-          (await this.shouldExtendToSubscriberSchedule(job, notification.critical ?? false))
+          (await this.shouldExtendToSubscriberSchedule(job, notification.critical ?? false, workflow))
         ) {
           this.logger.info(
             {
@@ -227,7 +230,7 @@ export class RunJob {
           );
 
           // Update workflow run delivery lifecycle after schedule-based cancellation
-          await this.conditionallyUpdateDeliveryLifecycle(job, WorkflowRunStatusEnum.COMPLETED);
+          await this.conditionallyUpdateDeliveryLifecycle(job, WorkflowRunStatusEnum.COMPLETED, workflow);
 
           return;
         }
@@ -271,6 +274,7 @@ export class RunJob {
           severity: notification.severity,
           statelessPreferences: job.preferences,
           contextKeys: job.contextKeys,
+          workflow,
         })
       );
 
@@ -285,7 +289,7 @@ export class RunJob {
         });
 
         // Update workflow run delivery lifecycle after successful step completion
-        await this.conditionallyUpdateDeliveryLifecycle(job, WorkflowRunStatusEnum.PROCESSING);
+        await this.conditionallyUpdateDeliveryLifecycle(job, WorkflowRunStatusEnum.PROCESSING, workflow);
       } else if (sendMessageResult.status === 'failed') {
         await this.jobRepository.update(
           {
@@ -307,7 +311,7 @@ export class RunJob {
         });
 
         // Update workflow run delivery lifecycle after step failure
-        await this.conditionallyUpdateDeliveryLifecycle(job, WorkflowRunStatusEnum.COMPLETED);
+        await this.conditionallyUpdateDeliveryLifecycle(job, WorkflowRunStatusEnum.COMPLETED, workflow);
 
         if (shouldHaltOnStepFailure(job)) {
           shouldQueueNextJob = false;
@@ -330,7 +334,7 @@ export class RunJob {
         });
 
         // Update workflow run delivery lifecycle after step skip/cancellation
-        await this.conditionallyUpdateDeliveryLifecycle(job, WorkflowRunStatusEnum.PROCESSING);
+        await this.conditionallyUpdateDeliveryLifecycle(job, WorkflowRunStatusEnum.PROCESSING, workflow);
       }
     } catch (caughtError: unknown) {
       error = caughtError as Error;
@@ -446,7 +450,7 @@ export class RunJob {
           );
 
           // Update workflow run delivery lifecycle after step skip
-          await this.conditionallyUpdateDeliveryLifecycle(nextJob, WorkflowRunStatusEnum.PROCESSING);
+          await this.conditionallyUpdateDeliveryLifecycle(nextJob, WorkflowRunStatusEnum.PROCESSING, undefined);
 
           currentJob = nextJob; // if skipped, continue to the next job
         } else {
@@ -654,7 +658,8 @@ export class RunJob {
    */
   private async conditionallyUpdateDeliveryLifecycle(
     job: JobEntity,
-    workflowStatus: WorkflowRunStatusEnum
+    workflowStatus: WorkflowRunStatusEnum,
+    workflow?: NotificationTemplateEntity
   ): Promise<void> {
     this.logger.debug({ nv: { job } }, 'Conditionally updating delivery lifecycle');
 
@@ -668,19 +673,21 @@ export class RunJob {
       return;
     }
 
-    const workflow: SelectedWorkflowFields | null = await this.notificationTemplateRepository.findOne(
-      {
-        _id: job._templateId,
-        _environmentId: job._environmentId,
-      },
-      SELECTED_WORKFLOW_FIELDS_PROJECTION
-    );
+    const workflowWithSteps: SelectedWorkflowFields | null =
+      workflow ??
+      (await this.notificationTemplateRepository.findOne(
+        {
+          _id: job._templateId,
+          _environmentId: job._environmentId,
+        },
+        SELECTED_WORKFLOW_FIELDS_PROJECTION
+      ));
 
-    if (!workflow || !workflow.steps) {
+    if (!workflowWithSteps || !workflowWithSteps.steps) {
       return;
     }
 
-    const isLastStep = await this.isLastStepInWorkflow(job, workflow);
+    const isLastStep = await this.isLastStepInWorkflow(job, workflowWithSteps);
     if (isLastStep) {
       this.logger.trace(
         { nv: { jobId: job._id, stepId: job.step?._id } },
@@ -689,7 +696,7 @@ export class RunJob {
       return;
     }
 
-    const hasActionSteps = await this.hasRemainingActionSteps(job, workflow);
+    const hasActionSteps = await this.hasRemainingActionSteps(job, workflowWithSteps);
 
     if (hasActionSteps) {
       this.logger.trace(
@@ -724,7 +731,11 @@ export class RunJob {
     return false;
   }
 
-  private async shouldExtendToSubscriberSchedule(job: JobEntity, critical: boolean): Promise<boolean> {
+  private async shouldExtendToSubscriberSchedule(
+    job: JobEntity,
+    critical: boolean,
+    workflow?: NotificationTemplateEntity
+  ): Promise<boolean> {
     // should only extend to schedule for delay and digest when the workflow is not critical
     if ((job.type === StepTypeEnum.DELAY || job.type === StepTypeEnum.DIGEST) && !critical) {
       const bridgeResponse = await this.executeBridgeJob.execute(
@@ -736,6 +747,7 @@ export class RunJob {
           jobId: job._id,
           job: job,
           variables: {},
+          workflow,
         })
       );
       const extendToSchedule = bridgeResponse?.outputs?.extendToSchedule as boolean | undefined;
diff --git a/apps/worker/src/app/workflow/usecases/send-message/send-message.command.ts b/apps/worker/src/app/workflow/usecases/send-message/send-message.command.ts
@@ -1,5 +1,5 @@
 import { EnvironmentWithUserCommand } from '@novu/application-generic';
-import type { JobEntity, NotificationStepEntity } from '@novu/dal';
+import type { JobEntity, NotificationStepEntity, NotificationTemplateEntity } from '@novu/dal';
 import type { SeverityLevelEnum, TriggerOverrides, WorkflowPreferences } from '@novu/shared';
 import { IsDefined, IsOptional, IsString } from 'class-validator';
 
@@ -53,4 +53,7 @@ export class SendMessageCommand extends EnvironmentWithUserCommand {
 
   @IsOptional()
   contextKeys?: string[];
+
+  @IsOptional()
+  workflow?: NotificationTemplateEntity;
 }
diff --git a/apps/worker/src/app/workflow/usecases/send-message/send-message.usecase.ts b/apps/worker/src/app/workflow/usecases/send-message/send-message.usecase.ts
@@ -102,6 +102,7 @@ export class SendMessage {
       bridgeResponse = await this.executeBridgeJob.execute({
         ...command,
         variables,
+        workflow: command.workflow,
       });
     }
     const isBridgeSkipped = bridgeResponse?.options?.skip;
@@ -329,10 +330,12 @@ export class SendMessage {
       return { result: true };
     }
 
-    const workflow = await this.getWorkflow({
-      _id: job._templateId,
-      environmentId: job._environmentId,
-    });
+    const workflow =
+      command.workflow ??
+      (await this.getWorkflow({
+        _id: job._templateId,
+        environmentId: job._environmentId,
+      }));
 
     const subscriber = await this.getSubscriberBySubscriberId({
       _environmentId: job._environmentId,
diff --git a/libs/application-generic/src/services/socket-worker/socket-worker.service.ts b/libs/application-generic/src/services/socket-worker/socket-worker.service.ts
@@ -131,7 +131,9 @@ export class SocketWorkerService {
           Authorization: `Bearer ${this.socketWorkerApiKey}`,
         },
         responseType: 'json',
-        timeout: 5000, // 5 second timeout
+        http2: true,
+        dnsCache: true,
+        timeout: 3000,
         retry: {
           limit: 2,
           methods: ['POST'],
