feat: add-create-update-record in workflow (#14654)

## Description

- this PR focuses on issue
https://github.com/twentyhq/core-team-issues/issues/1476
- Added upsert action


## Visual Appearance

<img width="1792" height="1041" alt="Screenshot 2025-10-03 at 12 57
58 PM"
src="https://github.com/user-attachments/assets/57afb96c-d4b3-4a87-95f0-11ac4bd61dd8"
/>




<img width="1792" height="1031" alt="Screenshot 2025-10-03 at 12 57
48 PM"
src="https://github.com/user-attachments/assets/9032d4c2-f0d2-46f1-8682-a7e5c280a303"
/>

---------

Co-authored-by: Félix Malfait <felix.malfait@gmail.com>
Co-authored-by: Thomas Trompette <thomas.trompette@sfr.fr>

packages/twenty-server/src/engine/core-modules/record-crud/exceptions/record-crud.exception.ts

@@ -15,5 +15,6 @@ export enum RecordCrudExceptionCode {
   RECORD_CREATION_FAILED = 'RECORD_CREATION_FAILED',
   RECORD_UPDATE_FAILED = 'RECORD_UPDATE_FAILED',
   RECORD_DELETION_FAILED = 'RECORD_DELETION_FAILED',
+  RECORD_UPSERT_FAILED = 'RECORD_UPSERT_FAILED',
   QUERY_FAILED = 'QUERY_FAILED',
 }

packages/twenty-server/src/engine/core-modules/record-crud/record-crud.module.ts

@@ -4,6 +4,7 @@ import { CreateRecordService } from 'src/engine/core-modules/record-crud/service
 import { DeleteRecordService } from 'src/engine/core-modules/record-crud/services/delete-record.service';
 import { FindRecordsService } from 'src/engine/core-modules/record-crud/services/find-records.service';
 import { UpdateRecordService } from 'src/engine/core-modules/record-crud/services/update-record.service';
+import { UpsertRecordService } from 'src/engine/core-modules/record-crud/services/upsert-record.service';
 import { RecordPositionModule } from 'src/engine/core-modules/record-position/record-position.module';
 import { RecordTransformerModule } from 'src/engine/core-modules/record-transformer/record-transformer.module';
 import { TwentyORMModule } from 'src/engine/twenty-orm/twenty-orm.module';
@@ -21,12 +22,14 @@ import { WorkflowCommonModule } from 'src/modules/workflow/common/workflow-commo
     UpdateRecordService,
     DeleteRecordService,
     FindRecordsService,
+    UpsertRecordService,
   ],
   exports: [
     CreateRecordService,
     UpdateRecordService,
     DeleteRecordService,
     FindRecordsService,
+    UpsertRecordService,
   ],
 })
 export class RecordCrudModule {}

packages/twenty-server/src/engine/core-modules/record-crud/services/upsert-record.service.ts

@@ -0,0 +1,194 @@
+import { Injectable, Logger } from '@nestjs/common';
+
+import { isDefined } from 'twenty-shared/utils';
+import { canObjectBeManagedByWorkflow } from 'twenty-shared/workflow';
+
+import {
+  RecordCrudException,
+  RecordCrudExceptionCode,
+} from 'src/engine/core-modules/record-crud/exceptions/record-crud.exception';
+import { UpsertRecordParams } from 'src/engine/core-modules/record-crud/types/upsert-record-params.type';
+import { RecordInputTransformerService } from 'src/engine/core-modules/record-transformer/services/record-input-transformer.service';
+import { type ToolOutput } from 'src/engine/core-modules/tool/types/tool-output.type';
+import { computeCompositeColumnName } from 'src/engine/metadata-modules/field-metadata/utils/compute-column-name.util';
+import { getCompositeTypeOrThrow } from 'src/engine/metadata-modules/field-metadata/utils/get-composite-type-or-throw.util';
+import { isCompositeFieldMetadataType } from 'src/engine/metadata-modules/field-metadata/utils/is-composite-field-metadata-type.util';
+import { computeUniqueIndexWhereClause } from 'src/engine/metadata-modules/index-metadata/utils/compute-unique-index-where-clause.util';
+import { TwentyORMGlobalManager } from 'src/engine/twenty-orm/twenty-orm-global.manager';
+import { WorkflowCommonWorkspaceService } from 'src/modules/workflow/common/workspace-services/workflow-common.workspace-service';
+
+@Injectable()
+// eslint-disable-next-line @nx/workspace-inject-workspace-repository
+export class UpsertRecordService {
+  private readonly logger = new Logger(UpsertRecordService.name);
+
+  constructor(
+    private readonly twentyORMGlobalManager: TwentyORMGlobalManager,
+    private readonly recordInputTransformerService: RecordInputTransformerService,
+    private readonly workflowCommonWorkspaceService: WorkflowCommonWorkspaceService,
+  ) {}
+
+  async execute(params: UpsertRecordParams): Promise<ToolOutput> {
+    const {
+      objectName,
+      objectRecord,
+      fieldsToUpdate,
+      workspaceId,
+      rolePermissionConfig,
+    } = params;
+
+    if (!workspaceId) {
+      return {
+        success: false,
+        message: 'Failed to upsert record: Workspace ID is required',
+        error: 'Workspace ID not found',
+      };
+    }
+
+    try {
+      const repository =
+        await this.twentyORMGlobalManager.getRepositoryForWorkspace(
+          workspaceId,
+          objectName,
+          rolePermissionConfig,
+        );
+
+      const fieldsToUpdateArray = fieldsToUpdate || Object.keys(objectRecord);
+
+      const { objectMetadataItemWithFieldsMaps } =
+        await this.workflowCommonWorkspaceService.getObjectMetadataItemWithFieldsMaps(
+          objectName,
+          workspaceId,
+        );
+
+      if (
+        !canObjectBeManagedByWorkflow({
+          nameSingular: objectMetadataItemWithFieldsMaps.nameSingular,
+          isSystem: objectMetadataItemWithFieldsMaps.isSystem,
+        })
+      ) {
+        throw new RecordCrudException(
+          'Failed to update: Object cannot be updated by workflow',
+          RecordCrudExceptionCode.INVALID_REQUEST,
+        );
+      }
+
+      const objectRecordWithFilteredFields = Object.keys(objectRecord).reduce(
+        (acc, key) => {
+          if (fieldsToUpdateArray.includes(key)) {
+            return {
+              ...acc,
+              [key]: objectRecord[key],
+            };
+          }
+
+          return acc;
+        },
+        {},
+      );
+
+      const transformedObjectRecord =
+        await this.recordInputTransformerService.process({
+          recordInput: objectRecordWithFilteredFields,
+          objectMetadataMapItem: objectMetadataItemWithFieldsMaps,
+        });
+
+      const uniqueFieldsToUpdate = fieldsToUpdateArray
+        .map((field) => objectMetadataItemWithFieldsMaps.fieldIdByName[field])
+        .map((fieldId) => objectMetadataItemWithFieldsMaps.fieldsById[fieldId])
+        .filter((field) => field.isUnique || field.name === 'id');
+
+      const conflictPathsUniqueFieldsToUpdate = uniqueFieldsToUpdate.flatMap(
+        (field) => {
+          if (isCompositeFieldMetadataType(field.type)) {
+            const compositeType = getCompositeTypeOrThrow(field.type);
+
+            const uniqueProperties = compositeType.properties.filter(
+              (prop) => prop.isIncludedInUniqueConstraint,
+            );
+
+            const propertiesToUse =
+              uniqueProperties.length > 0
+                ? uniqueProperties
+                : [compositeType.properties[0]];
+
+            return propertiesToUse.map((prop) =>
+              computeCompositeColumnName(field, prop),
+            );
+          }
+
+          return [field.name];
+        },
+      );
+
+      const conflictPaths =
+        conflictPathsUniqueFieldsToUpdate.length > 0
+          ? conflictPathsUniqueFieldsToUpdate
+          : ['id'];
+
+      const indexPredicate = uniqueFieldsToUpdate
+        .map((field) =>
+          computeUniqueIndexWhereClause({
+            type: field.type,
+            name: field.name,
+          }),
+        )
+        .filter(isDefined);
+
+      const upsertResult = await repository.upsert(transformedObjectRecord, {
+        conflictPaths: conflictPaths,
+        indexPredicate:
+          indexPredicate.length > 0
+            ? `${indexPredicate.join(' AND ')}`
+            : undefined,
+      });
+
+      const upsertedRecordId = upsertResult.identifiers?.[0].id;
+
+      if (!isDefined(upsertedRecordId)) {
+        throw new RecordCrudException(
+          `Failed to upsert record in ${objectName}`,
+          RecordCrudExceptionCode.RECORD_UPSERT_FAILED,
+        );
+      }
+
+      const upsertedRecord = await repository.findOne({
+        where: {
+          id: upsertedRecordId,
+        },
+      });
+
+      if (!upsertedRecord) {
+        throw new RecordCrudException(
+          `Record not found after upsert with id ${upsertedRecordId} in ${objectName}`,
+          RecordCrudExceptionCode.RECORD_UPSERT_FAILED,
+        );
+      }
+
+      this.logger.log(`Record upserted successfully in ${objectName}`);
+
+      return {
+        success: true,
+        message: `Record upserted successfully in ${objectName}`,
+        result: upsertedRecord,
+      };
+    } catch (error) {
+      if (error instanceof RecordCrudException) {
+        return {
+          success: false,
+          message: `Failed to upsert record in ${objectName}`,
+          error: error.message,
+        };
+      }
+
+      this.logger.error(`Failed to upsert record: ${error}`);
+
+      return {
+        success: false,
+        message: `Failed to upsert record in ${objectName}`,
+        error:
+          error instanceof Error ? error.message : 'Failed to upsert record',
+      };
+    }
+  }
+}

packages/twenty-server/src/engine/core-modules/record-crud/types/upsert-record-params.type.ts

@@ -0,0 +1,10 @@
+import { type ObjectRecordProperties } from 'src/engine/core-modules/record-crud/types/object-record-properties.type';
+import { type RolePermissionConfig } from 'src/engine/twenty-orm/types/role-permission-config';
+
+export type UpsertRecordParams = {
+  objectName: string;
+  objectRecord: ObjectRecordProperties;
+  fieldsToUpdate?: string[];
+  workspaceId: string;
+  rolePermissionConfig?: RolePermissionConfig;
+};

packages/twenty-server/src/engine/core-modules/workflow/dtos/workflow-action.dto.ts

@@ -1,10 +1,14 @@
-import { Field, ObjectType } from '@nestjs/graphql';
+import { Field, ObjectType, registerEnumType } from '@nestjs/graphql';
 
 import graphqlTypeJson from 'graphql-type-json';
 
 import { UUIDScalarType } from 'src/engine/api/graphql/workspace-schema-builder/graphql-types/scalars';
-import { WorkflowActionType } from 'src/modules/workflow/workflow-executor/workflow-actions/types/workflow-action.type';
 import { WorkflowStepPosition } from 'src/engine/core-modules/workflow/dtos/workflow-step-position.dto';
+import { WorkflowActionType } from 'src/modules/workflow/workflow-executor/workflow-actions/types/workflow-action.type';
+
+registerEnumType(WorkflowActionType, {
+  name: 'WorkflowActionType',
+});
 
 @ObjectType('WorkflowAction')
 export class WorkflowActionDTO {
@@ -14,7 +18,7 @@ export class WorkflowActionDTO {
   @Field(() => String)
   name: string;
 
-  @Field(() => String)
+  @Field(() => WorkflowActionType)
   type: WorkflowActionType;
 
   @Field(() => graphqlTypeJson)