Move the workflow draft version overriding to the backend (#9328)

- In the `formatFieldMetadataValue` function, allow people to call
TypeORM's `save()` method with unserialized JSON data.
- Create an `overrideWorkflowDraftVersion` mutation that takes a
workflow id and the id of the workflow version to use as the new draft
  - If no draft exists yet, create one
  - If a draft already exists, deactivate its serverless functions
- Duplicate every step. For serverless function steps, it includes
duplicating the functions
  - Save the data of the step in DB
- Call the `overrideWorkflowDraftVersion` mutation in the old workflow
header and in the new Cmd+K actions
- I chose to not update the Apollo cache manually as the information of
the new draft are going to be automatically fetched once the user lands
on the workflow's show page. Note that we redirect the user to this page
after overriding the draft version.

packages/twenty-server/src/engine/core-modules/serverless/utils/serverless-get-folder.utils.ts

@@ -9,7 +9,7 @@ export const getServerlessFolder = ({
   version,
 }: {
   serverlessFunction: ServerlessFunctionEntity;
-  version?: string;
+  version?: 'draft' | 'latest' | (string & NonNullable<unknown>);
 }) => {
   const computedVersion =
     version === 'latest' ? serverlessFunction.latestVersion : version;

packages/twenty-server/src/engine/core-modules/workflow/dtos/create-draft-from-workflow-version-input.ts

@@ -0,0 +1,16 @@
+import { Field, InputType } from '@nestjs/graphql';
+
+@InputType()
+export class CreateDraftFromWorkflowVersionInput {
+  @Field(() => String, {
+    description: 'Workflow ID',
+    nullable: false,
+  })
+  workflowId: string;
+
+  @Field(() => String, {
+    description: 'Workflow version ID',
+    nullable: false,
+  })
+  workflowVersionIdToCopy: string;
+}

packages/twenty-server/src/engine/core-modules/workflow/resolvers/workflow-version-step.resolver.ts

@@ -1,16 +1,17 @@
 import { UseFilters, UseGuards } from '@nestjs/common';
 import { Args, Mutation, Resolver } from '@nestjs/graphql';
 
+import { CreateDraftFromWorkflowVersionInput } from 'src/engine/core-modules/workflow/dtos/create-draft-from-workflow-version-input';
+import { CreateWorkflowVersionStepInput } from 'src/engine/core-modules/workflow/dtos/create-workflow-version-step-input.dto';
+import { DeleteWorkflowVersionStepInput } from 'src/engine/core-modules/workflow/dtos/delete-workflow-version-step-input.dto';
+import { UpdateWorkflowVersionStepInput } from 'src/engine/core-modules/workflow/dtos/update-workflow-version-step-input.dto';
+import { WorkflowActionDTO } from 'src/engine/core-modules/workflow/dtos/workflow-step.dto';
 import { WorkflowTriggerGraphqlApiExceptionFilter } from 'src/engine/core-modules/workflow/filters/workflow-trigger-graphql-api-exception.filter';
+import { Workspace } from 'src/engine/core-modules/workspace/workspace.entity';
+import { AuthWorkspace } from 'src/engine/decorators/auth/auth-workspace.decorator';
 import { UserAuthGuard } from 'src/engine/guards/user-auth.guard';
 import { WorkspaceAuthGuard } from 'src/engine/guards/workspace-auth.guard';
-import { CreateWorkflowVersionStepInput } from 'src/engine/core-modules/workflow/dtos/create-workflow-version-step-input.dto';
-import { UpdateWorkflowVersionStepInput } from 'src/engine/core-modules/workflow/dtos/update-workflow-version-step-input.dto';
-import { DeleteWorkflowVersionStepInput } from 'src/engine/core-modules/workflow/dtos/delete-workflow-version-step-input.dto';
 import { WorkflowVersionStepWorkspaceService } from 'src/modules/workflow/common/workspace-services/workflow-version-step.workspace-service';
-import { AuthWorkspace } from 'src/engine/decorators/auth/auth-workspace.decorator';
-import { Workspace } from 'src/engine/core-modules/workspace/workspace.entity';
-import { WorkflowActionDTO } from 'src/engine/core-modules/workflow/dtos/workflow-step.dto';
 
 @Resolver()
 @UseGuards(WorkspaceAuthGuard, UserAuthGuard)
@@ -58,4 +59,24 @@ export class WorkflowVersionStepResolver {
       stepId,
     });
   }
+
+  @Mutation(() => Boolean)
+  async createDraftFromWorkflowVersion(
+    @AuthWorkspace() { id: workspaceId }: Workspace,
+    @Args('input')
+    {
+      workflowId,
+      workflowVersionIdToCopy,
+    }: CreateDraftFromWorkflowVersionInput,
+  ) {
+    await this.workflowVersionStepWorkspaceService.createDraftFromWorkflowVersion(
+      {
+        workspaceId,
+        workflowId,
+        workflowVersionIdToCopy,
+      },
+    );
+
+    return true;
+  }
 }

packages/twenty-server/src/engine/metadata-modules/serverless-function/serverless-function.service.ts

@@ -13,15 +13,23 @@ import { AnalyticsService } from 'src/engine/core-modules/analytics/analytics.se
 import { EnvironmentService } from 'src/engine/core-modules/environment/environment.service';
 import { FileStorageService } from 'src/engine/core-modules/file-storage/file-storage.service';
 import { readFileContent } from 'src/engine/core-modules/file-storage/utils/read-file-content';
+import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator';
+import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants';
+import { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service';
 import { ENV_FILE_NAME } from 'src/engine/core-modules/serverless/drivers/constants/env-file-name';
 import { INDEX_FILE_NAME } from 'src/engine/core-modules/serverless/drivers/constants/index-file-name';
 import { LAST_LAYER_VERSION } from 'src/engine/core-modules/serverless/drivers/layers/last-layer-version';
 import { getBaseTypescriptProjectFiles } from 'src/engine/core-modules/serverless/drivers/utils/get-base-typescript-project-files';
+import { getLayerDependencies } from 'src/engine/core-modules/serverless/drivers/utils/get-last-layer-dependencies';
 import { ServerlessService } from 'src/engine/core-modules/serverless/serverless.service';
 import { getServerlessFolder } from 'src/engine/core-modules/serverless/utils/serverless-get-folder.utils';
 import { ThrottlerService } from 'src/engine/core-modules/throttler/throttler.service';
 import { CreateServerlessFunctionInput } from 'src/engine/metadata-modules/serverless-function/dtos/create-serverless-function.input';
 import { UpdateServerlessFunctionInput } from 'src/engine/metadata-modules/serverless-function/dtos/update-serverless-function.input';
+import {
+  BuildServerlessFunctionBatchEvent,
+  BuildServerlessFunctionJob,
+} from 'src/engine/metadata-modules/serverless-function/jobs/build-serverless-function.job';
 import {
   ServerlessFunctionEntity,
   ServerlessFunctionSyncStatus,
@@ -31,14 +39,6 @@ import {
   ServerlessFunctionExceptionCode,
 } from 'src/engine/metadata-modules/serverless-function/serverless-function.exception';
 import { isDefined } from 'src/utils/is-defined';
-import { getLayerDependencies } from 'src/engine/core-modules/serverless/drivers/utils/get-last-layer-dependencies';
-import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator';
-import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants';
-import { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service';
-import {
-  BuildServerlessFunctionBatchEvent,
-  BuildServerlessFunctionJob,
-} from 'src/engine/metadata-modules/serverless-function/jobs/build-serverless-function.job';
 
 @Injectable()
 export class ServerlessFunctionService {
@@ -348,6 +348,68 @@ export class ServerlessFunctionService {
     });
   }
 
+  async copyOneServerlessFunction({
+    serverlessFunctionToCopyId,
+    serverlessFunctionToCopyVersion,
+    workspaceId,
+  }: {
+    serverlessFunctionToCopyId: string;
+    serverlessFunctionToCopyVersion: string;
+    workspaceId: string;
+  }) {
+    const serverlessFunctionToCopy =
+      await this.serverlessFunctionRepository.findOneBy({
+        workspaceId,
+        id: serverlessFunctionToCopyId,
+        latestVersion: serverlessFunctionToCopyVersion,
+      });
+
+    if (!serverlessFunctionToCopy) {
+      throw new ServerlessFunctionException(
+        'Function does not exist',
+        ServerlessFunctionExceptionCode.SERVERLESS_FUNCTION_NOT_FOUND,
+      );
+    }
+
+    const serverlessFunctionToCreate = this.serverlessFunctionRepository.create(
+      {
+        name: serverlessFunctionToCopy?.name,
+        description: serverlessFunctionToCopy?.description,
+        workspaceId,
+        layerVersion: LAST_LAYER_VERSION,
+      },
+    );
+
+    const copiedServerlessFunction =
+      await this.serverlessFunctionRepository.save(serverlessFunctionToCreate);
+
+    const serverlessFunctionToCopyFileFolder = getServerlessFolder({
+      serverlessFunction: serverlessFunctionToCopy,
+      version: 'latest',
+    });
+    const copiedServerlessFunctionFileFolder = getServerlessFolder({
+      serverlessFunction: copiedServerlessFunction,
+      version: 'draft',
+    });
+
+    await this.fileStorageService.copy({
+      from: {
+        folderPath: serverlessFunctionToCopyFileFolder,
+      },
+      to: {
+        folderPath: copiedServerlessFunctionFileFolder,
+      },
+    });
+
+    await this.buildServerlessFunction({
+      serverlessFunctionId: copiedServerlessFunction.id,
+      serverlessFunctionVersion: 'draft',
+      workspaceId,
+    });
+
+    return copiedServerlessFunction;
+  }
+
   private async throttleExecution(workspaceId: string) {
     try {
       await this.throttlerService.throttle(

packages/twenty-server/src/engine/twenty-orm/utils/format-data.util.ts

@@ -82,7 +82,10 @@ function formatFieldMetadataValue(
   value: any,
   fieldMetadata: FieldMetadataInterface,
 ) {
-  if (fieldMetadata.type === FieldMetadataType.RAW_JSON) {
+  if (
+    fieldMetadata.type === FieldMetadataType.RAW_JSON &&
+    typeof value === 'string'
+  ) {
     return JSON.parse(value as string);
   }
 

packages/twenty-server/src/modules/workflow/common/utils/assert-workflow-version-has-steps.ts

@@ -0,0 +1,19 @@
+import { WorkflowVersionWorkspaceEntity } from 'src/modules/workflow/common/standard-objects/workflow-version.workspace-entity';
+import { WorkflowAction } from 'src/modules/workflow/workflow-executor/workflow-actions/types/workflow-action.type';
+import {
+  WorkflowTriggerException,
+  WorkflowTriggerExceptionCode,
+} from 'src/modules/workflow/workflow-trigger/exceptions/workflow-trigger.exception';
+
+export function assertWorkflowVersionHasSteps(
+  workflowVersion: WorkflowVersionWorkspaceEntity,
+): asserts workflowVersion is WorkflowVersionWorkspaceEntity & {
+  steps: WorkflowAction[];
+} {
+  if (workflowVersion.steps === null || workflowVersion.steps.length < 1) {
+    throw new WorkflowTriggerException(
+      'Workflow version does not contain at least one step',
+      WorkflowTriggerExceptionCode.INVALID_WORKFLOW_VERSION,
+    );
+  }
+}

packages/twenty-server/src/modules/workflow/common/workspace-services/workflow-version-step.workspace-service.ts

@@ -13,7 +13,13 @@ import {
   WorkflowVersionStepException,
   WorkflowVersionStepExceptionCode,
 } from 'src/modules/workflow/common/exceptions/workflow-version-step.exception';
-import { WorkflowVersionWorkspaceEntity } from 'src/modules/workflow/common/standard-objects/workflow-version.workspace-entity';
+import {
+  WorkflowVersionStatus,
+  WorkflowVersionWorkspaceEntity,
+} from 'src/modules/workflow/common/standard-objects/workflow-version.workspace-entity';
+import { assertWorkflowVersionHasSteps } from 'src/modules/workflow/common/utils/assert-workflow-version-has-steps';
+import { assertWorkflowVersionIsDraft } from 'src/modules/workflow/common/utils/assert-workflow-version-is-draft.util';
+import { assertWorkflowVersionTriggerIsDefined } from 'src/modules/workflow/common/utils/assert-workflow-version-trigger-is-defined.util';
 import { WorkflowBuilderWorkspaceService } from 'src/modules/workflow/workflow-builder/workflow-builder.workspace-service';
 import { BaseWorkflowActionSettings } from 'src/modules/workflow/workflow-executor/workflow-actions/types/workflow-action-settings.type';
 import {
@@ -56,7 +62,7 @@ export class WorkflowVersionStepWorkspaceService {
   }): Promise<WorkflowAction> {
     const newStepId = v4();
 
-    switch (`${type}`) {
+    switch (type) {
       case WorkflowActionType.CODE: {
         const newServerlessFunction =
           await this.serverlessFunctionService.createOneServerlessFunction(
@@ -185,6 +191,48 @@ export class WorkflowVersionStepWorkspaceService {
     }
   }
 
+  private async duplicateStep({
+    step,
+    workspaceId,
+  }: {
+    step: WorkflowAction;
+    workspaceId: string;
+  }): Promise<WorkflowAction> {
+    const newStepId = v4();
+
+    switch (step.type) {
+      case WorkflowActionType.CODE: {
+        const copiedServerlessFunction =
+          await this.serverlessFunctionService.copyOneServerlessFunction({
+            serverlessFunctionToCopyId:
+              step.settings.input.serverlessFunctionId,
+            serverlessFunctionToCopyVersion:
+              step.settings.input.serverlessFunctionVersion,
+            workspaceId,
+          });
+
+        return {
+          ...step,
+          id: newStepId,
+          settings: {
+            ...step.settings,
+            input: {
+              ...step.settings.input,
+              serverlessFunctionId: copiedServerlessFunction.id,
+              serverlessFunctionVersion: copiedServerlessFunction.latestVersion,
+            },
+          },
+        };
+      }
+      default: {
+        return {
+          ...step,
+          id: newStepId,
+        };
+      }
+    }
+  }
+
   private async enrichOutputSchema({
     step,
     workspaceId,
@@ -363,14 +411,113 @@ export class WorkflowVersionStepWorkspaceService {
       workflowVersionUpdates,
     );
 
-    switch (stepToDelete.type) {
-      case WorkflowActionType.CODE:
-        await this.serverlessFunctionService.deleteOneServerlessFunction(
-          stepToDelete.settings.input.serverlessFunctionId,
-          workspaceId,
-        );
-    }
+    await this.runWorkflowVersionStepDeletionSideEffects({
+      step: stepToDelete,
+      workspaceId,
+    });
 
     return stepToDelete;
   }
+
+  async createDraftFromWorkflowVersion({
+    workspaceId,
+    workflowId,
+    workflowVersionIdToCopy,
+  }: {
+    workspaceId: string;
+    workflowId: string;
+    workflowVersionIdToCopy: string;
+  }) {
+    const workflowVersionRepository =
+      await this.twentyORMManager.getRepository<WorkflowVersionWorkspaceEntity>(
+        'workflowVersion',
+      );
+
+    const workflowVersionToCopy = await workflowVersionRepository.findOne({
+      where: {
+        id: workflowVersionIdToCopy,
+        workflowId,
+      },
+    });
+
+    if (!isDefined(workflowVersionToCopy)) {
+      throw new WorkflowVersionStepException(
+        'WorkflowVersion to copy not found',
+        WorkflowVersionStepExceptionCode.NOT_FOUND,
+      );
+    }
+
+    assertWorkflowVersionTriggerIsDefined(workflowVersionToCopy);
+    assertWorkflowVersionHasSteps(workflowVersionToCopy);
+
+    let draftWorkflowVersion = await workflowVersionRepository.findOne({
+      where: {
+        workflowId,
+        status: WorkflowVersionStatus.DRAFT,
+      },
+    });
+
+    if (!isDefined(draftWorkflowVersion)) {
+      const workflowVersionsCount = await workflowVersionRepository.count({
+        where: {
+          workflowId,
+        },
+      });
+
+      draftWorkflowVersion = await workflowVersionRepository.save({
+        workflowId,
+        name: `v${workflowVersionsCount + 1}`,
+        status: WorkflowVersionStatus.DRAFT,
+      });
+    }
+
+    assertWorkflowVersionIsDraft(draftWorkflowVersion);
+
+    if (Array.isArray(draftWorkflowVersion.steps)) {
+      await Promise.all(
+        draftWorkflowVersion.steps.map((step) =>
+          this.runWorkflowVersionStepDeletionSideEffects({
+            step,
+            workspaceId,
+          }),
+        ),
+      );
+    }
+
+    const newWorkflowVersionTrigger = workflowVersionToCopy.trigger;
+    const newWorkflowVersionSteps: WorkflowAction[] = [];
+
+    for (const step of workflowVersionToCopy.steps) {
+      const duplicatedStep = await this.duplicateStep({
+        step,
+        workspaceId,
+      });
+
+      newWorkflowVersionSteps.push(duplicatedStep);
+    }
+
+    await workflowVersionRepository.update(draftWorkflowVersion.id, {
+      steps: newWorkflowVersionSteps,
+      trigger: newWorkflowVersionTrigger,
+    });
+  }
+
+  private async runWorkflowVersionStepDeletionSideEffects({
+    step,
+    workspaceId,
+  }: {
+    step: WorkflowAction;
+    workspaceId: string;
+  }) {
+    switch (step.type) {
+      case WorkflowActionType.CODE: {
+        await this.serverlessFunctionService.deleteOneServerlessFunction(
+          step.settings.input.serverlessFunctionId,
+          workspaceId,
+        );
+
+        break;
+      }
+    }
+  }
 }