Add workflow run entity (#6622)

- create a workflow run every time a workflow is triggered in
not_started status. This status will be helpful later for once workflows
will be scheduled
- update run status once workflow starts running
- complete status once the workflow finished running
- add a failed status if an error occurs

packages/twenty-server/src/modules/workflow/workflow-runner/workflow-runner.exception.ts

@@ -0,0 +1,11 @@
+import { CustomException } from 'src/utils/custom-exception';
+
+export class WorkflowRunnerException extends CustomException {
+  constructor(message: string, code: string) {
+    super(message, code);
+  }
+}
+
+export enum WorkflowRunnerExceptionCode {
+  WORKFLOW_FAILED = 'WORKFLOW_FAILED',
+}

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

@@ -1,54 +1,60 @@
+import { Scope } from '@nestjs/common';
+
 import { Process } from 'src/engine/integrations/message-queue/decorators/process.decorator';
 import { Processor } from 'src/engine/integrations/message-queue/decorators/processor.decorator';
 import { MessageQueue } from 'src/engine/integrations/message-queue/message-queue.constants';
-import { TwentyORMGlobalManager } from 'src/engine/twenty-orm/twenty-orm-global.manager';
-import { WorkflowWorkspaceEntity } from 'src/modules/workflow/common/standard-objects/workflow.workspace-entity';
+import { WorkflowRunStatus } from 'src/modules/workflow/common/standard-objects/workflow-run.workspace-entity';
 import { WorkflowCommonService } from 'src/modules/workflow/common/workflow-common.services';
 import { WorkflowRunnerService } from 'src/modules/workflow/workflow-runner/workflow-runner.service';
+import { WorkflowStatusWorkspaceService } from 'src/modules/workflow/workflow-status/workflow-status.workspace-service';
 
 export type RunWorkflowJobData = {
   workspaceId: string;
-  workflowId: string;
+  workflowVersionId: string;
+  workflowRunId: string;
   payload: object;
 };
 
-@Processor(MessageQueue.workflowQueue)
+@Processor({ queueName: MessageQueue.workflowQueue, scope: Scope.REQUEST })
 export class WorkflowRunnerJob {
   constructor(
     private readonly workflowCommonService: WorkflowCommonService,
     private readonly workflowRunnerService: WorkflowRunnerService,
-    private readonly twentyORMGlobalManager: TwentyORMGlobalManager,
+    private readonly workflowStatusWorkspaceService: WorkflowStatusWorkspaceService,
   ) {}
 
   @Process(WorkflowRunnerJob.name)
   async handle({
     workspaceId,
-    workflowId,
+    workflowVersionId,
+    workflowRunId,
     payload,
   }: RunWorkflowJobData): Promise<void> {
-    const workflowRepository =
-      await this.twentyORMGlobalManager.getRepositoryForWorkspace<WorkflowWorkspaceEntity>(
-        workspaceId,
-        'workflow',
-      );
-
-    const workflow = await workflowRepository.findOneByOrFail({
-      id: workflowId,
-    });
-
-    if (!workflow.publishedVersionId) {
-      throw new Error('Workflow has no published version');
-    }
+    await this.workflowStatusWorkspaceService.startWorkflowRun(workflowRunId);
 
     const workflowVersion = await this.workflowCommonService.getWorkflowVersion(
       workspaceId,
-      workflow.publishedVersionId,
+      workflowVersionId,
     );
 
-    await this.workflowRunnerService.run({
-      action: workflowVersion.trigger.nextAction,
-      workspaceId,
-      payload,
-    });
+    try {
+      await this.workflowRunnerService.run({
+        action: workflowVersion.trigger.nextAction,
+        workspaceId,
+        payload,
+      });
+
+      await this.workflowStatusWorkspaceService.endWorkflowRun(
+        workflowRunId,
+        WorkflowRunStatus.COMPLETED,
+      );
+    } catch (error) {
+      await this.workflowStatusWorkspaceService.endWorkflowRun(
+        workflowRunId,
+        WorkflowRunStatus.FAILED,
+      );
+
+      throw error;
+    }
   }
 }

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

@@ -1,12 +1,17 @@
 import { Module } from '@nestjs/common';
 
-import { WorkflowRunnerService } from 'src/modules/workflow/workflow-runner/workflow-runner.service';
-import { WorkflowRunnerJob } from 'src/modules/workflow/workflow-runner/workflow-runner.job';
 import { WorkflowCommonModule } from 'src/modules/workflow/common/workflow-common.module';
 import { WorkflowActionRunnerModule } from 'src/modules/workflow/workflow-action-runner/workflow-action-runner.module';
+import { WorkflowRunnerJob } from 'src/modules/workflow/workflow-runner/workflow-runner.job';
+import { WorkflowRunnerService } from 'src/modules/workflow/workflow-runner/workflow-runner.service';
+import { WorkflowStatusModule } from 'src/modules/workflow/workflow-status/workflow-status.module';
 
 @Module({
-  imports: [WorkflowCommonModule, WorkflowActionRunnerModule],
+  imports: [
+    WorkflowCommonModule,
+    WorkflowActionRunnerModule,
+    WorkflowStatusModule,
+  ],
   providers: [WorkflowRunnerService, WorkflowRunnerJob],
   exports: [WorkflowRunnerService],
 })

packages/twenty-server/src/modules/workflow/workflow-runner/workflow-runner.service.ts

@@ -2,9 +2,18 @@ import { Injectable } from '@nestjs/common';
 
 import { WorkflowAction } from 'src/modules/workflow/common/types/workflow-action.type';
 import { WorkflowActionRunnerFactory } from 'src/modules/workflow/workflow-action-runner/workflow-action-runner.factory';
+import {
+  WorkflowRunnerException,
+  WorkflowRunnerExceptionCode,
+} from 'src/modules/workflow/workflow-runner/workflow-runner.exception';
 
 const MAX_RETRIES_ON_FAILURE = 3;
 
+export type WorkflowRunOutput = {
+  data?: object;
+  error?: object;
+};
+
 @Injectable()
 export class WorkflowRunnerService {
   constructor(
@@ -21,9 +30,11 @@ export class WorkflowRunnerService {
     workspaceId: string;
     payload?: object;
     attemptCount?: number;
-  }) {
+  }): Promise<WorkflowRunOutput> {
     if (!action) {
-      return payload;
+      return {
+        data: payload,
+      };
     }
 
     const workflowActionRunner = this.workflowActionRunnerFactory.get(
@@ -45,7 +56,10 @@ export class WorkflowRunnerService {
     }
 
     if (!result.error) {
-      throw new Error('Execution result error, no data or error');
+      throw new WorkflowRunnerException(
+        'Execution result error, no data or error',
+        WorkflowRunnerExceptionCode.WORKFLOW_FAILED,
+      );
     }
 
     if (action.settings.errorHandlingOptions.continueOnFailure.value) {
@@ -68,6 +82,9 @@ export class WorkflowRunnerService {
       });
     }
 
-    return result.error;
+    throw new WorkflowRunnerException(
+      `Workflow failed: ${result.error}`,
+      WorkflowRunnerExceptionCode.WORKFLOW_FAILED,
+    );
   }
 }