[permissions] Update permission check layer (#13485)

Fixes https://github.com/twentyhq/core-team-issues/issues/1262

In this PR we add the update permission check layer by 
- for the graphql api: extracting columns to update from the
expressionMap
- for rest api: .save() is used so we need to add the permission layer
to .save directly. We also take advantage of this PR to filter out
non-readable fields from save response (other save returns the whole
entity) - this was planned in
https://github.com/twentyhq/core-team-issues/issues/1216

The current solution does not work with rest api depth 2 queries, but
this seem to already not work on main (for timeout reasons though, so
different). I offer to create a ticket to fix it altogether later.

packages/twenty-server/src/engine/twenty-orm/entity-manager/workspace-entity-manager.spec.ts

@@ -15,6 +15,30 @@ jest.mock('src/engine/twenty-orm/repository/permissions.utils', () => ({
   validateOperationIsPermittedOrThrow: jest.fn(),
 }));
 
+jest.mock(
+  'src/engine/twenty-orm/utils/get-object-metadata-from-entity-target.util',
+  () => ({
+    getObjectMetadataFromEntityTarget: jest.fn().mockReturnValue({}),
+  }),
+);
+
+jest.mock('src/engine/twenty-orm/utils/format-data.util', () => ({
+  formatData: jest.fn().mockReturnValue([]),
+}));
+
+jest.mock('src/engine/twenty-orm/utils/format-result.util', () => ({
+  formatResult: jest.fn().mockReturnValue([]),
+}));
+
+jest.mock(
+  'src/engine/twenty-orm/entity-manager/workspace-entity-manager',
+  () => ({
+    ...jest.requireActual(
+      'src/engine/twenty-orm/entity-manager/workspace-entity-manager',
+    ),
+  }),
+);
+
 const mockedWorkspaceUpdateQueryBuilder = {
   set: jest.fn().mockImplementation(() => ({
     where: jest.fn().mockReturnThis(),
@@ -122,7 +146,22 @@ describe('WorkspaceEntityManager', () => {
     } as WorkspaceInternalContext;
 
     mockDataSource = {
-      featureFlagMap: {},
+      featureFlagMap: {
+        IS_AIRTABLE_INTEGRATION_ENABLED: false,
+        IS_POSTGRESQL_INTEGRATION_ENABLED: false,
+        IS_STRIPE_INTEGRATION_ENABLED: false,
+        IS_UNIQUE_INDEXES_ENABLED: false,
+        IS_JSON_FILTER_ENABLED: false,
+        IS_AI_ENABLED: false,
+        IS_IMAP_SMTP_CALDAV_ENABLED: false,
+        IS_MORPH_RELATION_ENABLED: false,
+        IS_WORKFLOW_FILTERING_ENABLED: false,
+        IS_RELATION_CONNECT_ENABLED: false,
+        IS_WORKSPACE_API_KEY_WEBHOOK_GRAPHQL_ENABLED: false,
+        IS_FIELDS_PERMISSIONS_ENABLED: true,
+        IS_CORE_VIEW_SYNCING_ENABLED: false,
+        IS_TWO_FACTOR_AUTHENTICATION_ENABLED: false,
+      },
       permissionsPerRoleId: {},
     } as WorkspaceDataSource;
 
@@ -191,6 +230,15 @@ describe('WorkspaceEntityManager', () => {
 
     jest.spyOn(entityManager as any, 'validatePermissions');
     jest.spyOn(entityManager as any, 'createQueryBuilder');
+    jest
+      .spyOn(entityManager as any, 'getFormattedResultWithoutNonReadableFields')
+      .mockImplementation(
+        ({ formattedResult }: { formattedResult: string[] }) => formattedResult,
+      );
+
+    jest.spyOn(entityManager as any, 'getFeatureFlagMap').mockReturnValue({
+      IS_FIELDS_PERMISSIONS_ENABLED: true,
+    });
 
     jest
       .spyOn(entityManager as any, 'extractTargetNameSingularFromEntityTarget')
@@ -276,15 +324,18 @@ describe('WorkspaceEntityManager', () => {
         operationType: 'update',
         permissionOptions: mockPermissionOptions,
         selectedColumns: [],
+        updatedColumns: [],
       });
       expect(validateOperationIsPermittedOrThrow).toHaveBeenCalledWith({
         entityName: 'test-entity',
+        isFieldPermissionsEnabled: true,
         operationType: 'update',
         objectMetadataMaps: mockInternalContext.objectMetadataMaps,
         objectRecordsPermissions:
           mockPermissionOptions.objectRecordsPermissions,
         selectedColumns: [],
         allFieldsSelected: false,
+        updatedColumns: [],
       });
     });
   });
@@ -313,11 +364,13 @@ describe('WorkspaceEntityManager', () => {
       expect(validateOperationIsPermittedOrThrow).toHaveBeenCalledWith({
         entityName: 'test-entity',
         operationType: 'delete',
+        isFieldPermissionsEnabled: true,
         objectMetadataMaps: mockInternalContext.objectMetadataMaps,
         objectRecordsPermissions:
           mockPermissionOptions.objectRecordsPermissions,
         selectedColumns: [],
         allFieldsSelected: false,
+        updatedColumns: [],
       });
     });
   });

packages/twenty-server/src/engine/twenty-orm/entity-manager/workspace-entity-manager.ts

@@ -1,4 +1,4 @@
-import { Entity } from '@microsoft/microsoft-graph-types';
+import isEmpty from 'lodash.isempty';
 import { ObjectRecordsPermissions } from 'twenty-shared/types';
 import { isDefined } from 'twenty-shared/utils';
 import {
@@ -35,10 +35,12 @@ import { WorkspaceInternalContext } from 'src/engine/twenty-orm/interfaces/works
 
 import { DatabaseEventAction } from 'src/engine/api/graphql/graphql-query-runner/enums/database-event-action';
 import { AuthContext } from 'src/engine/core-modules/auth/types/auth-context.type';
+import { InternalServerError } from 'src/engine/core-modules/graphql/utils/graphql-errors.util';
 import {
   PermissionsException,
   PermissionsExceptionCode,
 } from 'src/engine/metadata-modules/permissions/permissions.exception';
+import { ObjectMetadataItemWithFieldMaps } from 'src/engine/metadata-modules/types/object-metadata-item-with-field-maps';
 import { WorkspaceDataSource } from 'src/engine/twenty-orm/datasource/workspace.datasource';
 import { DeepPartialWithNestedRelationFields } from 'src/engine/twenty-orm/entity-manager/types/deep-partial-entity-with-nested-relation-fields.type';
 import { QueryDeepPartialEntityWithNestedRelationFields } from 'src/engine/twenty-orm/entity-manager/types/query-deep-partial-entity-with-nested-relation-fields.type';
@@ -352,6 +354,7 @@ export class WorkspaceEntityManager extends EntityManager {
     operationType,
     permissionOptions,
     selectedColumns,
+    updatedColumns = [],
   }: {
     target: EntityTarget<Entity> | Entity;
     operationType: OperationType;
@@ -360,6 +363,7 @@ export class WorkspaceEntityManager extends EntityManager {
       objectRecordsPermissions?: ObjectRecordsPermissions;
     };
     selectedColumns: string[];
+    updatedColumns?: string[];
   }): void {
     if (permissionOptions?.shouldBypassPermissionChecks === true) {
       return;
@@ -378,6 +382,9 @@ export class WorkspaceEntityManager extends EntityManager {
       objectMetadataMaps: this.internalContext.objectMetadataMaps,
       selectedColumns,
       allFieldsSelected: false,
+      updatedColumns,
+      isFieldPermissionsEnabled:
+        this.getFeatureFlagMap().IS_FIELDS_PERMISSIONS_ENABLED,
     });
   }
 
@@ -702,7 +709,7 @@ export class WorkspaceEntityManager extends EntityManager {
       permissionOptions,
     )
       .setFindOptions(options || {})
-      .select('1')
+      .select('id')
       .limit(1)
       .getRawOne()
       .then((result) => isDefined(result));
@@ -722,7 +729,7 @@ export class WorkspaceEntityManager extends EntityManager {
       permissionOptions,
     )
       .setFindOptions({ where })
-      .select('1')
+      .select('id')
       .limit(1)
       .getRawOne()
       .then((result) => isDefined(result));
@@ -762,7 +769,7 @@ export class WorkspaceEntityManager extends EntityManager {
       .getCount();
   }
 
-  async callAggregateFunCustom(
+  async callAggregateFunCustom<Entity extends ObjectLiteral>(
     entityClass: EntityTarget<Entity>,
     fnName: string,
     columnName: string,
@@ -1033,13 +1040,6 @@ export class WorkspaceEntityManager extends EntityManager {
         ? maybeOptionsOrMaybePermissionOptions
         : permissionOptions;
 
-    this.validatePermissions({
-      target: targetOrEntity,
-      operationType: 'update',
-      permissionOptions: permissionOptionsFromArgs,
-      selectedColumns: [], // TODO
-    });
-
     let target =
       arguments.length > 1 &&
       (typeof targetOrEntity === 'function' ||
@@ -1047,7 +1047,9 @@ export class WorkspaceEntityManager extends EntityManager {
         typeof targetOrEntity === 'string')
         ? targetOrEntity
         : undefined;
+
     const entity = target ? entityOrMaybeOptions : targetOrEntity;
+
     const options = target
       ? maybeOptionsOrMaybePermissionOptions
       : entityOrMaybeOptions;
@@ -1094,7 +1096,7 @@ export class WorkspaceEntityManager extends EntityManager {
       {
         where: { id: In(entityIds) },
       },
-      permissionOptions,
+      { shouldBypassPermissionChecks: true }, // Bypass as this is for event emission
     );
 
     const beforeUpdateMapById = beforeUpdate.reduce(
@@ -1116,6 +1118,18 @@ export class WorkspaceEntityManager extends EntityManager {
       objectMetadataItem,
     );
 
+    const updatedColumns = formattedEntityOrEntities
+      .map((e) => Object.keys(e))
+      .flat();
+
+    this.validatePermissions({
+      target: targetOrEntity,
+      operationType: 'update',
+      permissionOptions: permissionOptionsFromArgs,
+      selectedColumns: [],
+      updatedColumns,
+    });
+
     const result = await new EntityPersistExecutor(
       this.connection,
       queryRunnerForEntityPersistExecutor,
@@ -1130,7 +1144,7 @@ export class WorkspaceEntityManager extends EntityManager {
 
     const resultArray = Array.isArray(result) ? result : [result];
 
-    const formattedResult = formatResult<Entity[]>(
+    let formattedResult = formatResult<Entity[]>(
       resultArray,
       objectMetadataItem,
       this.internalContext.objectMetadataMaps,
@@ -1157,9 +1171,79 @@ export class WorkspaceEntityManager extends EntityManager {
       }
     }
 
+    const isFieldPermissionsEnabled =
+      this.getFeatureFlagMap().IS_FIELDS_PERMISSIONS_ENABLED;
+
+    const permissionCheckApplies =
+      permissionOptionsFromArgs?.shouldBypassPermissionChecks !== true &&
+      objectMetadataItem.isSystem !== true;
+
+    if (isFieldPermissionsEnabled && permissionCheckApplies) {
+      formattedResult = this.getFormattedResultWithoutNonReadableFields({
+        formattedResult,
+        objectMetadataItem,
+        permissionOptionsFromArgs,
+      });
+    }
+
     return isEntityArray ? formattedResult : formattedResult[0];
   }
 
+  private getFormattedResultWithoutNonReadableFields<
+    Entity extends ObjectLiteral,
+  >({
+    formattedResult,
+    objectMetadataItem,
+    permissionOptionsFromArgs,
+  }: {
+    formattedResult: Entity[];
+    objectMetadataItem: ObjectMetadataItemWithFieldMaps;
+    permissionOptionsFromArgs: PermissionOptions | undefined;
+  }): Entity[] {
+    const restrictedFields =
+      permissionOptionsFromArgs?.objectRecordsPermissions?.[
+        objectMetadataItem.id
+      ].restrictedFields;
+
+    if (!restrictedFields) {
+      throw new InternalServerError('Restricted fields not found');
+    }
+
+    if (isEmpty(restrictedFields)) {
+      return formattedResult;
+    }
+
+    const objectMetadataItemWithFieldMaps =
+      this.internalContext.objectMetadataMaps.byId[objectMetadataItem.id];
+
+    const restrictedFieldNames = new Set(
+      Object.entries(restrictedFields)
+        .filter(([_, fieldPermissions]) => fieldPermissions.canRead === false)
+        .map(([fieldMetadataId]) => {
+          const fieldMetadata =
+            objectMetadataItemWithFieldMaps?.fieldsById[fieldMetadataId];
+
+          if (!isDefined(fieldMetadata)) {
+            throw new InternalServerError(
+              `Field metadata not found for field ${fieldMetadataId}`,
+            );
+          }
+
+          return fieldMetadata.name;
+        }),
+    );
+
+    const filteredResult = formattedResult.map((individualFormattedResult) => {
+      return Object.fromEntries(
+        Object.entries(individualFormattedResult).filter(
+          ([key]) => !restrictedFieldNames.has(key),
+        ),
+      );
+    });
+
+    return filteredResult as Entity[];
+  }
+
   override remove<Entity>(
     entity: Entity,
     options?: RemoveOptions,

packages/twenty-server/src/engine/twenty-orm/repository/permissions.utils.ts

@@ -15,7 +15,7 @@ import {
   PermissionsExceptionMessage,
 } from 'src/engine/metadata-modules/permissions/permissions.exception';
 import { ObjectMetadataMaps } from 'src/engine/metadata-modules/types/object-metadata-maps';
-import { getFieldMetadataIdForColumnNameMap } from 'src/engine/twenty-orm/utils/get-field-metadata-id-for-column-name.util';
+import { getColumnNameToFieldMetadataIdMap } from 'src/engine/twenty-orm/utils/get-column-name-to-field-metadata-id.util';
 
 const getTargetEntityAndOperationType = (expressionMap: QueryExpressionMap) => {
   const mainEntity = expressionMap.aliases[0].metadata.name;
@@ -43,6 +43,7 @@ export const validateOperationIsPermittedOrThrow = ({
   selectedColumns,
   isFieldPermissionsEnabled,
   allFieldsSelected,
+  updatedColumns,
 }: {
   entityName: string;
   operationType: OperationType;
@@ -51,6 +52,7 @@ export const validateOperationIsPermittedOrThrow = ({
   selectedColumns: string[];
   isFieldPermissionsEnabled?: boolean;
   allFieldsSelected: boolean;
+  updatedColumns: string[];
 }) => {
   const objectMetadataIdForEntity =
     objectMetadataMaps.idByNameSingular[entityName];
@@ -77,8 +79,8 @@ export const validateOperationIsPermittedOrThrow = ({
     return;
   }
 
-  const fieldMetadataIdForColumnNameMap = isFieldPermissionsEnabled
-    ? getFieldMetadataIdForColumnNameMap(objectMetadata)
+  const columnNameToFieldMetadataIdMap = isFieldPermissionsEnabled
+    ? getColumnNameToFieldMetadataIdMap(objectMetadata)
     : {};
 
   const permissionsForEntity =
@@ -97,7 +99,7 @@ export const validateOperationIsPermittedOrThrow = ({
         validateReadFieldPermissionOrThrow({
           restrictedFields: permissionsForEntity.restrictedFields,
           selectedColumns,
-          fieldMetadataIdForColumnNameMap,
+          columnNameToFieldMetadataIdMap,
           allFieldsSelected,
         });
       }
@@ -115,7 +117,15 @@ export const validateOperationIsPermittedOrThrow = ({
         validateReadFieldPermissionOrThrow({
           restrictedFields: permissionsForEntity.restrictedFields,
           selectedColumns,
-          fieldMetadataIdForColumnNameMap,
+          columnNameToFieldMetadataIdMap,
+        });
+      }
+
+      if (isFieldPermissionsEnabled && updatedColumns.length > 0) {
+        validateUpdateFieldPermissionOrThrow({
+          restrictedFields: permissionsForEntity.restrictedFields,
+          updatedColumns,
+          columnNameToFieldMetadataIdMap,
         });
       }
       break;
@@ -131,7 +141,7 @@ export const validateOperationIsPermittedOrThrow = ({
         validateReadFieldPermissionOrThrow({
           restrictedFields: permissionsForEntity.restrictedFields,
           selectedColumns,
-          fieldMetadataIdForColumnNameMap,
+          columnNameToFieldMetadataIdMap,
         });
       }
       break;
@@ -148,7 +158,7 @@ export const validateOperationIsPermittedOrThrow = ({
         validateReadFieldPermissionOrThrow({
           restrictedFields: permissionsForEntity.restrictedFields,
           selectedColumns,
-          fieldMetadataIdForColumnNameMap,
+          columnNameToFieldMetadataIdMap,
         });
       }
       break;
@@ -189,6 +199,7 @@ export const validateQueryIsPermittedOrThrow = ({
   );
 
   let selectedColumns: string[] = [];
+  let updatedColumns: string[] = [];
 
   if (isFieldPermissionsEnabled) {
     selectedColumns = getSelectedColumnsFromExpressionMap({
@@ -196,6 +207,26 @@ export const validateQueryIsPermittedOrThrow = ({
       expressionMap,
       allFieldsSelected,
     });
+
+    if (operationType !== 'select') {
+      const valuesSet = expressionMap.valuesSet;
+
+      if (Array.isArray(valuesSet)) {
+        updatedColumns = valuesSet.reduce((acc, value) => {
+          const keys = Object.keys(value);
+
+          keys.forEach((key) => {
+            if (!acc.includes(key)) {
+              acc.push(key);
+            }
+          });
+
+          return acc;
+        }, []);
+      } else {
+        updatedColumns = Object.keys(valuesSet ?? {});
+      }
+    }
   }
 
   validateOperationIsPermittedOrThrow({
@@ -203,21 +234,22 @@ export const validateQueryIsPermittedOrThrow = ({
     operationType: operationType as OperationType,
     objectRecordsPermissions,
     objectMetadataMaps,
-    selectedColumns: selectedColumns,
+    selectedColumns,
     isFieldPermissionsEnabled,
     allFieldsSelected,
+    updatedColumns,
   });
 };
 
 const validateReadFieldPermissionOrThrow = ({
   restrictedFields,
   selectedColumns,
-  fieldMetadataIdForColumnNameMap,
+  columnNameToFieldMetadataIdMap,
   allFieldsSelected,
 }: {
   restrictedFields: RestrictedFields;
   selectedColumns: string[];
-  fieldMetadataIdForColumnNameMap: Record<string, string>;
+  columnNameToFieldMetadataIdMap: Record<string, string>;
   allFieldsSelected?: boolean;
 }) => {
   if (isEmpty(restrictedFields)) {
@@ -232,7 +264,7 @@ const validateReadFieldPermissionOrThrow = ({
   }
 
   for (const column of selectedColumns) {
-    const fieldMetadataId = fieldMetadataIdForColumnNameMap[column];
+    const fieldMetadataId = columnNameToFieldMetadataIdMap[column];
 
     if (!fieldMetadataId) {
       throw new InternalServerError(
@@ -249,6 +281,37 @@ const validateReadFieldPermissionOrThrow = ({
   }
 };
 
+const validateUpdateFieldPermissionOrThrow = ({
+  restrictedFields,
+  updatedColumns,
+  columnNameToFieldMetadataIdMap,
+}: {
+  restrictedFields: RestrictedFields;
+  updatedColumns: string[];
+  columnNameToFieldMetadataIdMap: Record<string, string>;
+}) => {
+  if (isEmpty(restrictedFields)) {
+    return;
+  }
+
+  for (const column of updatedColumns) {
+    const fieldMetadataId = columnNameToFieldMetadataIdMap[column];
+
+    if (!fieldMetadataId) {
+      throw new InternalServerError(
+        `Field metadata id not found for column name ${column}`,
+      );
+    }
+
+    if (restrictedFields[fieldMetadataId]?.canUpdate === false) {
+      throw new PermissionsException(
+        PermissionsExceptionMessage.PERMISSION_DENIED,
+        PermissionsExceptionCode.PERMISSION_DENIED,
+      );
+    }
+  }
+};
+
 const getSelectedColumnsFromExpressionMap = ({
   operationType,
   expressionMap,

packages/twenty-server/src/engine/twenty-orm/utils/__tests__/get-column-name-to-field-metadata-id.util.spec.ts

@@ -0,0 +1,265 @@
+import { FieldMetadataType } from 'twenty-shared/types';
+
+import { ObjectMetadataItemWithFieldMaps } from 'src/engine/metadata-modules/types/object-metadata-item-with-field-maps';
+import { getColumnNameToFieldMetadataIdMap } from 'src/engine/twenty-orm/utils/get-column-name-to-field-metadata-id.util';
+
+describe('getColumnNameToFieldMetadataIdMap', () => {
+  const createMockObjectMetadataItemWithFieldMaps = (
+    fieldsById: Record<string, any>,
+  ): ObjectMetadataItemWithFieldMaps =>
+    ({
+      id: 'test-object-id',
+      nameSingular: 'test',
+      namePlural: 'tests',
+      labelSingular: 'Test',
+      labelPlural: 'Tests',
+      description: 'Test object',
+      icon: 'IconTest',
+      targetTableName: 'test',
+      isCustom: false,
+      isRemote: false,
+      isActive: true,
+      isSystem: false,
+      isAuditLogged: false,
+      isSearchable: false,
+      labelIdentifierFieldMetadataId: '',
+      imageIdentifierFieldMetadataId: '',
+      workspaceId: 'test-workspace-id',
+      indexMetadatas: [],
+      fieldsById,
+      fieldIdByName: {},
+      fieldIdByJoinColumnName: {},
+    }) as unknown as ObjectMetadataItemWithFieldMaps;
+
+  const createMockFieldMetadata = (
+    id: string,
+    name: string,
+    type: FieldMetadataType,
+  ) => ({
+    id,
+    name,
+    type,
+    label: name,
+    objectMetadataId: 'test-object-id',
+    isLabelSyncedWithName: true,
+    isNullable: true,
+    isUnique: false,
+    workspaceId: 'test-workspace-id',
+    createdAt: new Date(),
+    updatedAt: new Date(),
+  });
+
+  describe('with simple field types', () => {
+    it('should return a map with column name to field metadata id for simple field types', () => {
+      const fieldsById = {
+        'field-1': createMockFieldMetadata(
+          'field-1',
+          'name',
+          FieldMetadataType.TEXT,
+        ),
+        'field-2': createMockFieldMetadata(
+          'field-2',
+          'age',
+          FieldMetadataType.NUMBER,
+        ),
+      };
+
+      const objectMetadataItemWithFieldMaps =
+        createMockObjectMetadataItemWithFieldMaps(fieldsById);
+
+      const result = getColumnNameToFieldMetadataIdMap(
+        objectMetadataItemWithFieldMaps,
+      );
+
+      expect(result['name']).toBe('field-1');
+      expect(result['age']).toBe('field-2');
+      expect(Object.keys(result)).toHaveLength(2);
+    });
+  });
+
+  describe('with composite field types', () => {
+    it('should return column names to field metadata id for FULL_NAME composite type', () => {
+      const fieldsById = {
+        'field-1': createMockFieldMetadata(
+          'field-1',
+          'fullName',
+          FieldMetadataType.FULL_NAME,
+        ),
+      };
+
+      const objectMetadataItemWithFieldMaps =
+        createMockObjectMetadataItemWithFieldMaps(fieldsById);
+
+      const result = getColumnNameToFieldMetadataIdMap(
+        objectMetadataItemWithFieldMaps,
+      );
+
+      expect(result['fullNameFirstName']).toBe('field-1');
+      expect(result['fullNameLastName']).toBe('field-1');
+      expect(Object.keys(result)).toHaveLength(2);
+    });
+
+    it('should return column names to field metadata id for CURRENCY composite type', () => {
+      const fieldsById = {
+        'field-1': createMockFieldMetadata(
+          'field-1',
+          'price',
+          FieldMetadataType.CURRENCY,
+        ),
+      };
+
+      const objectMetadataItemWithFieldMaps =
+        createMockObjectMetadataItemWithFieldMaps(fieldsById);
+
+      const result = getColumnNameToFieldMetadataIdMap(
+        objectMetadataItemWithFieldMaps,
+      );
+
+      expect(result['priceAmountMicros']).toBe('field-1');
+      expect(result['priceCurrencyCode']).toBe('field-1');
+      expect(Object.keys(result)).toHaveLength(2);
+    });
+
+    it('should handle multiple composite fields', () => {
+      const fieldsById = {
+        'field-1': createMockFieldMetadata(
+          'field-1',
+          'fullName',
+          FieldMetadataType.FULL_NAME,
+        ),
+        'field-2': createMockFieldMetadata(
+          'field-2',
+          'price',
+          FieldMetadataType.CURRENCY,
+        ),
+        'field-3': createMockFieldMetadata(
+          'field-3',
+          'name',
+          FieldMetadataType.TEXT,
+        ),
+      };
+
+      const objectMetadataItemWithFieldMaps =
+        createMockObjectMetadataItemWithFieldMaps(fieldsById);
+
+      const result = getColumnNameToFieldMetadataIdMap(
+        objectMetadataItemWithFieldMaps,
+      );
+
+      expect(result['fullNameFirstName']).toBe('field-1');
+      expect(result['fullNameLastName']).toBe('field-1');
+      expect(result['priceAmountMicros']).toBe('field-2');
+      expect(result['priceCurrencyCode']).toBe('field-2');
+      expect(result['name']).toBe('field-3');
+      expect(Object.keys(result)).toHaveLength(5);
+    });
+  });
+
+  describe('with mixed field types', () => {
+    it('should handle both simple and composite field types', () => {
+      const fieldsById = {
+        'field-1': createMockFieldMetadata(
+          'field-1',
+          'name',
+          FieldMetadataType.TEXT,
+        ),
+        'field-2': createMockFieldMetadata(
+          'field-2',
+          'fullName',
+          FieldMetadataType.FULL_NAME,
+        ),
+        'field-3': createMockFieldMetadata(
+          'field-3',
+          'age',
+          FieldMetadataType.NUMBER,
+        ),
+        'field-4': createMockFieldMetadata(
+          'field-4',
+          'price',
+          FieldMetadataType.CURRENCY,
+        ),
+      };
+
+      const objectMetadataItemWithFieldMaps =
+        createMockObjectMetadataItemWithFieldMaps(fieldsById);
+
+      const result = getColumnNameToFieldMetadataIdMap(
+        objectMetadataItemWithFieldMaps,
+      );
+
+      expect(result['name']).toBe('field-1');
+      expect(result['fullNameFirstName']).toBe('field-2');
+      expect(result['fullNameLastName']).toBe('field-2');
+      expect(result['age']).toBe('field-3');
+      expect(result['priceAmountMicros']).toBe('field-4');
+      expect(result['priceCurrencyCode']).toBe('field-4');
+      expect(Object.keys(result)).toHaveLength(6);
+    });
+  });
+
+  describe('with relation field types', () => {
+    it('should handle relation field types with join column name', () => {
+      const fieldsById = {
+        'field-1': {
+          ...createMockFieldMetadata(
+            'field-1',
+            'company',
+            FieldMetadataType.RELATION,
+          ),
+          settings: {
+            relationType: 'ONE_TO_ONE',
+            joinColumnName: 'companyId',
+          },
+        },
+        'field-2': createMockFieldMetadata(
+          'field-2',
+          'name',
+          FieldMetadataType.TEXT,
+        ),
+      };
+
+      const objectMetadataItemWithFieldMaps =
+        createMockObjectMetadataItemWithFieldMaps(fieldsById);
+
+      const result = getColumnNameToFieldMetadataIdMap(
+        objectMetadataItemWithFieldMaps,
+      );
+
+      expect(result['companyId']).toBe('field-1');
+      expect(result['name']).toBe('field-2');
+      expect(Object.keys(result)).toHaveLength(2);
+    });
+
+    it('should skip ONE_TO_MANY relation field types', () => {
+      const fieldsById = {
+        'field-1': {
+          ...createMockFieldMetadata(
+            'field-1',
+            'employees',
+            FieldMetadataType.RELATION,
+          ),
+          settings: {
+            relationType: 'ONE_TO_MANY',
+            joinColumnName: 'companyId',
+          },
+        },
+        'field-2': createMockFieldMetadata(
+          'field-2',
+          'name',
+          FieldMetadataType.TEXT,
+        ),
+      };
+
+      const objectMetadataItemWithFieldMaps =
+        createMockObjectMetadataItemWithFieldMaps(fieldsById);
+
+      const result = getColumnNameToFieldMetadataIdMap(
+        objectMetadataItemWithFieldMaps,
+      );
+
+      expect(result['name']).toBe('field-2');
+      expect(result['companyId']).toBeUndefined();
+      expect(Object.keys(result)).toHaveLength(1);
+    });
+  });
+});

packages/twenty-server/src/engine/twenty-orm/utils/get-column-name-to-field-metadata-id.util.ts

@@ -0,0 +1,61 @@
+import { CompositeType } from 'src/engine/metadata-modules/field-metadata/interfaces/composite-type.interface';
+
+import { FieldMetadataEntity } from 'src/engine/metadata-modules/field-metadata/field-metadata.entity';
+import { computeCompositeColumnName } from 'src/engine/metadata-modules/field-metadata/utils/compute-column-name.util';
+import { ObjectMetadataItemWithFieldMaps } from 'src/engine/metadata-modules/types/object-metadata-item-with-field-maps';
+import {
+  ColumnNameProcessor,
+  processFieldMetadataForColumnNameMapping,
+} from 'src/engine/twenty-orm/utils/process-field-metadata-for-column-name-mapping.util';
+
+export function getColumnNameToFieldMetadataIdMap(
+  objectMetadataItemWithFieldMaps: ObjectMetadataItemWithFieldMaps,
+) {
+  const columnNameToFieldMetadataIdMap: Record<string, string> = {};
+
+  const processor: ColumnNameProcessor = {
+    processCompositeField: ({
+      fieldMetadataId,
+      fieldMetadata,
+      compositeType,
+    }: {
+      fieldMetadataId: string;
+      fieldMetadata: FieldMetadataEntity;
+      compositeType: CompositeType;
+    }) => {
+      compositeType.properties.forEach((compositeProperty) => {
+        const columnName = computeCompositeColumnName(
+          fieldMetadata.name,
+          compositeProperty,
+        );
+
+        columnNameToFieldMetadataIdMap[columnName] = fieldMetadataId;
+      });
+    },
+    processRelationField: ({
+      fieldMetadataId,
+      columnName,
+    }: {
+      fieldMetadataId: string;
+      columnName: string;
+    }) => {
+      columnNameToFieldMetadataIdMap[columnName] = fieldMetadataId;
+    },
+    processSimpleField: ({
+      fieldMetadataId,
+      columnName,
+    }: {
+      fieldMetadataId: string;
+      columnName: string;
+    }) => {
+      columnNameToFieldMetadataIdMap[columnName] = fieldMetadataId;
+    },
+  };
+
+  processFieldMetadataForColumnNameMapping(
+    objectMetadataItemWithFieldMaps,
+    processor,
+  );
+
+  return columnNameToFieldMetadataIdMap;
+}

packages/twenty-server/src/engine/twenty-orm/utils/get-field-metadata-id-for-column-name.util.ts

@@ -1,47 +0,0 @@
-import { FieldMetadataType } from 'twenty-shared/types';
-
-import { InternalServerError } from 'src/engine/core-modules/graphql/utils/graphql-errors.util';
-import { compositeTypeDefinitions } from 'src/engine/metadata-modules/field-metadata/composite-types';
-import {
-  computeColumnName,
-  computeCompositeColumnName,
-} from 'src/engine/metadata-modules/field-metadata/utils/compute-column-name.util';
-import { isCompositeFieldMetadataType } from 'src/engine/metadata-modules/field-metadata/utils/is-composite-field-metadata-type.util';
-import { ObjectMetadataItemWithFieldMaps } from 'src/engine/metadata-modules/types/object-metadata-item-with-field-maps';
-
-export function getFieldMetadataIdForColumnNameMap(
-  objectMetadataItemWithFieldMaps: ObjectMetadataItemWithFieldMaps,
-) {
-  const columnNameToFieldMetadataIdMap: Record<string, string> = {};
-
-  for (const [fieldMetadataId, fieldMetadata] of Object.entries(
-    objectMetadataItemWithFieldMaps.fieldsById,
-  )) {
-    if (isCompositeFieldMetadataType(fieldMetadata.type)) {
-      const compositeType = compositeTypeDefinitions.get(fieldMetadata.type);
-
-      if (!compositeType) {
-        throw new InternalServerError(
-          `Composite type not found for field metadata type ${fieldMetadata.type}`,
-        );
-      }
-
-      compositeType.properties.forEach((compositeProperty) => {
-        const columnName = computeCompositeColumnName(
-          fieldMetadata.name,
-          compositeProperty,
-        );
-
-        columnNameToFieldMetadataIdMap[columnName] = fieldMetadataId;
-      });
-    } else {
-      const columnName = computeColumnName(fieldMetadata, {
-        isForeignKey: fieldMetadata.type === FieldMetadataType.RELATION,
-      });
-
-      columnNameToFieldMetadataIdMap[columnName] = fieldMetadataId;
-    }
-  }
-
-  return columnNameToFieldMetadataIdMap;
-}

packages/twenty-server/src/engine/twenty-orm/utils/get-field-metadata-id-to-column-names-map.util.ts

@@ -1,29 +1,28 @@
-import { InternalServerError } from 'src/engine/core-modules/graphql/utils/graphql-errors.util';
-import { compositeTypeDefinitions } from 'src/engine/metadata-modules/field-metadata/composite-types';
-import {
-  computeColumnName,
-  computeCompositeColumnName,
-} from 'src/engine/metadata-modules/field-metadata/utils/compute-column-name.util';
-import { isCompositeFieldMetadataType } from 'src/engine/metadata-modules/field-metadata/utils/is-composite-field-metadata-type.util';
+import { CompositeType } from 'src/engine/metadata-modules/field-metadata/interfaces/composite-type.interface';
+
+import { FieldMetadataEntity } from 'src/engine/metadata-modules/field-metadata/field-metadata.entity';
+import { computeCompositeColumnName } from 'src/engine/metadata-modules/field-metadata/utils/compute-column-name.util';
 import { ObjectMetadataItemWithFieldMaps } from 'src/engine/metadata-modules/types/object-metadata-item-with-field-maps';
+import {
+  ColumnNameProcessor,
+  processFieldMetadataForColumnNameMapping,
+} from 'src/engine/twenty-orm/utils/process-field-metadata-for-column-name-mapping.util';
 
 export function getFieldMetadataIdToColumnNamesMap(
   objectMetadataItemWithFieldMaps: ObjectMetadataItemWithFieldMaps,
 ) {
   const fieldMetadataToColumnNamesMap = new Map<string, string[]>();
 
-  for (const [fieldMetadataId, fieldMetadata] of Object.entries(
-    objectMetadataItemWithFieldMaps.fieldsById,
-  )) {
-    if (isCompositeFieldMetadataType(fieldMetadata.type)) {
-      const compositeType = compositeTypeDefinitions.get(fieldMetadata.type);
-
-      if (!compositeType) {
-        throw new InternalServerError(
-          `Composite type not found for field metadata type ${fieldMetadata.type}`,
-        );
-      }
-
+  const processor: ColumnNameProcessor = {
+    processCompositeField: ({
+      fieldMetadataId,
+      fieldMetadata,
+      compositeType,
+    }: {
+      fieldMetadataId: string;
+      fieldMetadata: FieldMetadataEntity;
+      compositeType: CompositeType;
+    }) => {
       compositeType.properties.forEach((compositeProperty) => {
         const columnName = computeCompositeColumnName(
           fieldMetadata.name,
@@ -38,12 +37,33 @@ export function getFieldMetadataIdToColumnNamesMap(
           columnName,
         ]);
       });
-    } else {
-      const columnName = computeColumnName(fieldMetadata);
-
+    },
+    processRelationField: ({
+      fieldMetadataId,
+      columnName,
+    }: {
+      fieldMetadataId: string;
+      fieldMetadata: FieldMetadataEntity;
+      columnName: string;
+    }) => {
+      fieldMetadataToColumnNamesMap.set(fieldMetadataId, [columnName]); // TODO test
+    },
+    processSimpleField: ({
+      fieldMetadataId,
+      columnName,
+    }: {
+      fieldMetadataId: string;
+      fieldMetadata: FieldMetadataEntity;
+      columnName: string;
+    }) => {
       fieldMetadataToColumnNamesMap.set(fieldMetadataId, [columnName]);
-    }
-  }
+    },
+  };
+
+  processFieldMetadataForColumnNameMapping(
+    objectMetadataItemWithFieldMaps,
+    processor,
+  );
 
   return fieldMetadataToColumnNamesMap;
 }

packages/twenty-server/src/engine/twenty-orm/utils/process-field-metadata-for-column-name-mapping.util.ts

@@ -0,0 +1,100 @@
+import { CompositeType } from 'src/engine/metadata-modules/field-metadata/interfaces/composite-type.interface';
+import { FieldMetadataRelationSettings } from 'src/engine/metadata-modules/field-metadata/interfaces/field-metadata-settings.interface';
+import { RelationType } from 'src/engine/metadata-modules/field-metadata/interfaces/relation-type.interface';
+
+import { isFieldMetadataRelationOrMorphRelation } from 'src/engine/api/graphql/workspace-schema-builder/utils/is-field-metadata-relation-or-morph-relation.utils';
+import { compositeTypeDefinitions } from 'src/engine/metadata-modules/field-metadata/composite-types';
+import { FieldMetadataEntity } from 'src/engine/metadata-modules/field-metadata/field-metadata.entity';
+import { computeColumnName } from 'src/engine/metadata-modules/field-metadata/utils/compute-column-name.util';
+import { isCompositeFieldMetadataType } from 'src/engine/metadata-modules/field-metadata/utils/is-composite-field-metadata-type.util';
+import {
+  PermissionsException,
+  PermissionsExceptionCode,
+} from 'src/engine/metadata-modules/permissions/permissions.exception';
+import { ObjectMetadataItemWithFieldMaps } from 'src/engine/metadata-modules/types/object-metadata-item-with-field-maps';
+
+export type ColumnNameProcessor = {
+  processCompositeField: ({
+    fieldMetadataId,
+    fieldMetadata,
+    compositeType,
+  }: {
+    fieldMetadataId: string;
+    fieldMetadata: FieldMetadataEntity;
+    compositeType: CompositeType;
+  }) => void;
+  processRelationField: ({
+    fieldMetadataId,
+    fieldMetadata,
+    columnName,
+  }: {
+    fieldMetadataId: string;
+    fieldMetadata: FieldMetadataEntity;
+    columnName: string;
+  }) => void;
+  processSimpleField: ({
+    fieldMetadataId,
+    fieldMetadata,
+    columnName,
+  }: {
+    fieldMetadataId: string;
+    fieldMetadata: FieldMetadataEntity;
+    columnName: string;
+  }) => void;
+};
+
+export function processFieldMetadataForColumnNameMapping(
+  objectMetadataItemWithFieldMaps: ObjectMetadataItemWithFieldMaps,
+  processor: ColumnNameProcessor,
+) {
+  for (const [fieldMetadataId, fieldMetadata] of Object.entries(
+    objectMetadataItemWithFieldMaps.fieldsById,
+  )) {
+    if (isCompositeFieldMetadataType(fieldMetadata.type)) {
+      const compositeType = compositeTypeDefinitions.get(fieldMetadata.type);
+
+      if (!compositeType) {
+        throw new PermissionsException(
+          `Composite type not found for field metadata type ${fieldMetadata.type}`,
+          PermissionsExceptionCode.COMPOSITE_TYPE_NOT_FOUND,
+        );
+      }
+
+      processor.processCompositeField({
+        fieldMetadataId,
+        fieldMetadata,
+        compositeType,
+      });
+    } else {
+      if (isFieldMetadataRelationOrMorphRelation(fieldMetadata)) {
+        const fieldMetadataSettings =
+          fieldMetadata.settings as FieldMetadataRelationSettings;
+
+        if (fieldMetadataSettings?.relationType === RelationType.ONE_TO_MANY) {
+          continue;
+        }
+        const columnName = fieldMetadataSettings?.joinColumnName;
+
+        if (!columnName) {
+          throw new PermissionsException(
+            `Join column name is required for relation field metadata ${fieldMetadata.name}`,
+            PermissionsExceptionCode.JOIN_COLUMN_NAME_REQUIRED,
+          );
+        }
+        processor.processRelationField({
+          fieldMetadataId,
+          fieldMetadata,
+          columnName,
+        });
+      } else {
+        const columnName = computeColumnName(fieldMetadata);
+
+        processor.processSimpleField({
+          fieldMetadataId,
+          fieldMetadata,
+          columnName,
+        });
+      }
+    }
+  }
+}