feat(echo): add skip support

apps/api/e2e/utils.ts

@@ -0,0 +1,122 @@
+//# cSpell:disable
+
+export const fakerUtil = {
+  phoneNumber: () =>
+    `(${Math.floor(100 * Math.random()) + 100}) ${Math.floor(100 * Math.random()) + 100}-${
+      Math.floor(10000 * Math.random()) + 10000
+    }`,
+  firstName: () => `John-${new Date().toISOString()}`,
+  lastName: () => `Smith-${new Date().toISOString()}`,
+  email: () => `Doe-Smith-${new Date().toISOString()}@gmail.com`,
+  words: generateWords(),
+  url: generateUrl(),
+  jobTitle: jobTitle(),
+};
+
+function generateWords() {
+  return (length: number | undefined) => {
+    return Array.from({ length: length ?? 3 }, () => loremWords[Math.floor(Math.random() * loremWords.length)]).join(
+      ' '
+    );
+  };
+}
+
+function jobTitle() {
+  return () => {
+    const prefixes = ['Senior', 'Lead', 'Junior', 'Assistant', 'Chief', 'Head'];
+    const roles = ['Engineer', 'Developer', 'Manager', 'Analyst', 'Consultant', 'Specialist'];
+    const domains = ['Software', 'Hardware', 'Network', 'Security', 'Data', 'Cloud'];
+
+    const prefix = prefixes[Math.floor(Math.random() * prefixes.length)];
+    const role = roles[Math.floor(Math.random() * roles.length)];
+    const domain = domains[Math.floor(Math.random() * domains.length)];
+
+    return `${prefix} ${domain} ${role}`;
+  };
+}
+
+function generateUrl() {
+  return () => {
+    const protocols = ['http', 'https'];
+    const domains = ['example.com', 'test.com', 'site.com'];
+    const paths = ['path', 'to', 'resource'];
+
+    const protocol = protocols[Math.floor(Math.random() * protocols.length)];
+    const domain = domains[Math.floor(Math.random() * domains.length)];
+    const path = Array.from({ length: 3 }, () => paths[Math.floor(Math.random() * paths.length)]).join('/');
+
+    return `${protocol}://${domain}/${path}`;
+  };
+}
+
+const loremWords = [
+  'Lorem',
+  'ipsum',
+  'dolor',
+  'sit',
+  'amet',
+  'consectetur',
+  'adipiscing',
+  'elit',
+  'sed',
+  'do',
+  'eiusmod',
+  'tempor',
+  'incididunt',
+  'ut',
+  'labore',
+  'et',
+  'dolore',
+  'magna',
+  'aliqua',
+  'Ut',
+  'enim',
+  'ad',
+  'minim',
+  'veniam',
+  'quis',
+  'nostrud',
+  'exercitation',
+  'ullamco',
+  'laboris',
+  'nisi',
+  'ut',
+  'aliquip',
+  'ex',
+  'ea',
+  'commodo',
+  'consequat',
+  'Duis',
+  'aute',
+  'irure',
+  'dolor',
+  'in',
+  'reprehenderit',
+  'in',
+  'voluptate',
+  'velit',
+  'esse',
+  'cillum',
+  'dolore',
+  'eu',
+  'fugiat',
+  'nulla',
+  'pariatur',
+  'Excepteur',
+  'sint',
+  'occaecat',
+  'cupidatat',
+  'non',
+  'proident',
+  'sunt',
+  'in',
+  'culpa',
+  'qui',
+  'officia',
+  'deserunt',
+  'mollit',
+  'anim',
+  'id',
+  'est',
+  'laborum',
+];

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

@@ -1,5 +1,4 @@
 import { expect } from 'chai';
-import { faker } from '@faker-js/faker';
 import { createHash } from 'crypto';
 
 import { UserSession } from '@novu/testing';
@@ -8,6 +7,7 @@ import { EnvironmentRepository } from '@novu/dal';
 import { decryptApiKey } from '@novu/application-generic';
 
 import { encryptApiKeysMigration } from './encrypt-api-keys-migration';
+import { fakerUtil } from '../../e2e/utils';
 
 async function pruneIntegration({ environmentRepository }: { environmentRepository: EnvironmentRepository }) {
   const old = await environmentRepository.find({});
@@ -32,7 +32,7 @@ describe('Encrypt Old api keys', function () {
     for (let i = 0; i < 2; i++) {
       await environmentRepository.create({
         identifier: 'identifier' + i,
-        name: faker.name.jobTitle(),
+        name: fakerUtil.jobTitle(),
         _organizationId: session.organization._id,
         apiKeys: [
           {
@@ -85,7 +85,7 @@ describe('Encrypt Old api keys', function () {
     for (let i = 0; i < 2; i++) {
       await environmentRepository.create({
         identifier: 'identifier' + i,
-        name: faker.name.jobTitle(),
+        name: fakerUtil.jobTitle(),
         _organizationId: session.organization._id,
         apiKeys: [
           {

apps/api/migrations/normalize-message-template-cta-action/normalize-message-template-cta-action-migration.spec.ts

@@ -1,12 +1,12 @@
 import { expect } from 'chai';
-import { faker } from '@faker-js/faker';
 
 import { UserSession } from '@novu/testing';
 import { MessageRepository, MessageTemplateRepository } from '@novu/dal';
 import { StepTypeEnum } from '@novu/shared';
 
 import { normalizeMessageTemplateCtaAction } from './normalize-message-template-cta-action-migration';
 import { normalizeMessageCtaAction } from './normalize-message-cta-action-migration';
+import { fakerUtil } from '../../e2e/utils';
 
 describe('Normalize cta action', function () {
   let session: UserSession;
@@ -29,8 +29,8 @@ describe('Normalize cta action', function () {
         action: {
           buttons: [
             {
-              title: faker.lorem.words(3),
-              url: faker.internet.url(),
+              title: fakerUtil.words(3),
+              url: fakerUtil.url(),
             },
           ],
         },
@@ -70,8 +70,8 @@ describe('Normalize cta action', function () {
         action: {
           buttons: [
             {
-              title: faker.lorem.words(3),
-              url: faker.internet.url(),
+              title: fakerUtil.words(3),
+              url: fakerUtil.url(),
             },
           ],
         },

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,70 @@ describe('Echo Trigger ', async () => {
     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);
+  });
+
   it('should have execution detail errors for invalid trigger payload', async () => {
     const workflowId = 'missing-payload-var';
     await echoServer.echo.workflow(
@@ -133,7 +183,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 +207,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 +269,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 +350,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,11 +366,43 @@ 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({
+    url: 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');
 
   await session.testAgent.post(`/v1/echo/sync`).send({
-    chimeraUrl: echoServer.serverPath + '/echo',
+    url: echoServer.serverPath + '/echo',
     workflows: resultDiscover.data.workflows,
   });
 }