Emit proper event on createOrUpdate csv import operation (#12163)

- use proper event emitter when upserting records with csv import
- After:


https://github.com/user-attachments/assets/8303da38-2e35-4f4c-bb13-8a7a222971b7

packages/twenty-server/src/engine/api/graphql/graphql-query-runner/resolvers/graphql-query-create-many-resolver.service.ts

@@ -1,6 +1,6 @@
 import { Injectable } from '@nestjs/common';
 
-import { capitalize } from 'twenty-shared/utils';
+import { capitalize, isDefined } from 'twenty-shared/utils';
 import { In, InsertResult, ObjectLiteral } from 'typeorm';
 
 import {
@@ -21,6 +21,7 @@ import { ObjectMetadataMaps } from 'src/engine/metadata-modules/types/object-met
 import { WorkspaceRepository } from 'src/engine/twenty-orm/repository/workspace.repository';
 import { formatData } from 'src/engine/twenty-orm/utils/format-data.util';
 import { formatResult } from 'src/engine/twenty-orm/utils/format-result.util';
+import { AuthContext } from 'src/engine/core-modules/auth/types/auth-context.type';
 
 @Injectable()
 export class GraphqlQueryCreateManyResolverService extends GraphqlQueryBaseResolverService<
@@ -30,7 +31,7 @@ export class GraphqlQueryCreateManyResolverService extends GraphqlQueryBaseResol
   async resolve(
     executionArgs: GraphqlQueryResolverExecutionArgs<CreateManyResolverArgs>,
   ): Promise<ObjectRecord[]> {
-    const { authContext, objectMetadataItemWithFieldMaps, objectMetadataMaps } =
+    const { objectMetadataItemWithFieldMaps, objectMetadataMaps } =
       executionArgs.options;
 
     const { roleId } = executionArgs;
@@ -44,12 +45,6 @@ export class GraphqlQueryCreateManyResolverService extends GraphqlQueryBaseResol
       objectMetadataMaps,
     );
 
-    this.apiEventEmitterService.emitCreateEvents(
-      upsertedRecords,
-      authContext,
-      objectMetadataItemWithFieldMaps,
-    );
-
     const shouldBypassPermissionChecks = executionArgs.isExecutedByApiKey;
 
     await this.processNestedRelationsIfNeeded(
@@ -102,18 +97,24 @@ export class GraphqlQueryCreateManyResolverService extends GraphqlQueryBaseResol
       raw: [],
     };
 
-    await this.processRecordsToUpdate(
+    await this.processRecordsToUpdate({
-      recordsToUpdate,
+      partialRecordsToUpdate: recordsToUpdate,
-      executionArgs.repository,
+      existingRecords,
+      repository: executionArgs.repository,
       objectMetadataItemWithFieldMaps,
+      objectMetadataMaps: executionArgs.options.objectMetadataMaps,
       result,
-    );
+      authContext: executionArgs.options.authContext,
+    });
 
-    await this.processRecordsToInsert(
+    await this.processRecordsToInsert({
       recordsToInsert,
-      executionArgs.repository,
+      repository: executionArgs.repository,
       result,
-    );
+      objectMetadataItemWithFieldMaps,
+      objectMetadataMaps: executionArgs.options.objectMetadataMaps,
+      authContext: executionArgs.options.authContext,
+    });
 
     return result;
   }
@@ -259,47 +260,116 @@ export class GraphqlQueryCreateManyResolverService extends GraphqlQueryBaseResol
     return { recordsToUpdate, recordsToInsert };
   }
 
-  private async processRecordsToUpdate(
+  private async processRecordsToUpdate({
-    recordsToUpdate: Partial<ObjectRecord>[],
+    partialRecordsToUpdate,
-    repository: WorkspaceRepository<ObjectLiteral>,
+    existingRecords,
-    objectMetadataItemWithFieldMaps: ObjectMetadataItemWithFieldMaps,
+    repository,
-    result: InsertResult,
+    objectMetadataItemWithFieldMaps,
-  ): Promise<void> {
+    objectMetadataMaps,
-    for (const record of recordsToUpdate) {
+    result,
-      const recordId = record.id as string;
+    authContext,
+  }: {
+    partialRecordsToUpdate: Partial<ObjectRecord>[];
+    existingRecords: Partial<ObjectRecord>[];
+    repository: WorkspaceRepository<ObjectLiteral>;
+    objectMetadataItemWithFieldMaps: ObjectMetadataItemWithFieldMaps;
+    objectMetadataMaps: ObjectMetadataMaps;
+    result: InsertResult;
+    authContext: AuthContext;
+  }): Promise<void> {
+    for (const partialRecordToUpdate of partialRecordsToUpdate) {
+      const recordId = partialRecordToUpdate.id as string;
 
       // we should not update an existing record's createdBy value
-      const recordWithoutCreatedByUpdate = this.getRecordWithoutCreatedBy(
+      const partialRecordToUpdateWithoutCreatedByUpdate =
-        record,
+        this.getRecordWithoutCreatedBy(
-        objectMetadataItemWithFieldMaps,
+          partialRecordToUpdate,
-      );
+          objectMetadataItemWithFieldMaps,
+        );
 
-      const formattedRecord = formatData(
+      const formattedPartialRecordToUpdate = formatData(
-        recordWithoutCreatedByUpdate,
+        partialRecordToUpdateWithoutCreatedByUpdate,
         objectMetadataItemWithFieldMaps,
       );
 
       // TODO: we should align update and insert
       // For insert, formating is done in the server
       // While for update, formatting is done at the resolver level
-      await repository.update(recordId, formattedRecord);
+      await repository.update(recordId, formattedPartialRecordToUpdate);
+
       result.identifiers.push({ id: recordId });
       result.generatedMaps.push({ id: recordId });
+
+      const [updatedRecord] = await repository.find({
+        where: { id: recordId },
+      });
+
+      if (!isDefined(updatedRecord)) {
+        continue;
+      }
+
+      const record = formatResult<ObjectRecord>(
+        updatedRecord,
+        objectMetadataItemWithFieldMaps,
+        objectMetadataMaps,
+      );
+
+      const existingRecord = formatResult<ObjectRecord>(
+        existingRecords.find((record) => record.id === recordId),
+        objectMetadataItemWithFieldMaps,
+        objectMetadataMaps,
+      );
+
+      this.apiEventEmitterService.emitUpdateEvents({
+        existingRecords: [existingRecord],
+        records: [record],
+        updatedFields: Object.keys(formattedPartialRecordToUpdate),
+        authContext,
+        objectMetadataItem: objectMetadataItemWithFieldMaps,
+      });
     }
   }
 
-  private async processRecordsToInsert(
+  private async processRecordsToInsert({
-    recordsToInsert: Partial<ObjectRecord>[],
+    recordsToInsert,
-    repository: WorkspaceRepository<ObjectLiteral>,
+    repository,
-    result: InsertResult,
+    objectMetadataItemWithFieldMaps,
-  ): Promise<void> {
+    objectMetadataMaps,
+    result,
+    authContext,
+  }: {
+    recordsToInsert: Partial<ObjectRecord>[];
+    repository: WorkspaceRepository<ObjectLiteral>;
+    objectMetadataItemWithFieldMaps: ObjectMetadataItemWithFieldMaps;
+    objectMetadataMaps: ObjectMetadataMaps;
+    result: InsertResult;
+    authContext: AuthContext;
+  }): Promise<void> {
+    const formattedInsertedRecords: ObjectRecord[] = [];
+
     if (recordsToInsert.length > 0) {
       const insertResult = await repository.insert(recordsToInsert);
 
       result.identifiers.push(...insertResult.identifiers);
       result.generatedMaps.push(...insertResult.generatedMaps);
       result.raw.push(...insertResult.raw);
+
+      formattedInsertedRecords.push(
+        ...insertResult.raw.map((record: ObjectRecord) =>
+          formatResult<ObjectRecord>(
+            record,
+            objectMetadataItemWithFieldMaps,
+            objectMetadataMaps,
+          ),
+        ),
+      );
     }
+
+    this.apiEventEmitterService.emitCreateEvents({
+      records: formattedInsertedRecords,
+      authContext,
+      objectMetadataItem: objectMetadataItemWithFieldMaps,
+    });
   }
 
   private async fetchUpsertedRecords(

packages/twenty-server/src/engine/api/graphql/graphql-query-runner/resolvers/graphql-query-create-one-resolver.service.ts

@@ -53,11 +53,11 @@ export class GraphqlQueryCreateOneResolverService extends GraphqlQueryBaseResolv
       objectMetadataMaps,
     );
 
-    this.apiEventEmitterService.emitCreateEvents(
+    this.apiEventEmitterService.emitCreateEvents({
-      upsertedRecords,
+      records: upsertedRecords,
       authContext,
-      objectMetadataItemWithFieldMaps,
+      objectMetadataItem: objectMetadataItemWithFieldMaps,
-    );
+    });
 
     if (executionArgs.graphqlQuerySelectedFieldsResult.relations) {
       await this.processNestedRelationsHelper.processNestedRelations({

packages/twenty-server/src/engine/api/graphql/graphql-query-runner/resolvers/graphql-query-delete-many-resolver.service.ts

@@ -54,11 +54,11 @@ export class GraphqlQueryDeleteManyResolverService extends GraphqlQueryBaseResol
       objectMetadataMaps,
     );
 
-    this.apiEventEmitterService.emitDeletedEvents(
+    this.apiEventEmitterService.emitDeletedEvents({
-      formattedDeletedRecords,
+      records: formattedDeletedRecords,
       authContext,
-      objectMetadataItemWithFieldMaps,
+      objectMetadataItem: objectMetadataItemWithFieldMaps,
-    );
+    });
 
     if (executionArgs.graphqlQuerySelectedFieldsResult.relations) {
       await this.processNestedRelationsHelper.processNestedRelations({

packages/twenty-server/src/engine/api/graphql/graphql-query-runner/resolvers/graphql-query-delete-one-resolver.service.ts

@@ -47,11 +47,11 @@ export class GraphqlQueryDeleteOneResolverService extends GraphqlQueryBaseResolv
       objectMetadataMaps,
     );
 
-    this.apiEventEmitterService.emitDeletedEvents(
+    this.apiEventEmitterService.emitDeletedEvents({
-      formattedDeletedRecords,
+      records: formattedDeletedRecords,
       authContext,
-      objectMetadataItemWithFieldMaps,
+      objectMetadataItem: objectMetadataItemWithFieldMaps,
-    );
+    });
 
     if (formattedDeletedRecords.length === 0) {
       throw new GraphqlQueryRunnerException(

packages/twenty-server/src/engine/api/graphql/graphql-query-runner/resolvers/graphql-query-destroy-many-resolver.service.ts

@@ -52,11 +52,11 @@ export class GraphqlQueryDestroyManyResolverService extends GraphqlQueryBaseReso
       objectMetadataMaps,
     );
 
-    this.apiEventEmitterService.emitDestroyEvents(
+    this.apiEventEmitterService.emitDestroyEvents({
-      deletedRecords,
+      records: deletedRecords,
       authContext,
-      objectMetadataItemWithFieldMaps,
+      objectMetadataItem: objectMetadataItemWithFieldMaps,
-    );
+    });
 
     if (executionArgs.graphqlQuerySelectedFieldsResult.relations) {
       await this.processNestedRelationsHelper.processNestedRelations({

packages/twenty-server/src/engine/api/graphql/graphql-query-runner/resolvers/graphql-query-destroy-one-resolver.service.ts

@@ -52,11 +52,11 @@ export class GraphqlQueryDestroyOneResolverService extends GraphqlQueryBaseResol
       objectMetadataMaps,
     );
 
-    this.apiEventEmitterService.emitDestroyEvents(
+    this.apiEventEmitterService.emitDestroyEvents({
-      deletedRecords,
+      records: deletedRecords,
       authContext,
-      objectMetadataItemWithFieldMaps,
+      objectMetadataItem: objectMetadataItemWithFieldMaps,
-    );
+    });
 
     if (executionArgs.graphqlQuerySelectedFieldsResult.relations) {
       await this.processNestedRelationsHelper.processNestedRelations({

packages/twenty-server/src/engine/api/graphql/graphql-query-runner/resolvers/graphql-query-restore-many-resolver.service.ts

@@ -54,11 +54,11 @@ export class GraphqlQueryRestoreManyResolverService extends GraphqlQueryBaseReso
       objectMetadataMaps,
     );
 
-    this.apiEventEmitterService.emitRestoreEvents(
+    this.apiEventEmitterService.emitRestoreEvents({
-      formattedRestoredRecords,
+      records: formattedRestoredRecords,
       authContext,
-      objectMetadataItemWithFieldMaps,
+      objectMetadataItem: objectMetadataItemWithFieldMaps,
-    );
+    });
 
     if (executionArgs.graphqlQuerySelectedFieldsResult.relations) {
       await this.processNestedRelationsHelper.processNestedRelations({

packages/twenty-server/src/engine/api/graphql/graphql-query-runner/resolvers/graphql-query-restore-one-resolver.service.ts

@@ -47,11 +47,11 @@ export class GraphqlQueryRestoreOneResolverService extends GraphqlQueryBaseResol
       objectMetadataMaps,
     );
 
-    this.apiEventEmitterService.emitRestoreEvents(
+    this.apiEventEmitterService.emitRestoreEvents({
-      formattedRestoredRecords,
+      records: formattedRestoredRecords,
       authContext,
-      objectMetadataItemWithFieldMaps,
+      objectMetadataItem: objectMetadataItemWithFieldMaps,
-    );
+    });
 
     if (formattedRestoredRecords.length === 0) {
       throw new GraphqlQueryRunnerException(

packages/twenty-server/src/engine/api/graphql/graphql-query-runner/resolvers/graphql-query-update-many-resolver.service.ts

@@ -89,13 +89,13 @@ export class GraphqlQueryUpdateManyResolverService extends GraphqlQueryBaseResol
       objectMetadataMaps,
     );
 
-    this.apiEventEmitterService.emitUpdateEvents(
+    this.apiEventEmitterService.emitUpdateEvents({
-      formattedExistingRecords,
+      existingRecords: formattedExistingRecords,
-      formattedUpdatedRecords,
+      records: formattedUpdatedRecords,
-      Object.keys(executionArgs.args.data),
+      updatedFields: Object.keys(executionArgs.args.data),
       authContext,
-      objectMetadataItemWithFieldMaps,
+      objectMetadataItem: objectMetadataItemWithFieldMaps,
-    );
+    });
 
     if (executionArgs.graphqlQuerySelectedFieldsResult.relations) {
       await this.processNestedRelationsHelper.processNestedRelations({

packages/twenty-server/src/engine/api/graphql/graphql-query-runner/resolvers/graphql-query-update-one-resolver.service.ts

@@ -74,13 +74,13 @@ export class GraphqlQueryUpdateOneResolverService extends GraphqlQueryBaseResolv
       objectMetadataMaps,
     );
 
-    this.apiEventEmitterService.emitUpdateEvents(
+    this.apiEventEmitterService.emitUpdateEvents({
-      formattedExistingRecords,
+      existingRecords: formattedExistingRecords,
-      formattedUpdatedRecords,
+      records: formattedUpdatedRecords,
-      Object.keys(executionArgs.args.data),
+      updatedFields: Object.keys(executionArgs.args.data),
       authContext,
-      objectMetadataItemWithFieldMaps,
+      objectMetadataItem: objectMetadataItemWithFieldMaps,
-    );
+    });
 
     if (formattedUpdatedRecords.length === 0) {
       throw new GraphqlQueryRunnerException(

packages/twenty-server/src/engine/api/graphql/graphql-query-runner/services/api-event-emitter.service.ts

@@ -12,11 +12,15 @@ import { WorkspaceEventEmitter } from 'src/engine/workspace-event-emitter/worksp
 export class ApiEventEmitterService {
   constructor(private readonly workspaceEventEmitter: WorkspaceEventEmitter) {}
 
-  public emitCreateEvents<T extends ObjectRecord>(
+  public emitCreateEvents<T extends ObjectRecord>({
-    records: T[],
+    records,
-    authContext: AuthContext,
+    authContext,
-    objectMetadataItem: ObjectMetadataInterface,
+    objectMetadataItem,
-  ): void {
+  }: {
+    records: T[];
+    authContext: AuthContext;
+    objectMetadataItem: ObjectMetadataInterface;
+  }): void {
     this.workspaceEventEmitter.emitDatabaseBatchEvent({
       objectMetadataNameSingular: objectMetadataItem.nameSingular,
       action: DatabaseEventAction.CREATED,
@@ -33,13 +37,19 @@ export class ApiEventEmitterService {
     });
   }
 
-  public emitUpdateEvents<T extends ObjectRecord>(
+  public emitUpdateEvents<T extends ObjectRecord>({
-    existingRecords: T[],
+    existingRecords,
-    records: T[],
+    records,
-    updatedFields: string[],
+    updatedFields,
-    authContext: AuthContext,
+    authContext,
-    objectMetadataItem: ObjectMetadataInterface,
+    objectMetadataItem,
-  ): void {
+  }: {
+    existingRecords: T[];
+    records: T[];
+    updatedFields: string[];
+    authContext: AuthContext;
+    objectMetadataItem: ObjectMetadataInterface;
+  }): void {
     const mappedExistingRecords = existingRecords.reduce(
       (acc, { id, ...record }) => ({
         ...acc,
@@ -78,11 +88,15 @@ export class ApiEventEmitterService {
     });
   }
 
-  public emitDeletedEvents<T extends ObjectRecord>(
+  public emitDeletedEvents<T extends ObjectRecord>({
-    records: T[],
+    records,
-    authContext: AuthContext,
+    authContext,
-    objectMetadataItem: ObjectMetadataInterface,
+    objectMetadataItem,
-  ): void {
+  }: {
+    records: T[];
+    authContext: AuthContext;
+    objectMetadataItem: ObjectMetadataInterface;
+  }): void {
     this.workspaceEventEmitter.emitDatabaseBatchEvent({
       objectMetadataNameSingular: objectMetadataItem.nameSingular,
       action: DatabaseEventAction.DELETED,
@@ -101,11 +115,15 @@ export class ApiEventEmitterService {
     });
   }
 
-  public emitRestoreEvents<T extends ObjectRecord>(
+  public emitRestoreEvents<T extends ObjectRecord>({
-    records: T[],
+    records,
-    authContext: AuthContext,
+    authContext,
-    objectMetadataItem: ObjectMetadataInterface,
+    objectMetadataItem,
-  ): void {
+  }: {
+    records: T[];
+    authContext: AuthContext;
+    objectMetadataItem: ObjectMetadataInterface;
+  }): void {
     this.workspaceEventEmitter.emitDatabaseBatchEvent({
       objectMetadataNameSingular: objectMetadataItem.nameSingular,
       action: DatabaseEventAction.RESTORED,
@@ -124,11 +142,15 @@ export class ApiEventEmitterService {
     });
   }
 
-  public emitDestroyEvents<T extends ObjectRecord>(
+  public emitDestroyEvents<T extends ObjectRecord>({
-    records: T[],
+    records,
-    authContext: AuthContext,
+    authContext,
-    objectMetadataItem: ObjectMetadataInterface,
+    objectMetadataItem,
-  ): void {
+  }: {
+    records: T[];
+    authContext: AuthContext;
+    objectMetadataItem: ObjectMetadataInterface;
+  }): void {
     this.workspaceEventEmitter.emitDatabaseBatchEvent({
       objectMetadataNameSingular: objectMetadataItem.nameSingular,
       action: DatabaseEventAction.DESTROYED,

packages/twenty-server/src/engine/api/rest/core/handlers/rest-api-create-many.handler.ts

@@ -54,11 +54,11 @@ export class RestApiCreateManyHandler extends RestApiBaseHandler {
 
     const createdRecords = await repository.save(recordsToCreate);
 
-    this.apiEventEmitterService.emitCreateEvents(
+    this.apiEventEmitterService.emitCreateEvents({
-      createdRecords,
+      records: createdRecords,
-      this.getAuthContextFromRequest(request),
+      authContext: this.getAuthContextFromRequest(request),
-      objectMetadata.objectMetadataMapItem,
+      objectMetadataItem: objectMetadata.objectMetadataMapItem,
-    );
+    });
 
     const records = await this.getRecord({
       recordIds: createdRecords.map((record) => record.id),

packages/twenty-server/src/engine/api/rest/core/handlers/rest-api-create-one.handler.ts

@@ -37,11 +37,11 @@ export class RestApiCreateOneHandler extends RestApiBaseHandler {
 
     const createdRecord = await repository.save(recordToCreate);
 
-    this.apiEventEmitterService.emitCreateEvents(
+    this.apiEventEmitterService.emitCreateEvents({
-      [createdRecord],
+      records: [createdRecord],
-      this.getAuthContextFromRequest(request),
+      authContext: this.getAuthContextFromRequest(request),
-      objectMetadata.objectMetadataMapItem,
+      objectMetadataItem: objectMetadata.objectMetadataMapItem,
-    );
+    });
 
     const records = await this.getRecord({
       recordIds: [createdRecord.id],

packages/twenty-server/src/engine/api/rest/core/handlers/rest-api-delete-one.handler.ts

@@ -23,11 +23,11 @@ export class RestApiDeleteOneHandler extends RestApiBaseHandler {
 
     await repository.delete(recordId);
 
-    this.apiEventEmitterService.emitDestroyEvents(
+    this.apiEventEmitterService.emitDestroyEvents({
-      [recordToDelete],
+      records: [recordToDelete],
-      this.getAuthContextFromRequest(request),
+      authContext: this.getAuthContextFromRequest(request),
-      objectMetadata.objectMetadataMapItem,
+      objectMetadataItem: objectMetadata.objectMetadataMapItem,
-    );
+    });
 
     return this.formatResult({
       operation: 'delete',

packages/twenty-server/src/engine/api/rest/core/handlers/rest-api-update-one.handler.ts

@@ -33,13 +33,13 @@ export class RestApiUpdateOneHandler extends RestApiBaseHandler {
       ...overriddenBody,
     });
 
-    this.apiEventEmitterService.emitUpdateEvents(
+    this.apiEventEmitterService.emitUpdateEvents({
-      [recordToUpdate],
+      existingRecords: [recordToUpdate],
-      [updatedRecord],
+      records: [updatedRecord],
-      Object.keys(request.body),
+      updatedFields: Object.keys(request.body),
-      this.getAuthContextFromRequest(request),
+      authContext: this.getAuthContextFromRequest(request),
-      objectMetadata.objectMetadataMapItem,
+      objectMetadataItem: objectMetadata.objectMetadataMapItem,
-    );
+    });
 
     const records = await this.getRecord({
       recordIds: [updatedRecord.id],