Add submit form step endpoint (#10538)

- add endpoint to submit form step - update context and output of workflow run - resume workflow execution
2025-02-28 10:03:45 +01:00
parent 15d3751b73
commit a29c26c8d7
11 changed files with 285 additions and 64 deletions
@@ -5,8 +5,10 @@ import { Process } from 'src/engine/core-modules/message-queue/decorators/proces
 import { Processor } from 'src/engine/core-modules/message-queue/decorators/processor.decorator';
 import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants';
 import { ThrottlerService } from 'src/engine/core-modules/throttler/throttler.service';
+import { TwentyORMManager } from 'src/engine/twenty-orm/twenty-orm.manager';
 import { WorkflowRunStatus } from 'src/modules/workflow/common/standard-objects/workflow-run.workspace-entity';
 import { WorkflowCommonWorkspaceService } from 'src/modules/workflow/common/workspace-services/workflow-common.workspace-service';
+import { WorkflowAction } from 'src/modules/workflow/workflow-executor/workflow-actions/types/workflow-action.type';
 import { WorkflowExecutorWorkspaceService } from 'src/modules/workflow/workflow-executor/workspace-services/workflow-executor.workspace-service';
 import {
  WorkflowRunException,
@@ -16,9 +18,9 @@ import { WorkflowRunWorkspaceService } from 'src/modules/workflow/workflow-runne

 export type RunWorkflowJobData = {
  workspaceId: string;
-  workflowVersionId: string;
  workflowRunId: string;
-  payload: object;
+  payload?: object;
+  lastExecutedStepId?: string;
 };

@Processor({ queueName: MessageQueue.workflowQueue, scope: Scope.REQUEST })
@@ -29,61 +31,27 @@ export class RunWorkflowJob {
    private readonly workflowRunWorkspaceService: WorkflowRunWorkspaceService,
    private readonly throttlerService: ThrottlerService,
    private readonly environmentService: EnvironmentService,
+    private readonly twentyORMManager: TwentyORMManager,
  ) {}

  @Process(RunWorkflowJob.name)
  async handle({
-    workflowVersionId,
    workflowRunId,
    payload,
+    lastExecutedStepId,
  }: RunWorkflowJobData): Promise<void> {
-    const context = {
-      trigger: payload,
-    };
-
    try {
-      const workflowVersion =
-        await this.workflowCommonWorkspaceService.getWorkflowVersionOrFail(
-          workflowVersionId,
-        );
-
-      if (!workflowVersion.trigger || !workflowVersion.steps) {
-        throw new WorkflowRunException(
-          'Workflow version has no trigger or steps',
-          WorkflowRunExceptionCode.WORKFLOW_RUN_INVALID,
-        );
-      }
-
-      await this.workflowRunWorkspaceService.startWorkflowRun({
-        workflowRunId,
-        context,
-        output: {
-          flow: {
-            trigger: workflowVersion.trigger,
-            steps: workflowVersion.steps,
-          },
-        },
-      });
-
-      await this.throttleExecution(workflowVersion.workflowId);
-
-      const { error, pendingEvent } =
-        await this.workflowExecutorWorkspaceService.execute({
+      if (lastExecutedStepId) {
+        await this.resumeWorkflowExecution({
          workflowRunId,
-          currentStepIndex: 0,
-          steps: workflowVersion.steps,
-          context,
+          lastExecutedStepId,
+        });
+      } else {
+        await this.startWorkflowExecution({
+          workflowRunId,
+          payload: payload ?? {},
        });
-
-      if (pendingEvent) {
-        return;
      }
-
-      await this.workflowRunWorkspaceService.endWorkflowRun({
-        workflowRunId,
-        status: error ? WorkflowRunStatus.FAILED : WorkflowRunStatus.COMPLETED,
-        error,
-      });
    } catch (error) {
      await this.workflowRunWorkspaceService.endWorkflowRun({
        workflowRunId,
@@ -93,6 +61,123 @@ export class RunWorkflowJob {
    }
  }

+  private async startWorkflowExecution({
+    workflowRunId,
+    payload,
+  }: {
+    workflowRunId: string;
+    payload: object;
+  }): Promise<void> {
+    const context = {
+      trigger: payload,
+    };
+
+    const workflowRun =
+      await this.workflowRunWorkspaceService.getWorkflowRunOrFail(
+        workflowRunId,
+      );
+
+    const workflowVersion =
+      await this.workflowCommonWorkspaceService.getWorkflowVersionOrFail(
+        workflowRun.workflowVersionId,
+      );
+
+    if (!workflowVersion.trigger || !workflowVersion.steps) {
+      throw new WorkflowRunException(
+        'Workflow version has no trigger or steps',
+        WorkflowRunExceptionCode.WORKFLOW_RUN_INVALID,
+      );
+    }
+
+    await this.workflowRunWorkspaceService.startWorkflowRun({
+      workflowRunId,
+      context,
+      output: {
+        flow: {
+          trigger: workflowVersion.trigger,
+          steps: workflowVersion.steps,
+        },
+      },
+    });
+
+    await this.throttleExecution(workflowVersion.workflowId);
+
+    await this.executeWorkflow({
+      workflowRunId,
+      currentStepIndex: 0,
+      steps: workflowVersion.steps,
+      context,
+    });
+  }
+
+  private async resumeWorkflowExecution({
+    workflowRunId,
+    lastExecutedStepId,
+  }: {
+    workflowRunId: string;
+    lastExecutedStepId: string;
+  }): Promise<void> {
+    const workflowRun =
+      await this.workflowRunWorkspaceService.getWorkflowRunOrFail(
+        workflowRunId,
+      );
+
+    if (workflowRun.status !== WorkflowRunStatus.RUNNING) {
+      throw new WorkflowRunException(
+        'Workflow is not running',
+        WorkflowRunExceptionCode.WORKFLOW_RUN_INVALID,
+      );
+    }
+
+    const lastExecutedStepIndex = workflowRun.output?.flow?.steps?.findIndex(
+      (step) => step.id === lastExecutedStepId,
+    );
+
+    if (lastExecutedStepIndex === undefined) {
+      throw new WorkflowRunException(
+        'Last executed step not found',
+        WorkflowRunExceptionCode.INVALID_INPUT,
+      );
+    }
+
+    await this.executeWorkflow({
+      workflowRunId,
+      currentStepIndex: lastExecutedStepIndex + 1,
+      steps: workflowRun.output?.flow?.steps ?? [],
+      context: workflowRun.context ?? {},
+    });
+  }
+
+  private async executeWorkflow({
+    workflowRunId,
+    currentStepIndex,
+    steps,
+    context,
+  }: {
+    workflowRunId: string;
+    currentStepIndex: number;
+    steps: WorkflowAction[];
+    context: Record<string, any>;
+  }) {
+    const { error, pendingEvent } =
+      await this.workflowExecutorWorkspaceService.execute({
+        workflowRunId,
+        currentStepIndex,
+        steps,
+        context,
+      });
+
+    if (pendingEvent) {
+      return;
+    }
+
+    await this.workflowRunWorkspaceService.endWorkflowRun({
+      workflowRunId,
+      status: error ? WorkflowRunStatus.FAILED : WorkflowRunStatus.COMPLETED,
+      error,
+    });
+  }
+
  private async throttleExecution(workflowId: string) {
    try {
      await this.throttlerService.throttle(
@@ -163,4 +163,26 @@ export class WorkflowRunWorkspaceService {
      context,
    });
  }
+
+  async getWorkflowRunOrFail(
+    workflowRunId: string,
+  ): Promise<WorkflowRunWorkspaceEntity> {
+    const workflowRunRepository =
+      await this.twentyORMManager.getRepository<WorkflowRunWorkspaceEntity>(
+        'workflowRun',
+      );
+
+    const workflowRun = await workflowRunRepository.findOne({
+      where: { id: workflowRunId },
+    });
+
+    if (!workflowRun) {
+      throw new WorkflowRunException(
+        'Workflow run not found',
+        WorkflowRunExceptionCode.WORKFLOW_RUN_NOT_FOUND,
+      );
+    }
+
+    return workflowRun;
+  }
 }
@@ -45,7 +45,6 @@ export class WorkflowRunnerWorkspaceService {
      RunWorkflowJob.name,
      {
        workspaceId,
-        workflowVersionId,
        payload: payload,
        workflowRunId,
      },
@@ -53,4 +52,23 @@ export class WorkflowRunnerWorkspaceService {

    return { workflowRunId };
  }
+
+  async resume({
+    workspaceId,
+    workflowRunId,
+    lastExecutedStepId,
+  }: {
+    workspaceId: string;
+    workflowRunId: string;
+    lastExecutedStepId: string;
+  }) {
+    await this.messageQueueService.add<RunWorkflowJobData>(
+      RunWorkflowJob.name,
+      {
+        workspaceId,
+        workflowRunId,
+        lastExecutedStepId,
+      },
+    );
+  }
 }