Execute variables in action input (#7715)

- send context from all previous steps rather than unique payload
- wrap input data in settings into input field
- add email into send email action settings
- update output shape

<img width="553" alt="Capture d’écran 2024-10-15 à 15 21 32"
src="https://github.com/user-attachments/assets/0f5ed004-0d6e-4a59-969b-a5710f3f3985">

<img width="761" alt="Capture d’écran 2024-10-15 à 15 20 09"
src="https://github.com/user-attachments/assets/ac140846-c383-483b-968a-eab469b76785">

packages/twenty-server/src/modules/mail-sender/workflow-actions/send-email.workflow-action.ts

@@ -1,40 +1,37 @@
 import { Injectable, Logger } from '@nestjs/common';
 
-import { z } from 'zod';
-import Handlebars from 'handlebars';
-import { JSDOM } from 'jsdom';
 import DOMPurify from 'dompurify';
+import { JSDOM } from 'jsdom';
+import { z } from 'zod';
+
+import { WorkflowAction } from 'src/modules/workflow/workflow-executor/interfaces/workflow-action.interface';
 
-import { WorkflowActionResult } from 'src/modules/workflow/workflow-executor/types/workflow-action-result.type';
-import { WorkflowSendEmailStep } from 'src/modules/workflow/workflow-executor/types/workflow-action.type';
-import { EnvironmentService } from 'src/engine/core-modules/environment/environment.service';
-import { EmailService } from 'src/engine/core-modules/email/email.service';
-import { ConnectedAccountWorkspaceEntity } from 'src/modules/connected-account/standard-objects/connected-account.workspace-entity';
-import {
-  WorkflowStepExecutorException,
-  WorkflowStepExecutorExceptionCode,
-} from 'src/modules/workflow/workflow-executor/exceptions/workflow-step-executor.exception';
 import { ScopedWorkspaceContextFactory } from 'src/engine/twenty-orm/factories/scoped-workspace-context.factory';
+import { TwentyORMGlobalManager } from 'src/engine/twenty-orm/twenty-orm-global.manager';
+import { ConnectedAccountWorkspaceEntity } from 'src/modules/connected-account/standard-objects/connected-account.workspace-entity';
 import {
   MailSenderException,
   MailSenderExceptionCode,
 } from 'src/modules/mail-sender/exceptions/mail-sender.exception';
 import { GmailClientProvider } from 'src/modules/messaging/message-import-manager/drivers/gmail/providers/gmail-client.provider';
-import { TwentyORMGlobalManager } from 'src/engine/twenty-orm/twenty-orm-global.manager';
+import {
+  WorkflowStepExecutorException,
+  WorkflowStepExecutorExceptionCode,
+} from 'src/modules/workflow/workflow-executor/exceptions/workflow-step-executor.exception';
+import { WorkflowActionResult } from 'src/modules/workflow/workflow-executor/types/workflow-action-result.type';
+import { WorkflowSendEmailStepInput } from 'src/modules/workflow/workflow-executor/types/workflow-step-settings.type';
 import { isDefined } from 'src/utils/is-defined';
 
 @Injectable()
-export class SendEmailWorkflowAction {
+export class SendEmailWorkflowAction implements WorkflowAction {
   private readonly logger = new Logger(SendEmailWorkflowAction.name);
   constructor(
-    private readonly environmentService: EnvironmentService,
-    private readonly emailService: EmailService,
     private readonly gmailClientProvider: GmailClientProvider,
     private readonly scopedWorkspaceContextFactory: ScopedWorkspaceContextFactory,
     private readonly twentyORMGlobalManager: TwentyORMGlobalManager,
   ) {}
 
-  private async getEmailClient(step: WorkflowSendEmailStep) {
+  private async getEmailClient(connectedAccountId: string) {
     const { workspaceId } = this.scopedWorkspaceContextFactory.create();
 
     if (!workspaceId) {
@@ -50,12 +47,12 @@ export class SendEmailWorkflowAction {
         'connectedAccount',
       );
     const connectedAccount = await connectedAccountRepository.findOneBy({
-      id: step.settings.connectedAccountId,
+      id: connectedAccountId,
     });
 
     if (!isDefined(connectedAccount)) {
       throw new MailSenderException(
-        `Connected Account '${step.settings.connectedAccountId}' not found`,
+        `Connected Account '${connectedAccountId}' not found`,
         MailSenderExceptionCode.CONNECTED_ACCOUNT_NOT_FOUND,
       );
     }
@@ -71,39 +68,32 @@ export class SendEmailWorkflowAction {
     }
   }
 
-  async execute({
-    step,
-    payload,
-  }: {
-    step: WorkflowSendEmailStep;
-    payload: {
-      email: string;
-      [key: string]: string;
-    };
-  }): Promise<WorkflowActionResult> {
-    const emailProvider = await this.getEmailClient(step);
+  async execute(
+    workflowStepInput: WorkflowSendEmailStepInput,
+  ): Promise<WorkflowActionResult> {
+    const emailProvider = await this.getEmailClient(
+      workflowStepInput.connectedAccountId,
+    );
+    const { email, body, subject } = workflowStepInput;
 
     try {
       const emailSchema = z.string().trim().email('Invalid email');
 
-      const result = emailSchema.safeParse(payload.email);
+      const result = emailSchema.safeParse(email);
 
       if (!result.success) {
-        this.logger.warn(`Email '${payload.email}' invalid`);
+        this.logger.warn(`Email '${email}' invalid`);
 
         return { result: { success: false } };
       }
 
-      const body = Handlebars.compile(step.settings.body)(payload);
-      const subject = Handlebars.compile(step.settings.subject)(payload);
-
       const window = new JSDOM('').window;
       const purify = DOMPurify(window);
       const safeBody = purify.sanitize(body || '');
       const safeSubject = purify.sanitize(subject || '');
 
       const message = [
-        `To: ${payload.email}`,
+        `To: ${email}`,
         `Subject: ${safeSubject || ''}`,
         'MIME-Version: 1.0',
         'Content-Type: text/plain; charset="UTF-8"',

packages/twenty-server/src/modules/serverless/workflow-actions/code.workflow-action.ts

@@ -1,28 +1,26 @@
 import { Injectable } from '@nestjs/common';
 
+import { WorkflowAction } from 'src/modules/workflow/workflow-executor/interfaces/workflow-action.interface';
+
 import { ServerlessFunctionService } from 'src/engine/metadata-modules/serverless-function/serverless-function.service';
 import { ScopedWorkspaceContextFactory } from 'src/engine/twenty-orm/factories/scoped-workspace-context.factory';
-import { WorkflowActionResult } from 'src/modules/workflow/workflow-executor/types/workflow-action-result.type';
-import { WorkflowCodeStep } from 'src/modules/workflow/workflow-executor/types/workflow-action.type';
 import {
   WorkflowStepExecutorException,
   WorkflowStepExecutorExceptionCode,
 } from 'src/modules/workflow/workflow-executor/exceptions/workflow-step-executor.exception';
+import { WorkflowActionResult } from 'src/modules/workflow/workflow-executor/types/workflow-action-result.type';
+import { WorkflowCodeStepInput } from 'src/modules/workflow/workflow-executor/types/workflow-step-settings.type';
 
 @Injectable()
-export class CodeWorkflowAction {
+export class CodeWorkflowAction implements WorkflowAction {
   constructor(
     private readonly serverlessFunctionService: ServerlessFunctionService,
     private readonly scopedWorkspaceContextFactory: ScopedWorkspaceContextFactory,
   ) {}
 
-  async execute({
-    step,
-    payload,
-  }: {
-    step: WorkflowCodeStep;
-    payload?: object;
-  }): Promise<WorkflowActionResult> {
+  async execute(
+    workflowStepInput: WorkflowCodeStepInput,
+  ): Promise<WorkflowActionResult> {
     const { workspaceId } = this.scopedWorkspaceContextFactory.create();
 
     if (!workspaceId) {
@@ -34,9 +32,9 @@ export class CodeWorkflowAction {
 
     const result =
       await this.serverlessFunctionService.executeOneServerlessFunction(
-        step.settings.serverlessFunctionId,
+        workflowStepInput.serverlessFunctionId,
         workspaceId,
-        payload || {},
+        {}, // TODO: input will be dynamically calculated from function input
       );
 
     if (result.error) {

packages/twenty-server/src/modules/workflow/common/standard-objects/workflow-run.workspace-entity.ts

@@ -32,17 +32,21 @@ export enum WorkflowRunStatus {
   FAILED = 'FAILED',
 }
 
-export type WorkflowRunOutput = {
-  steps: {
-    id: string;
-    name: string;
-    type: string;
+type StepRunOutput = {
+  id: string;
+  name: string;
+  type: string;
+  outputs: {
     attemptCount: number;
     result: object | undefined;
     error: string | undefined;
   }[];
 };
 
+export type WorkflowRunOutput = {
+  steps: Record<string, StepRunOutput>;
+};
+
 @WorkspaceEntity({
   standardId: STANDARD_OBJECT_IDS.workflowRun,
   namePlural: 'workflowRuns',

packages/twenty-server/src/modules/workflow/workflow-executor/exceptions/workflow-executor.exception.ts

@@ -8,4 +8,5 @@ export class WorkflowExecutorException extends CustomException {
 
 export enum WorkflowExecutorExceptionCode {
   WORKFLOW_FAILED = 'WORKFLOW_FAILED',
+  VARIABLE_EVALUATION_FAILED = 'VARIABLE_EVALUATION_FAILED',
 }

packages/twenty-server/src/modules/workflow/workflow-executor/interfaces/workflow-action.interface.ts

@@ -1,12 +1,5 @@
 import { WorkflowActionResult } from 'src/modules/workflow/workflow-executor/types/workflow-action-result.type';
-import { WorkflowStep } from 'src/modules/workflow/workflow-executor/types/workflow-action.type';
 
 export interface WorkflowAction {
-  execute({
-    step,
-    payload,
-  }: {
-    step: WorkflowStep;
-    payload?: object;
-  }): Promise<WorkflowActionResult>;
+  execute(workflowStepInput: unknown): Promise<WorkflowActionResult>;
 }

packages/twenty-server/src/modules/workflow/workflow-executor/types/workflow-step-settings.type.ts

@@ -9,12 +9,21 @@ type BaseWorkflowStepSettings = {
   };
 };
 
-export type WorkflowCodeStepSettings = BaseWorkflowStepSettings & {
+export type WorkflowCodeStepInput = {
   serverlessFunctionId: string;
 };
 
-export type WorkflowSendEmailStepSettings = BaseWorkflowStepSettings & {
+export type WorkflowCodeStepSettings = BaseWorkflowStepSettings & {
+  input: WorkflowCodeStepInput;
+};
+
+export type WorkflowSendEmailStepInput = {
   connectedAccountId: string;
+  email: string;
   subject?: string;
   body?: string;
 };
+
+export type WorkflowSendEmailStepSettings = BaseWorkflowStepSettings & {
+  input: WorkflowSendEmailStepInput;
+};

packages/twenty-server/src/modules/workflow/workflow-executor/utils/__tests__/variable-resolver.util.spec.ts

@@ -0,0 +1,81 @@
+import { resolveInput } from 'src/modules/workflow/workflow-executor/utils/variable-resolver.util';
+
+describe('resolveInput', () => {
+  const context = {
+    user: {
+      name: 'John Doe',
+      age: 30,
+    },
+    settings: {
+      theme: 'dark',
+      notifications: true,
+    },
+  };
+
+  it('should return null for null input', () => {
+    expect(resolveInput(null, context)).toBeNull();
+  });
+
+  it('should return undefined for undefined input', () => {
+    expect(resolveInput(undefined, context)).toBeUndefined();
+  });
+
+  it('should resolve a simple string variable', () => {
+    expect(resolveInput('{{user.name}}', context)).toBe('John Doe');
+  });
+
+  it('should resolve multiple variables in a string', () => {
+    expect(
+      resolveInput('Name: {{user.name}}, Age: {{user.age}}', context),
+    ).toBe('Name: John Doe, Age: 30');
+  });
+
+  it('should handle non-existent variables', () => {
+    expect(resolveInput('{{user.email}}', context)).toBe('');
+  });
+
+  it('should resolve variables in an array', () => {
+    const input = ['{{user.name}}', '{{settings.theme}}', 'static'];
+    const expected = ['John Doe', 'dark', 'static'];
+
+    expect(resolveInput(input, context)).toEqual(expected);
+  });
+
+  it('should resolve variables in an object', () => {
+    const input = {
+      name: '{{user.name}}',
+      theme: '{{settings.theme}}',
+      static: 'value',
+    };
+    const expected = {
+      name: 'John Doe',
+      theme: 'dark',
+      static: 'value',
+    };
+
+    expect(resolveInput(input, context)).toEqual(expected);
+  });
+
+  it('should handle nested objects and arrays', () => {
+    const input = {
+      user: {
+        displayName: '{{user.name}}',
+        preferences: ['{{settings.theme}}', '{{settings.notifications}}'],
+      },
+      staticData: [1, 2, 3],
+    };
+    const expected = {
+      user: {
+        displayName: 'John Doe',
+        preferences: ['dark', 'true'],
+      },
+      staticData: [1, 2, 3],
+    };
+
+    expect(resolveInput(input, context)).toEqual(expected);
+  });
+
+  it('should throw an error for invalid expressions', () => {
+    expect(() => resolveInput('{{invalidFunction()}}', context)).toThrow();
+  });
+});

packages/twenty-server/src/modules/workflow/workflow-executor/utils/variable-resolver.util.ts

@@ -0,0 +1,98 @@
+import { isNil, isString } from '@nestjs/common/utils/shared.utils';
+
+import Handlebars from 'handlebars';
+
+import {
+  WorkflowExecutorException,
+  WorkflowExecutorExceptionCode,
+} from 'src/modules/workflow/workflow-executor/exceptions/workflow-executor.exception';
+
+const VARIABLE_PATTERN = RegExp('\\{\\{(.*?)\\}\\}', 'g');
+
+export const resolveInput = (
+  unresolvedInput: unknown,
+  context: Record<string, unknown>,
+): unknown => {
+  if (isNil(unresolvedInput)) {
+    return unresolvedInput;
+  }
+
+  if (isString(unresolvedInput)) {
+    return resolveString(unresolvedInput, context);
+  }
+
+  if (Array.isArray(unresolvedInput)) {
+    return resolveArray(unresolvedInput, context);
+  }
+
+  if (typeof unresolvedInput === 'object' && unresolvedInput !== null) {
+    return resolveObject(unresolvedInput, context);
+  }
+
+  return unresolvedInput;
+};
+
+const resolveArray = (
+  input: unknown[],
+  context: Record<string, unknown>,
+): unknown[] => {
+  const resolvedArray = input;
+
+  for (let i = 0; i < input.length; ++i) {
+    resolvedArray[i] = resolveInput(input[i], context);
+  }
+
+  return resolvedArray;
+};
+
+const resolveObject = (
+  input: object,
+  context: Record<string, unknown>,
+): object => {
+  const resolvedObject = input;
+
+  const entries = Object.entries(resolvedObject);
+
+  for (const [key, value] of entries) {
+    resolvedObject[key] = resolveInput(value, context);
+  }
+
+  return resolvedObject;
+};
+
+const resolveString = (
+  input: string,
+  context: Record<string, unknown>,
+): string => {
+  const matchedTokens = input.match(VARIABLE_PATTERN);
+
+  if (!matchedTokens || matchedTokens.length === 0) {
+    return input;
+  }
+
+  if (matchedTokens.length === 1 && matchedTokens[0] === input) {
+    return evalFromContext(input, context);
+  }
+
+  return input.replace(VARIABLE_PATTERN, (matchedToken, _) => {
+    const processedToken = evalFromContext(matchedToken, context);
+
+    return processedToken;
+  });
+};
+
+const evalFromContext = (
+  input: string,
+  context: Record<string, unknown>,
+): string => {
+  try {
+    const inferredInput = Handlebars.compile(input)(context);
+
+    return inferredInput ?? '';
+  } catch (exception) {
+    throw new WorkflowExecutorException(
+      `Failed to evaluate variable ${input}: ${exception}`,
+      WorkflowExecutorExceptionCode.VARIABLE_EVALUATION_FAILED,
+    );
+  }
+};

packages/twenty-server/src/modules/workflow/workflow-executor/workflow-executor.module.ts

@@ -1,13 +1,13 @@
 import { Module } from '@nestjs/common';
 
-import { WorkflowCommonModule } from 'src/modules/workflow/common/workflow-common.module';
-import { WorkflowExecutorWorkspaceService } from 'src/modules/workflow/workflow-executor/workspace-services/workflow-executor.workspace-service';
-import { WorkflowActionFactory } from 'src/modules/workflow/workflow-executor/factories/workflow-action.factory';
-import { CodeWorkflowAction } from 'src/modules/serverless/workflow-actions/code.workflow-action';
-import { SendEmailWorkflowAction } from 'src/modules/mail-sender/workflow-actions/send-email.workflow-action';
 import { ServerlessFunctionModule } from 'src/engine/metadata-modules/serverless-function/serverless-function.module';
 import { ScopedWorkspaceContextFactory } from 'src/engine/twenty-orm/factories/scoped-workspace-context.factory';
+import { SendEmailWorkflowAction } from 'src/modules/mail-sender/workflow-actions/send-email.workflow-action';
 import { MessagingGmailDriverModule } from 'src/modules/messaging/message-import-manager/drivers/gmail/messaging-gmail-driver.module';
+import { CodeWorkflowAction } from 'src/modules/serverless/workflow-actions/code.workflow-action';
+import { WorkflowCommonModule } from 'src/modules/workflow/common/workflow-common.module';
+import { WorkflowActionFactory } from 'src/modules/workflow/workflow-executor/factories/workflow-action.factory';
+import { WorkflowExecutorWorkspaceService } from 'src/modules/workflow/workflow-executor/workspace-services/workflow-executor.workspace-service';
 
 @Module({
   imports: [

packages/twenty-server/src/modules/workflow/workflow-executor/workspace-services/workflow-executor.workspace-service.ts

@@ -6,6 +6,7 @@ import {
 } from 'src/modules/workflow/common/standard-objects/workflow-run.workspace-entity';
 import { WorkflowActionFactory } from 'src/modules/workflow/workflow-executor/factories/workflow-action.factory';
 import { WorkflowStep } from 'src/modules/workflow/workflow-executor/types/workflow-action.type';
+import { resolveInput } from 'src/modules/workflow/workflow-executor/utils/variable-resolver.util';
 
 const MAX_RETRIES_ON_FAILURE = 3;
 
@@ -21,14 +22,14 @@ export class WorkflowExecutorWorkspaceService {
   async execute({
     currentStepIndex,
     steps,
-    payload,
+    context,
     output,
     attemptCount = 1,
   }: {
     currentStepIndex: number;
     steps: WorkflowStep[];
     output: WorkflowExecutorOutput;
-    payload?: object;
+    context: Record<string, unknown>;
     attemptCount?: number;
   }): Promise<WorkflowExecutorOutput> {
     if (currentStepIndex >= steps.length) {
@@ -39,59 +40,55 @@ export class WorkflowExecutorWorkspaceService {
 
     const workflowAction = this.workflowActionFactory.get(step.type);
 
-    const result = await workflowAction.execute({
-      step,
-      payload,
-    });
+    const actionPayload = resolveInput(step.settings.input, context);
 
-    const baseStepOutput = {
+    const result = await workflowAction.execute(actionPayload);
+
+    const stepOutput = output.steps[step.id];
+
+    const error =
+      result.error?.errorMessage ??
+      (result.result ? undefined : 'Execution result error, no data or error');
+
+    const updatedStepOutput = {
       id: step.id,
       name: step.name,
       type: step.type,
-      attemptCount,
+      outputs: [
+        ...(stepOutput?.outputs ?? []),
+        {
+          attemptCount,
+          result: result.result,
+          error,
+        },
+      ],
     };
 
     const updatedOutput = {
       ...output,
-      steps: [
+      steps: {
         ...output.steps,
-        {
-          ...baseStepOutput,
-          result: result.result,
-          error: result.error?.errorMessage,
-        },
-      ],
+        [step.id]: updatedStepOutput,
+      },
     };
 
     if (result.result) {
       return await this.execute({
         currentStepIndex: currentStepIndex + 1,
         steps,
-        payload: result.result,
+        context: {
+          ...context,
+          [step.id]: result.result,
+        },
         output: updatedOutput,
       });
     }
 
-    if (!result.error) {
-      return {
-        ...output,
-        steps: [
-          ...output.steps,
-          {
-            ...baseStepOutput,
-            result: undefined,
-            error: 'Execution result error, no data or error',
-          },
-        ],
-        status: WorkflowRunStatus.FAILED,
-      };
-    }
-
     if (step.settings.errorHandlingOptions.continueOnFailure.value) {
       return await this.execute({
         currentStepIndex: currentStepIndex + 1,
         steps,
-        payload,
+        context,
         output: updatedOutput,
       });
     }
@@ -103,7 +100,7 @@ export class WorkflowExecutorWorkspaceService {
       return await this.execute({
         currentStepIndex,
         steps,
-        payload,
+        context,
         output: updatedOutput,
         attemptCount: attemptCount + 1,
       });

packages/twenty-server/src/modules/workflow/workflow-runner/jobs/run-workflow.job.ts

@@ -40,9 +40,11 @@ export class RunWorkflowJob {
       await this.workflowExecutorWorkspaceService.execute({
         currentStepIndex: 0,
         steps: workflowVersion.steps || [],
-        payload,
+        context: {
+          trigger: payload,
+        },
         output: {
-          steps: [],
+          steps: {},
           status: WorkflowRunStatus.RUNNING,
         },
       });