feat(echo): add skip support

apps/api/src/app/events/e2e/echo-trigger.e2e-ee.ts

@@ -1,8 +1,8 @@
 import axios from 'axios';
 import { expect } from 'chai';
 import { UserSession, SubscribersService } from '@novu/testing';
-import { MessageRepository, SubscriberEntity, NotificationTemplateRepository } from '@novu/dal';
-import { StepTypeEnum } from '@novu/shared';
+import { MessageRepository, SubscriberEntity, NotificationTemplateRepository, JobRepository } from '@novu/dal';
+import { JobStatusEnum, StepTypeEnum } from '@novu/shared';
 import { echoServer } from '../../../../e2e/echo.server';
 
 const eventTriggerPath = '/v1/events/trigger';
@@ -11,6 +11,7 @@ describe('Echo Trigger ', async () => {
   let session: UserSession;
   const messageRepository = new MessageRepository();
   const workflowsRepository = new NotificationTemplateRepository();
+  const jobRepository = new JobRepository();
   let subscriber: SubscriberEntity;
   let subscriberService: SubscribersService;
 
@@ -99,4 +100,98 @@ describe('Echo Trigger ', async () => {
     expect(messagesAfter.length).to.be.eq(1);
     expect(messagesAfter[0].subject).to.include('This is an email subject TEST');
   });
+
+  it('should skip step', async () => {
+    const workflowId = 'hello-world-2';
+    await echoServer.echo.workflow(
+      workflowId,
+      async ({ step, payload }) => {
+        await step.email(
+          'send-email',
+          async (inputs) => {
+            return {
+              subject: 'This is an email subject ' + inputs.name,
+              body: 'Body result ' + payload.name,
+            };
+          },
+          {
+            inputSchema: {
+              type: 'object',
+              properties: {
+                name: { type: 'string', default: 'TEST' },
+              },
+            } as const,
+            skip: () => true,
+          }
+        );
+      },
+      {
+        payloadSchema: {
+          type: 'object',
+          properties: {
+            name: { type: 'string', default: 'default_name' },
+          },
+          required: [],
+          additionalProperties: false,
+        } as const,
+      }
+    );
+
+    await syncWorkflow(session);
+
+    const workflow = await workflowsRepository.findByTriggerIdentifier(session.environment._id, workflowId);
+
+    expect(workflow).to.be.ok;
+    if (!workflow) throw new Error('Workflow not found');
+
+    await triggerEvent(session, workflowId, subscriber);
+    await session.awaitRunningJobs(workflow._id);
+
+    const executedMessage = await messageRepository.find({
+      _environmentId: session.environment._id,
+      _subscriberId: subscriber._id,
+      channel: StepTypeEnum.EMAIL,
+    });
+
+    expect(executedMessage.length).to.be.eq(0);
+
+    const executedMessage2 = await jobRepository.find({
+      _environmentId: session.environment._id,
+      _subscriberId: subscriber._id,
+      type: StepTypeEnum.EMAIL,
+    });
+
+    expect(executedMessage2.length).to.be.eq(1);
+    expect(executedMessage2[0].status).to.be.eq(JobStatusEnum.CANCELED);
+  });
 });
+
+async function syncWorkflow(session) {
+  const resultDiscover = await axios.get(echoServer.serverPath + '/echo?action=discover');
+
+  await session.testAgent.post(`/v1/echo/sync`).send({
+    chimeraUrl: echoServer.serverPath + '/echo',
+    workflows: resultDiscover.data.workflows,
+  });
+}
+
+async function triggerEvent(session, workflowId: string, subscriber) {
+  await axios.post(
+    `${session.serverUrl}${eventTriggerPath}`,
+    {
+      name: workflowId,
+      to: {
+        subscriberId: subscriber.subscriberId,
+        email: 'test@subscriber.com',
+      },
+      payload: {
+        name: 'test_name',
+      },
+    },
+    {
+      headers: {
+        authorization: `ApiKey ${session.apiKey}`,
+      },
+    }
+  );
+}

apps/api/src/app/integrations/usecases/index.ts

@@ -3,6 +3,7 @@ import {
   GetDecryptedIntegrations,
   CalculateLimitNovuIntegration,
   ConditionsFilter,
+  NormalizeVariables,
 } from '@novu/application-generic';
 
 import { GetWebhookSupportStatus } from './get-webhook-support-status/get-webhook-support-status.usecase';
@@ -33,4 +34,5 @@ export const USE_CASES = [
   CalculateLimitNovuIntegration,
   SetIntegrationAsPrimary,
   CreateNovuIntegrations,
+  NormalizeVariables,
 ];

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

@@ -30,6 +30,8 @@ import {
   requireInject,
   StandardQueueService,
   ExecuteOutput,
+  NormalizeVariablesCommand,
+  NormalizeVariables,
 } from '@novu/application-generic';
 
 export enum BackoffStrategiesEnum {
@@ -54,6 +56,7 @@ export class AddJob {
     private calculateDelayService: CalculateDelayService,
     @Inject(forwardRef(() => ConditionsFilter))
     private conditionsFilter: ConditionsFilter,
+    private normalizeVariablesUsecase: NormalizeVariables,
     private moduleRef: ModuleRef
   ) {
     this.resonateUsecase = requireInject('resonate', this.moduleRef);
@@ -80,6 +83,17 @@ export class AddJob {
     let filtered = false;
     let filterVariables: IFilterVariables | undefined;
     if ([StepTypeEnum.DELAY, StepTypeEnum.DIGEST].includes(job.type as StepTypeEnum)) {
+      const variables = await this.normalizeVariablesUsecase.execute(
+        NormalizeVariablesCommand.create({
+          filters: command.job.step.filters || [],
+          environmentId: command.environmentId,
+          organizationId: command.organizationId,
+          userId: command.userId,
+          step: job.step,
+          job: job,
+        })
+      );
+
       const shouldRun = await this.conditionsFilter.filter(
         ConditionsFilterCommand.create({
           filters: job.step.filters || [],
@@ -88,6 +102,7 @@ export class AddJob {
           userId: command.userId,
           step: job.step,
           job,
+          variables,
         })
       );
 

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

@@ -71,7 +71,7 @@ export class RunJob {
 
       await this.storageHelperService.getAttachments(job.payload?.attachments);
 
-      await this.sendMessage.execute(
+      const sendMessageResult = await this.sendMessage.execute(
         SendMessageCommand.create({
           identifier: job.identifier,
           payload: job.payload ?? {},
@@ -92,7 +92,9 @@ export class RunJob {
         })
       );
 
-      await this.jobRepository.updateStatus(job._environmentId, job._id, JobStatusEnum.COMPLETED);
+      if (sendMessageResult.status === 'success') {
+        await this.jobRepository.updateStatus(job._environmentId, job._id, JobStatusEnum.COMPLETED);
+      }
     } catch (error: any) {
       Logger.error({ error }, `Running job ${job._id} has thrown an error`, LOG_CONTEXT);
       if (job.step.shouldStopOnFail || this.shouldBackoff(error)) {