Fix ORM event mixing records in batch updates (#15985)

While investigating the issue a customer was facing, I discovered that
before records and after records could be in different order, making the
orm event and timeline activity engine mix records

packages/twenty-server/src/engine/twenty-orm/utils/__tests__/format-twenty-orm-event-to-database-batch-event.util.spec.ts

@@ -0,0 +1,171 @@
+import { FieldMetadataType } from 'twenty-shared/types';
+
+import { DatabaseEventAction } from 'src/engine/api/graphql/graphql-query-runner/enums/database-event-action';
+import { type ObjectRecordUpdateEvent } from 'src/engine/core-modules/event-emitter/types/object-record-update.event';
+import {
+  TwentyORMException,
+  TwentyORMExceptionCode,
+} from 'src/engine/twenty-orm/exceptions/twenty-orm.exception';
+import { formatTwentyOrmEventToDatabaseBatchEvent } from 'src/engine/twenty-orm/utils/format-twenty-orm-event-to-database-batch-event.util';
+import { getMockFieldMetadataEntity } from 'src/utils/__test__/get-field-metadata-entity.mock';
+import { getMockObjectMetadataItemWithFieldsMaps } from 'src/utils/__test__/get-object-metadata-item-with-fields-maps.mock';
+
+describe('formatTwentyOrmEventToDatabaseBatchEvent', () => {
+  const objectMetadataItemWithFieldMaps =
+    getMockObjectMetadataItemWithFieldsMaps({
+      id: 'object-id',
+      workspaceId: 'workspace-id',
+      nameSingular: 'person',
+      namePlural: 'people',
+      indexMetadatas: [],
+      fieldIdByJoinColumnName: {},
+      fieldIdByName: {
+        name: 'name-id',
+        age: 'age-id',
+        fullName: 'fullname-id',
+      },
+      fieldsById: {
+        'name-id': getMockFieldMetadataEntity({
+          workspaceId: 'workspace-id',
+          objectMetadataId: 'object-id',
+          id: 'name-id',
+          type: FieldMetadataType.TEXT,
+          name: 'name',
+          label: 'Name',
+          isLabelSyncedWithName: true,
+          isNullable: true,
+          createdAt: new Date(),
+          updatedAt: new Date(),
+        }),
+      },
+    });
+
+  const mockWorkspaceId = 'workspace-id';
+  const mockAuthContext = {
+    user: { id: 'user-id' },
+    workspaceMemberId: 'workspace-member-id',
+  } as any;
+
+  describe('UPDATED action', () => {
+    it('should throw TwentyORMException when no matching before entity is found in array of beforeEntities', () => {
+      const afterEntities = [
+        {
+          id: 'record-1',
+          name: 'John Doe Updated',
+        },
+        {
+          id: 'record-2',
+          name: 'Jane Doe Updated',
+        },
+      ];
+
+      const beforeEntities = [
+        {
+          id: 'record-1',
+          name: 'John Doe',
+        },
+        {
+          id: 'record-3',
+          name: 'Bob Smith',
+        },
+      ];
+
+      try {
+        formatTwentyOrmEventToDatabaseBatchEvent({
+          action: DatabaseEventAction.UPDATED,
+          objectMetadataItem: objectMetadataItemWithFieldMaps,
+          workspaceId: mockWorkspaceId,
+          authContext: mockAuthContext,
+          entities: afterEntities,
+          beforeEntities: beforeEntities,
+        });
+      } catch (error) {
+        expect(error).toBeInstanceOf(TwentyORMException);
+        expect((error as TwentyORMException).code).toBe(
+          TwentyORMExceptionCode.ORM_EVENT_DATA_CORRUPTED,
+        );
+        expect((error as TwentyORMException).message).toBe(
+          'Record mismatch detected while computing event data for UPDATED action',
+        );
+      }
+    });
+
+    it('should successfully create update events when matching before entities are found and in the right order', () => {
+      const afterEntities = [
+        {
+          id: 'record-1',
+          name: 'John Doe Updated',
+        },
+        {
+          id: 'record-2',
+          name: 'Jane Doe Updated',
+        },
+      ];
+
+      const beforeEntities = [
+        {
+          id: 'record-2',
+          name: 'Jane Doe',
+        },
+        {
+          id: 'record-1',
+          name: 'John Doe',
+        },
+      ];
+
+      const result = formatTwentyOrmEventToDatabaseBatchEvent({
+        action: DatabaseEventAction.UPDATED,
+        objectMetadataItem: objectMetadataItemWithFieldMaps,
+        workspaceId: mockWorkspaceId,
+        authContext: mockAuthContext,
+        entities: afterEntities,
+        beforeEntities: beforeEntities,
+      });
+
+      expect(result).toBeDefined();
+      expect(result?.action).toBe(DatabaseEventAction.UPDATED);
+      expect(result?.events).toHaveLength(2);
+
+      const updateEvent1 = result?.events[0] as ObjectRecordUpdateEvent<
+        (typeof afterEntities)[0]
+      >;
+      const updateEvent2 = result?.events[1] as ObjectRecordUpdateEvent<
+        (typeof afterEntities)[1]
+      >;
+
+      expect(updateEvent1.recordId).toBe('record-1');
+      expect(updateEvent2.recordId).toBe('record-2');
+
+      expect(updateEvent1.properties?.before?.name).toBe('John Doe');
+      expect(updateEvent1.properties?.after?.name).toBe('John Doe Updated');
+      expect(updateEvent2.properties?.before?.name).toBe('Jane Doe');
+      expect(updateEvent2.properties?.after?.name).toBe('Jane Doe Updated');
+    });
+
+    it('should handle single entity (non-array) for both before and after', () => {
+      const afterEntity = {
+        id: 'record-1',
+        name: 'John Doe Updated',
+      };
+
+      const beforeEntity = {
+        id: 'record-1',
+        name: 'John Doe',
+      };
+
+      const result = formatTwentyOrmEventToDatabaseBatchEvent({
+        action: DatabaseEventAction.UPDATED,
+        objectMetadataItem: objectMetadataItemWithFieldMaps,
+        workspaceId: mockWorkspaceId,
+        authContext: mockAuthContext,
+        entities: afterEntity,
+        beforeEntities: beforeEntity,
+      });
+
+      expect(result).toBeDefined();
+      expect(result?.action).toBe(DatabaseEventAction.UPDATED);
+      expect(result?.events).toHaveLength(1);
+      expect(result?.events[0].recordId).toBe('record-1');
+    });
+  });
+});

packages/twenty-server/src/engine/twenty-orm/utils/compute-event-select-query-builder.util.ts

@@ -0,0 +1,50 @@
+import { type ObjectsPermissions } from 'twenty-shared/types';
+import { type ObjectLiteral } from 'typeorm';
+import { type QueryExpressionMap } from 'typeorm/query-builder/QueryExpressionMap';
+
+import { type FeatureFlagMap } from 'src/engine/core-modules/feature-flag/interfaces/feature-flag-map.interface';
+import { type WorkspaceInternalContext } from 'src/engine/twenty-orm/interfaces/workspace-internal-context.interface';
+
+import { type AuthContext } from 'src/engine/core-modules/auth/types/auth-context.type';
+import { type WorkspaceDeleteQueryBuilder } from 'src/engine/twenty-orm/repository/workspace-delete-query-builder';
+import { type WorkspaceInsertQueryBuilder } from 'src/engine/twenty-orm/repository/workspace-insert-query-builder';
+import { WorkspaceSelectQueryBuilder } from 'src/engine/twenty-orm/repository/workspace-select-query-builder';
+import { type WorkspaceSoftDeleteQueryBuilder } from 'src/engine/twenty-orm/repository/workspace-soft-delete-query-builder';
+import { type WorkspaceUpdateQueryBuilder } from 'src/engine/twenty-orm/repository/workspace-update-query-builder';
+
+type EventSelectQueryBuilderFactoryArgs<T extends ObjectLiteral> = {
+  queryBuilder:
+    | WorkspaceUpdateQueryBuilder<T>
+    | WorkspaceSoftDeleteQueryBuilder<T>
+    | WorkspaceDeleteQueryBuilder<T>
+    | WorkspaceInsertQueryBuilder<T>;
+  authContext: AuthContext;
+  internalContext: WorkspaceInternalContext;
+  featureFlagMap: FeatureFlagMap;
+  expressionMap: QueryExpressionMap;
+  objectRecordsPermissions: ObjectsPermissions;
+};
+
+export const computeEventSelectQueryBuilder = <T extends ObjectLiteral>({
+  queryBuilder,
+  authContext,
+  featureFlagMap,
+  internalContext,
+  expressionMap,
+  objectRecordsPermissions,
+}: EventSelectQueryBuilderFactoryArgs<T>): WorkspaceSelectQueryBuilder<T> => {
+  const eventSelectQueryBuilder = new WorkspaceSelectQueryBuilder(
+    queryBuilder as unknown as WorkspaceSelectQueryBuilder<T>,
+    objectRecordsPermissions,
+    internalContext,
+    true,
+    authContext,
+    featureFlagMap,
+  );
+
+  eventSelectQueryBuilder.expressionMap.wheres = expressionMap.wheres;
+  eventSelectQueryBuilder.expressionMap.aliases = expressionMap.aliases;
+  eventSelectQueryBuilder.setParameters(expressionMap.parameters);
+
+  return eventSelectQueryBuilder;
+};

packages/twenty-server/src/engine/twenty-orm/utils/format-twenty-orm-event-to-database-batch-event.util.ts

@@ -1,5 +1,5 @@
-import { isDefined } from 'twenty-shared/utils';
 import { STANDARD_OBJECT_IDS } from 'twenty-shared/metadata';
+import { isDefined } from 'twenty-shared/utils';
 
 import type { ObjectLiteral } from 'typeorm';
 
@@ -13,6 +13,10 @@ import { ObjectRecordUpdateEvent } from 'src/engine/core-modules/event-emitter/t
 import { ObjectRecordUpsertEvent } from 'src/engine/core-modules/event-emitter/types/object-record-upsert.event';
 import { objectRecordChangedValues } from 'src/engine/core-modules/event-emitter/utils/object-record-changed-values';
 import type { ObjectMetadataItemWithFieldMaps } from 'src/engine/metadata-modules/types/object-metadata-item-with-field-maps';
+import {
+  TwentyORMException,
+  TwentyORMExceptionCode,
+} from 'src/engine/twenty-orm/exceptions/twenty-orm.exception';
 import { type DatabaseBatchEventInput } from 'src/engine/workspace-event-emitter/workspace-event-emitter';
 
 export const formatTwentyOrmEventToDatabaseBatchEvent = <
@@ -65,15 +69,22 @@ export const formatTwentyOrmEventToDatabaseBatchEvent = <
       break;
     case DatabaseEventAction.UPDATED:
       events = entityArray
-        .map((after, idx) => {
+        .map((after) => {
           if (!beforeEntities) {
             throw new Error('beforeEntities is required for UPDATED action');
           }
 
           const before = Array.isArray(beforeEntities)
-            ? beforeEntities?.[idx]
+            ? beforeEntities.find((before) => before.id === after.id)
             : beforeEntities;
 
+          if (!isDefined(before)) {
+            throw new TwentyORMException(
+              'Record mismatch detected while computing event data for UPDATED action',
+              TwentyORMExceptionCode.ORM_EVENT_DATA_CORRUPTED,
+            );
+          }
+
           const diff = objectRecordChangedValues(
             before,
             after,
@@ -127,7 +138,7 @@ export const formatTwentyOrmEventToDatabaseBatchEvent = <
       });
       break;
     case DatabaseEventAction.UPSERTED:
-      events = entityArray.map((after, index) => {
+      events = entityArray.map((after) => {
         const event = new ObjectRecordUpsertEvent<T>();
 
         event.userId = authContext?.user?.id;
@@ -136,7 +147,7 @@ export const formatTwentyOrmEventToDatabaseBatchEvent = <
 
         const before = beforeEntities
           ? Array.isArray(beforeEntities)
-            ? beforeEntities[index]
+            ? beforeEntities.find((before) => before.id === after.id)
             : beforeEntities
           : undefined;
 

packages/twenty-server/src/engine/twenty-orm/utils/twenty-orm-graphql-api-exception-handler.util.ts

@@ -13,6 +13,7 @@ export const twentyORMGraphqlApiExceptionHandler = (
     case TwentyORMExceptionCode.CONNECT_RECORD_NOT_FOUND:
     case TwentyORMExceptionCode.CONNECT_NOT_ALLOWED:
     case TwentyORMExceptionCode.CONNECT_UNIQUE_CONSTRAINT_ERROR:
+    case TwentyORMExceptionCode.TOO_MANY_RECORDS_TO_UPDATE:
       throw new UserInputError(error.message, {
         userFriendlyMessage: error.userFriendlyMessage,
       });