feat(echo): add skip support

apps/api/e2e/echo.server.ts

@@ -19,7 +19,11 @@ class EchoServer {
     this.server.use(express.json());
     this.server.use(serve({ client: this.echo }));
 
-    this.app = await this.server.listen(this.port);
+    await new Promise<void>((resolve) => {
+      this.app = this.server.listen(this.port, () => {
+        resolve();
+      });
+    });
   }
 
   async stop() {

apps/api/migrations/encrypt-api-keys/encrypt-api-keys-migration.spec.ts

@@ -1,6 +1,6 @@
 import { expect } from 'chai';
-import { faker } from '@faker-js/faker';
 import { createHash } from 'crypto';
+import { faker } from '@faker-js/faker';
 
 import { UserSession } from '@novu/testing';
 import { ChannelTypeEnum } from '@novu/stateless';

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

@@ -5,35 +5,22 @@ import {
   MessageRepository,
   SubscriberEntity,
   NotificationTemplateRepository,
+  JobRepository,
   ExecutionDetailsRepository,
 } from '@novu/dal';
-import { ExecutionDetailsStatusEnum, MarkMessagesAsEnum, StepTypeEnum } from '@novu/shared';
+import { ExecutionDetailsStatusEnum, JobStatusEnum, MarkMessagesAsEnum, StepTypeEnum } from '@novu/shared';
 import { echoServer } from '../../../../e2e/echo.server';
 
+const eventTriggerPath = '/v1/events/trigger';
+
 describe('Echo Trigger ', async () => {
   let session: UserSession;
   const messageRepository = new MessageRepository();
   const workflowsRepository = new NotificationTemplateRepository();
-  const executionDetailsRepository = new ExecutionDetailsRepository();
-
+  const jobRepository = new JobRepository();
   let subscriber: SubscriberEntity;
   let subscriberService: SubscribersService;
-
-  const triggerEvent = async (workflowId: string, payload): Promise<void> => {
-    await axios.post(
-      `${session.serverUrl}/v1/events/trigger`,
-      {
-        name: workflowId,
-        to: [subscriber.subscriberId],
-        payload,
-      },
-      {
-        headers: {
-          authorization: `ApiKey ${session.apiKey}`,
-        },
-      }
-    );
-  };
+  const executionDetailsRepository = new ExecutionDetailsRepository();
 
   beforeEach(async () => {
     session = new UserSession();
@@ -86,8 +73,7 @@ describe('Echo Trigger ', async () => {
       throw new Error('Workflow not found');
     }
 
-    await triggerEvent(workflowId, { name: 'test_name' });
-
+    await triggerEvent(session, workflowId, subscriber, { name: 'test_name' });
     await session.awaitRunningJobs(workflow._id);
 
     const messagesAfter = await messageRepository.find({
@@ -100,6 +86,141 @@ describe('Echo Trigger ', async () => {
     expect(messagesAfter[0].subject).to.include('This is an email subject TEST');
   });
 
+  it('should skip step', async () => {
+    // should skip static value
+    const workflowIdSkipByStatic = 'skip-by-static-value-workflow';
+    await echoServer.echo.workflow(
+      workflowIdSkipByStatic,
+      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 workflowByStatic = await workflowsRepository.findByTriggerIdentifier(
+      session.environment._id,
+      workflowIdSkipByStatic
+    );
+
+    expect(workflowByStatic).to.be.ok;
+    if (!workflowByStatic) throw new Error('Workflow not found');
+
+    await triggerEvent(session, workflowIdSkipByStatic, subscriber);
+    await session.awaitRunningJobs(workflowByStatic._id);
+
+    const executedMessageByStatic = await messageRepository.find({
+      _environmentId: session.environment._id,
+      _subscriberId: subscriber._id,
+      channel: StepTypeEnum.EMAIL,
+    });
+
+    expect(executedMessageByStatic.length).to.be.eq(0);
+
+    const cancelledJobByStatic = await jobRepository.find({
+      _environmentId: session.environment._id,
+      _subscriberId: subscriber._id,
+      type: StepTypeEnum.EMAIL,
+    });
+
+    expect(cancelledJobByStatic.length).to.be.eq(1);
+    expect(cancelledJobByStatic[0].status).to.be.eq(JobStatusEnum.CANCELED);
+
+    // should skip by variable default value
+    const workflowIdSkipByVariable = 'skip-by-variable-default-value';
+    await echoServer.echo.workflow(
+      workflowIdSkipByVariable,
+      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' },
+                shouldSkipVar: { type: 'boolean', default: true },
+              },
+            } as const,
+            skip: (inputs) => inputs.shouldSkipVar,
+          }
+        );
+      },
+      {
+        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,
+      workflowIdSkipByVariable
+    );
+
+    expect(workflow).to.be.ok;
+    if (!workflow) throw new Error('Workflow not found');
+
+    await triggerEvent(session, workflowIdSkipByVariable, 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 cancelledJobByVariable = await jobRepository.find({
+      _environmentId: session.environment._id,
+      _subscriberId: subscriber._id,
+      type: StepTypeEnum.EMAIL,
+    });
+
+    expect(cancelledJobByVariable.length).to.be.eq(2);
+    expect(cancelledJobByVariable[1].status).to.be.eq(JobStatusEnum.CANCELED);
+  });
+
   it('should have execution detail errors for invalid trigger payload', async () => {
     const workflowId = 'missing-payload-var';
     await echoServer.echo.workflow(
@@ -133,7 +254,7 @@ describe('Echo Trigger ', async () => {
       throw new Error('Workflow not found');
     }
 
-    await triggerEvent(workflowId, {});
+    await triggerEvent(session, workflowId, subscriber, {});
 
     await session.awaitRunningJobs(workflow._id);
 
@@ -157,7 +278,7 @@ describe('Echo Trigger ', async () => {
 
     await executionDetailsRepository.delete({ _environmentId: session.environment._id });
 
-    await triggerEvent(workflowId, { name: 4 });
+    await triggerEvent(session, workflowId, subscriber, { name: 4 });
     await session.awaitRunningJobs(workflow._id);
 
     const executionDetailsInvalidType = await executionDetailsRepository.find({
@@ -219,7 +340,7 @@ describe('Echo Trigger ', async () => {
       throw new Error('Workflow not found');
     }
 
-    await triggerEvent(workflowId, {});
+    await triggerEvent(session, workflowId, subscriber, {});
 
     await session.awaitRunningJobs(workflow._id);
 
@@ -300,8 +421,8 @@ describe('Echo Trigger ', async () => {
       throw new Error('Workflow not found');
     }
 
-    await triggerEvent(workflowId, { name: 'John' });
-    await triggerEvent(workflowId, { name: 'Bela' });
+    await triggerEvent(session, workflowId, subscriber, { name: 'John' });
+    await triggerEvent(session, workflowId, subscriber, { name: 'Bela' });
 
     await session.awaitRunningJobs(workflow?._id, false, 0);
 
@@ -316,6 +437,38 @@ describe('Echo Trigger ', async () => {
   });
 });
 
+async function syncWorkflow(session) {
+  const resultDiscover = await axios.get(echoServer.serverPath + '/echo?action=discover');
+
+  await session.testAgent.post(`/v1/echo/sync`).send({
+    bridgeUrl: echoServer.serverPath + '/echo',
+    workflows: resultDiscover.data.workflows,
+  });
+}
+
+async function triggerEvent(session, workflowId: string, subscriber, payload?: any) {
+  const defaultPayload = {
+    name: 'test_name',
+  };
+
+  await axios.post(
+    `${session.serverUrl}${eventTriggerPath}`,
+    {
+      name: workflowId,
+      to: {
+        subscriberId: subscriber.subscriberId,
+        email: 'test@subscriber.com',
+      },
+      payload: payload ?? defaultPayload,
+    },
+    {
+      headers: {
+        authorization: `ApiKey ${session.apiKey}`,
+      },
+    }
+  );
+}
+
 async function discoverAndSyncEcho(session: UserSession) {
   const resultDiscover = await axios.get(echoServer.serverPath + '/echo?action=discover');
 

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/api/src/app/subscribers/unit/update-subscriber-channel.spec.ts

@@ -1,10 +1,12 @@
-import { IntegrationRepository, SubscriberRepository } from '@novu/dal';
-import { SubscribersService, UserSession } from '@novu/testing';
 import { Test } from '@nestjs/testing';
 import { expect } from 'chai';
-import { ChannelTypeEnum, ChatProviderIdEnum, PushProviderIdEnum } from '@novu/shared';
 import { faker } from '@faker-js/faker';
+
+import { IntegrationRepository, SubscriberRepository } from '@novu/dal';
+import { SubscribersService, UserSession } from '@novu/testing';
+import { ChannelTypeEnum, ChatProviderIdEnum, PushProviderIdEnum } from '@novu/shared';
 import { OAuthHandlerEnum, UpdateSubscriberChannel, UpdateSubscriberChannelCommand } from '@novu/application-generic';
+
 import { SharedModule } from '../../shared/shared.module';
 
 describe('Update Subscriber channel credentials', function () {
@@ -418,8 +420,8 @@ describe('Update Subscriber channel credentials', function () {
   it('should update deviceTokens without duplication on channel creation (addChannelToSubscriber)', async function () {
     const subscriberId = SubscriberRepository.createObjectId();
     const test = await subscriberRepository.create({
-      lastName: faker.name.lastName(),
       firstName: faker.name.firstName(),
+      lastName: faker.name.lastName(),
       email: faker.internet.email(),
       phone: faker.phone.phoneNumber(),
       _environmentId: session.environment._id,