diff --git a/packages/twenty-server/src/database/typeorm/core/migrations/common/1754043158752-AddWorkspaceActivationStatusIndex.ts b/packages/twenty-server/src/database/typeorm/core/migrations/common/1754043158752-AddWorkspaceActivationStatusIndex.ts new file mode 100644 index 0000000000..33b7dcf6d0 --- /dev/null +++ b/packages/twenty-server/src/database/typeorm/core/migrations/common/1754043158752-AddWorkspaceActivationStatusIndex.ts @@ -0,0 +1,19 @@ +import { MigrationInterface, QueryRunner } from 'typeorm'; + +export class AddWorkspaceActivationStatusIndex1754043158752 + implements MigrationInterface +{ + name = 'AddWorkspaceActivationStatusIndex1754043158752'; + + public async up(queryRunner: QueryRunner): Promise { + await queryRunner.query( + `CREATE INDEX "IDX_WORKSPACE_ACTIVATION_STATUS" ON "core"."workspace" ("activationStatus") `, + ); + } + + public async down(queryRunner: QueryRunner): Promise { + await queryRunner.query( + `DROP INDEX "core"."IDX_WORKSPACE_ACTIVATION_STATUS"`, + ); + } +} diff --git a/packages/twenty-server/src/database/typeorm/typeorm.service.ts b/packages/twenty-server/src/database/typeorm/typeorm.service.ts index 574e9034c9..569dfc5566 100644 --- a/packages/twenty-server/src/database/typeorm/typeorm.service.ts +++ b/packages/twenty-server/src/database/typeorm/typeorm.service.ts @@ -1,4 +1,9 @@ -import { Injectable, OnModuleDestroy, OnModuleInit } from '@nestjs/common'; +import { + Injectable, + Logger, + OnModuleDestroy, + OnModuleInit, +} from '@nestjs/common'; import { DataSource } from 'typeorm'; @@ -7,6 +12,7 @@ import { TwentyConfigService } from 'src/engine/core-modules/twenty-config/twent @Injectable() export class TypeORMService implements OnModuleInit, OnModuleDestroy { private mainDataSource: DataSource; + private readonly logger = new Logger(TypeORMService.name); constructor(private readonly twentyConfigService: TwentyConfigService) { const isJest = process.argv.some((arg) => arg.includes('jest')); @@ -61,6 +67,7 @@ export class TypeORMService implements OnModuleInit, OnModuleDestroy { async onModuleDestroy() { // Destroy main data source "default" schema + this.logger.log('Destroying main data source'); await this.mainDataSource.destroy(); } } diff --git a/packages/twenty-server/src/engine/api/graphql/graphql-query-runner/resolvers/graphql-query-create-many-resolver.service.ts b/packages/twenty-server/src/engine/api/graphql/graphql-query-runner/resolvers/graphql-query-create-many-resolver.service.ts index bbe7ee1cbe..c9ff1a5873 100644 --- a/packages/twenty-server/src/engine/api/graphql/graphql-query-runner/resolvers/graphql-query-create-many-resolver.service.ts +++ b/packages/twenty-server/src/engine/api/graphql/graphql-query-runner/resolvers/graphql-query-create-many-resolver.service.ts @@ -298,7 +298,7 @@ export class GraphqlQueryCreateManyResolverService extends GraphqlQueryBaseResol repository, objectMetadataItemWithFieldMaps, result, - columnsToReturn, + columnsToReturn: _, }: { partialRecordsToUpdate: Partial[]; repository: WorkspaceRepository; @@ -306,26 +306,21 @@ export class GraphqlQueryCreateManyResolverService extends GraphqlQueryBaseResol result: InsertResult; columnsToReturn: string[]; }): Promise { - for (const partialRecordToUpdate of partialRecordsToUpdate) { - const recordId = partialRecordToUpdate.id as string; - - // we should not update an existing record's createdBy value - const partialRecordToUpdateWithoutCreatedByUpdate = - this.getRecordWithoutCreatedBy( - partialRecordToUpdate, - objectMetadataItemWithFieldMaps, - ); - - await repository.update( - recordId, - partialRecordToUpdateWithoutCreatedByUpdate, - undefined, - columnsToReturn, + const partialRecordsToUpdateWithoutCreatedByUpdate = + partialRecordsToUpdate.map((record) => + this.getRecordWithoutCreatedBy(record, objectMetadataItemWithFieldMaps), ); - result.identifiers.push({ id: recordId }); - result.generatedMaps.push({ id: recordId }); - } + const savedRecords = await repository.save( + partialRecordsToUpdateWithoutCreatedByUpdate, + ); + + result.identifiers.push( + ...savedRecords.map((record) => ({ id: record.id })), + ); + result.generatedMaps.push( + ...savedRecords.map((record) => ({ id: record.id })), + ); } private async processRecordsToInsert({ diff --git a/packages/twenty-server/src/engine/api/graphql/workspace-query-runner/listeners/entity-events-to-db.listener.ts b/packages/twenty-server/src/engine/api/graphql/workspace-query-runner/listeners/entity-events-to-db.listener.ts index 70c9698566..a2c8e64087 100644 --- a/packages/twenty-server/src/engine/api/graphql/workspace-query-runner/listeners/entity-events-to-db.listener.ts +++ b/packages/twenty-server/src/engine/api/graphql/workspace-query-runner/listeners/entity-events-to-db.listener.ts @@ -67,6 +67,10 @@ export class EntityEventsToDbListener { (event) => event.objectMetadata?.isAuditLogged, ); + if (filteredEvents.length === 0) { + return; + } + const batchEventEventsForWebhook: ObjectRecordEventForWebhook[] = batchEvent.events.map((event) => ({ ...event, diff --git a/packages/twenty-server/src/engine/core-modules/workspace/workspace.entity.ts b/packages/twenty-server/src/engine/core-modules/workspace/workspace.entity.ts index 8cb0d39641..b0429eda90 100644 --- a/packages/twenty-server/src/engine/core-modules/workspace/workspace.entity.ts +++ b/packages/twenty-server/src/engine/core-modules/workspace/workspace.entity.ts @@ -8,6 +8,7 @@ import { CreateDateColumn, DeleteDateColumn, Entity, + Index, OneToMany, PrimaryGeneratedColumn, Relation, @@ -112,6 +113,7 @@ export class Workspace { enum: WorkspaceActivationStatus, default: WorkspaceActivationStatus.INACTIVE, }) + @Index('IDX_WORKSPACE_ACTIVATION_STATUS') activationStatus: WorkspaceActivationStatus; @OneToMany( diff --git a/packages/twenty-server/src/engine/twenty-orm/entity-manager/workspace-entity-manager.ts b/packages/twenty-server/src/engine/twenty-orm/entity-manager/workspace-entity-manager.ts index b62301fb41..b3d099988a 100644 --- a/packages/twenty-server/src/engine/twenty-orm/entity-manager/workspace-entity-manager.ts +++ b/packages/twenty-server/src/engine/twenty-orm/entity-manager/workspace-entity-manager.ts @@ -134,7 +134,7 @@ export class WorkspaceEntityManager extends EntityManager { shouldBypassPermissionChecks: false, objectRecordsPermissions: {}, }, - ): SelectQueryBuilder | WorkspaceSelectQueryBuilder { + ): WorkspaceSelectQueryBuilder { let queryBuilder: SelectQueryBuilder; if (alias) { @@ -173,14 +173,15 @@ export class WorkspaceEntityManager extends EntityManager { selectedColumns: string[] = [], permissionOptions?: PermissionOptions, ): Promise { + const metadata = this.connection.getMetadata(target); + return this.createQueryBuilder( - undefined, - undefined, + target, + metadata.name, undefined, permissionOptions, ) .insert() - .into(target) .values(entity) .returning(selectedColumns) .execute(); @@ -316,6 +317,29 @@ export class WorkspaceEntityManager extends EntityManager { } } + public updateMany( + target: EntityTarget, + inputs: { + criteria: string; + partialEntity: QueryDeepPartialEntity; + }[], + permissionOptions?: PermissionOptions, + selectedColumns?: string[], + ): Promise { + const metadata = this.connection.getMetadata(target); + + return this.createQueryBuilder( + target, + metadata.name, + undefined, + permissionOptions, + ) + .update() + .setManyInputs(inputs) + .returning(selectedColumns ?? []) + .execute(); + } + override increment( target: EntityTarget, criteria: object, @@ -1077,7 +1101,7 @@ export class WorkspaceEntityManager extends EntityManager { entityTarget, ); - const updatedEntities = isDefined(relationNestedConfig) + const entityWithConnectedRelations = isDefined(relationNestedConfig) ? await relationNestedQueries.processRelationNestedQueries({ entities: entityArray, relationNestedConfig, @@ -1090,7 +1114,9 @@ export class WorkspaceEntityManager extends EntityManager { }) : entityArray; - const entityIds = entityArray.map((e) => (e as { id: string }).id); + const entityIds = entityArray + .map((entity) => (entity as { id: string }).id) + .filter(isDefined); const beforeUpdate = await this.find( entityTarget, { @@ -1114,7 +1140,7 @@ export class WorkspaceEntityManager extends EntityManager { ); const formattedEntityOrEntities = formatData( - updatedEntities, + entityWithConnectedRelations, objectMetadataItem, ); @@ -1150,26 +1176,29 @@ export class WorkspaceEntityManager extends EntityManager { this.internalContext.objectMetadataMaps, ); - for (const entity of formattedResult) { - const isUpdate = beforeUpdateMapById[entity.id]; + const updatedEntities = formattedResult.filter( + (entity) => beforeUpdateMapById[entity.id], + ); + const createdEntities = formattedResult.filter( + (entity) => !beforeUpdateMapById[entity.id], + ); - if (isUpdate) { - await this.internalContext.eventEmitterService.emitMutationEvent({ - action: DatabaseEventAction.UPDATED, - objectMetadataItem, - workspaceId: this.internalContext.workspaceId, - entities: [entity], - beforeEntities: beforeUpdateMapById[entity.id], - }); - } else { - await this.internalContext.eventEmitterService.emitMutationEvent({ - action: DatabaseEventAction.CREATED, - objectMetadataItem, - workspaceId: this.internalContext.workspaceId, - entities: [entity], - }); - } - } + await this.internalContext.eventEmitterService.emitMutationEvent({ + action: DatabaseEventAction.UPDATED, + objectMetadataItem, + workspaceId: this.internalContext.workspaceId, + entities: updatedEntities, + beforeEntities: updatedEntities.map( + (entity) => beforeUpdateMapById[entity.id], + ), + }); + + await this.internalContext.eventEmitterService.emitMutationEvent({ + action: DatabaseEventAction.CREATED, + objectMetadataItem, + workspaceId: this.internalContext.workspaceId, + entities: createdEntities, + }); const isFieldPermissionsEnabled = this.getFeatureFlagMap().IS_FIELDS_PERMISSIONS_ENABLED; diff --git a/packages/twenty-server/src/engine/twenty-orm/factories/workspace-datasource.factory.ts b/packages/twenty-server/src/engine/twenty-orm/factories/workspace-datasource.factory.ts index 8b6f20d3ca..0a3396c0f1 100644 --- a/packages/twenty-server/src/engine/twenty-orm/factories/workspace-datasource.factory.ts +++ b/packages/twenty-server/src/engine/twenty-orm/factories/workspace-datasource.factory.ts @@ -63,15 +63,15 @@ export class WorkspaceDatasourceFactory { ); if (isPoolSharingEnabled) { - this.logger.debug( - `PromiseMemoizer Event: A WorkspaceDataSource (using shared pool) is being cleared. Actual pool closure managed by PgPoolSharedService. Not calling dataSource.destroy().`, + this.logger.log( + `PromiseMemoizer Event: A WorkspaceDataSource for workspace ${dataSource.internalContext.workspaceId} is being cleared. Actual pool closure managed by PgPoolSharedService. Not calling dataSource.destroy().`, ); // We should NOT call dataSource.destroy() here, because that would end // the shared pool, potentially affecting other active users of that pool. // The PgPoolSharedService is responsible for the lifecycle of shared pools. } else { - this.logger.debug( - `PromiseMemoizer Event: A WorkspaceDataSource (using dedicated pool) is being cleared. Calling safelyDestroyDataSource.`, + this.logger.log( + `PromiseMemoizer Event: A WorkspaceDataSource for workspace ${dataSource.internalContext.workspaceId} is being cleared. Calling safelyDestroyDataSource.`, ); await this.safelyDestroyDataSource(dataSource); } diff --git a/packages/twenty-server/src/engine/twenty-orm/pg-shared-pool/pg-shared-pool.module.ts b/packages/twenty-server/src/engine/twenty-orm/pg-shared-pool/pg-shared-pool.module.ts index 9dafab619d..133942d768 100644 --- a/packages/twenty-server/src/engine/twenty-orm/pg-shared-pool/pg-shared-pool.module.ts +++ b/packages/twenty-server/src/engine/twenty-orm/pg-shared-pool/pg-shared-pool.module.ts @@ -1,5 +1,6 @@ import { Global, + Logger, Module, OnApplicationShutdown, OnModuleInit, @@ -19,6 +20,7 @@ import { PgPoolSharedService } from 'src/engine/twenty-orm/pg-shared-pool/pg-sha }) export class PgPoolSharedModule implements OnModuleInit, OnApplicationShutdown { constructor(private readonly pgPoolSharedService: PgPoolSharedService) {} + private readonly logger = new Logger(PgPoolSharedModule.name); /** * Initialize the pool sharing service when the module is initialized @@ -31,6 +33,7 @@ export class PgPoolSharedModule implements OnModuleInit, OnApplicationShutdown { * Clean up any resources when the application shuts down */ async onApplicationShutdown() { + this.logger.log('Shutting down PgPoolSharedModule'); await this.pgPoolSharedService.onApplicationShutdown(); } } diff --git a/packages/twenty-server/src/engine/twenty-orm/pg-shared-pool/pg-shared-pool.service.ts b/packages/twenty-server/src/engine/twenty-orm/pg-shared-pool/pg-shared-pool.service.ts index 4c53afbadd..7b17865415 100644 --- a/packages/twenty-server/src/engine/twenty-orm/pg-shared-pool/pg-shared-pool.service.ts +++ b/packages/twenty-server/src/engine/twenty-orm/pg-shared-pool/pg-shared-pool.service.ts @@ -140,6 +140,7 @@ export class PgPoolSharedService { */ async onApplicationShutdown(): Promise { this.stopStatsLogging(); + this.logger.log('onApplicationShutdown called in PgPoolSharedService'); await this.closeAllPools(); } diff --git a/packages/twenty-server/src/engine/twenty-orm/repository/workspace-delete-query-builder.ts b/packages/twenty-server/src/engine/twenty-orm/repository/workspace-delete-query-builder.ts index cd9de91d26..d0b241b4e5 100644 --- a/packages/twenty-server/src/engine/twenty-orm/repository/workspace-delete-query-builder.ts +++ b/packages/twenty-server/src/engine/twenty-orm/repository/workspace-delete-query-builder.ts @@ -78,6 +78,21 @@ export class WorkspaceDeleteQueryBuilder< this.internalContext, ); + const eventSelectQueryBuilder = new WorkspaceSelectQueryBuilder( + this as unknown as WorkspaceSelectQueryBuilder, + this.objectRecordsPermissions, + this.internalContext, + true, + this.authContext, + this.featureFlagMap, + ); + + eventSelectQueryBuilder.expressionMap.wheres = this.expressionMap.wheres; + eventSelectQueryBuilder.expressionMap.aliases = this.expressionMap.aliases; + eventSelectQueryBuilder.setParameters(this.getParameters()); + + const before = await eventSelectQueryBuilder.getOne(); + const result = await super.execute(); const formattedResult = formatResult( @@ -86,11 +101,17 @@ export class WorkspaceDeleteQueryBuilder< this.internalContext.objectMetadataMaps, ); + const formattedBefore = formatResult( + before, + objectMetadata, + this.internalContext.objectMetadataMaps, + ); + await this.internalContext.eventEmitterService.emitMutationEvent({ action: DatabaseEventAction.DESTROYED, objectMetadataItem: objectMetadata, workspaceId: this.internalContext.workspaceId, - entities: formattedResult, + entities: formattedBefore, authContext: this.authContext, }); diff --git a/packages/twenty-server/src/engine/twenty-orm/repository/workspace-insert-query-builder.ts b/packages/twenty-server/src/engine/twenty-orm/repository/workspace-insert-query-builder.ts index 061b18977c..d40df96cde 100644 --- a/packages/twenty-server/src/engine/twenty-orm/repository/workspace-insert-query-builder.ts +++ b/packages/twenty-server/src/engine/twenty-orm/repository/workspace-insert-query-builder.ts @@ -137,9 +137,24 @@ export class WorkspaceInsertQueryBuilder< } const result = await super.execute(); + const eventSelectQueryBuilder = new WorkspaceSelectQueryBuilder( + this as unknown as WorkspaceSelectQueryBuilder, + this.objectRecordsPermissions, + this.internalContext, + true, + this.authContext, + this.featureFlagMap, + ); + + eventSelectQueryBuilder.expressionMap.aliases = this.expressionMap.aliases; + eventSelectQueryBuilder.whereInIds( + result.identifiers.map((identifier) => identifier.id), + ); + + const afterResult = await eventSelectQueryBuilder.getMany(); const formattedResult = formatResult( - result.raw, + afterResult, objectMetadata, this.internalContext.objectMetadataMaps, ); @@ -153,7 +168,7 @@ export class WorkspaceInsertQueryBuilder< }); return { - raw: result.raw, + raw: afterResult, generatedMaps: formattedResult, identifiers: result.identifiers, }; diff --git a/packages/twenty-server/src/engine/twenty-orm/repository/workspace-update-query-builder.ts b/packages/twenty-server/src/engine/twenty-orm/repository/workspace-update-query-builder.ts index 1b0ee92913..5b3595ddc2 100644 --- a/packages/twenty-server/src/engine/twenty-orm/repository/workspace-update-query-builder.ts +++ b/packages/twenty-server/src/engine/twenty-orm/repository/workspace-update-query-builder.ts @@ -42,6 +42,10 @@ export class WorkspaceUpdateQueryBuilder< private relationNestedConfig: | [RelationConnectQueryConfig[], RelationDisconnectQueryFieldsByEntityIndex] | null; + private manyInputs: { + criteria: string; + partialEntity: QueryDeepPartialEntity; + }[]; constructor( queryBuilder: UpdateQueryBuilder, @@ -76,6 +80,10 @@ export class WorkspaceUpdateQueryBuilder< } override async execute(): Promise { + if (this.manyInputs) { + return this.executeMany(); + } + validateQueryIsPermittedOrThrow({ expressionMap: this.expressionMap, objectRecordsPermissions: this.objectRecordsPermissions, @@ -166,6 +174,120 @@ export class WorkspaceUpdateQueryBuilder< }; } + public async executeMany(): Promise { + for (const input of this.manyInputs) { + const fakeExpressionMapToValidatePermissions = Object.assign( + {}, + this.expressionMap, + { + wheres: input.criteria, + valuesSet: input.partialEntity, + }, + ); + + validateQueryIsPermittedOrThrow({ + expressionMap: fakeExpressionMapToValidatePermissions, + objectRecordsPermissions: this.objectRecordsPermissions, + objectMetadataMaps: this.internalContext.objectMetadataMaps, + shouldBypassPermissionChecks: this.shouldBypassPermissionChecks, + isFieldPermissionsEnabled: + this.featureFlagMap?.[FeatureFlagKey.IS_FIELDS_PERMISSIONS_ENABLED], + }); + } + + const mainAliasTarget = this.getMainAliasTarget(); + + const objectMetadata = getObjectMetadataFromEntityTarget( + mainAliasTarget, + this.internalContext, + ); + + const eventSelectQueryBuilder = new WorkspaceSelectQueryBuilder( + this as unknown as WorkspaceSelectQueryBuilder, + this.objectRecordsPermissions, + this.internalContext, + true, + this.authContext, + this.featureFlagMap, + ); + + eventSelectQueryBuilder.whereInIds( + this.manyInputs.map((input) => input.criteria), + ); + eventSelectQueryBuilder.expressionMap.aliases = this.expressionMap.aliases; + eventSelectQueryBuilder.setParameters(this.getParameters()); + + const beforeRecords = await eventSelectQueryBuilder.getMany(); + + const formattedBefore = formatResult( + beforeRecords, + objectMetadata, + this.internalContext.objectMetadataMaps, + ); + + const results: UpdateResult[] = []; + + for (const input of this.manyInputs) { + this.expressionMap.valuesSet = input.partialEntity; + this.where({ id: input.criteria }); + + const nestedRelationQueryBuilder = new WorkspaceSelectQueryBuilder( + this as unknown as WorkspaceSelectQueryBuilder, + this.objectRecordsPermissions, + this.internalContext, + this.shouldBypassPermissionChecks, + this.authContext, + ); + + if (isDefined(this.relationNestedConfig)) { + const updatedValues = + await this.relationNestedQueries.processRelationNestedQueries({ + entities: input.partialEntity as + | QueryDeepPartialEntityWithNestedRelationFields + | QueryDeepPartialEntityWithNestedRelationFields[], + relationNestedConfig: this.relationNestedConfig, + queryBuilder: nestedRelationQueryBuilder, + }); + + this.expressionMap.valuesSet = + updatedValues.length === 1 ? updatedValues[0] : updatedValues; + } + + const result = await super.execute(); + + results.push(result); + } + + const afterRecords = await eventSelectQueryBuilder.getMany(); + + const formattedAfter = formatResult( + afterRecords, + objectMetadata, + this.internalContext.objectMetadataMaps, + ); + + await this.internalContext.eventEmitterService.emitMutationEvent({ + action: DatabaseEventAction.UPDATED, + objectMetadataItem: objectMetadata, + workspaceId: this.internalContext.workspaceId, + entities: formattedAfter, + beforeEntities: formattedBefore, + authContext: this.authContext, + }); + + const formattedResults = formatResult( + results.map((result) => result.raw), + objectMetadata, + this.internalContext.objectMetadataMaps, + ); + + return { + raw: results.map((result) => result.raw), + generatedMaps: formattedResults, + affected: results.length, + }; + } + override set( _values: | QueryDeepPartialEntityWithNestedRelationFields @@ -243,4 +365,15 @@ export class WorkspaceUpdateQueryBuilder< return mainAliasTarget; } + + public setManyInputs( + inputs: { + criteria: string; + partialEntity: QueryDeepPartialEntity; + }[], + ): this { + this.manyInputs = inputs; + + return this; + } } diff --git a/packages/twenty-server/src/engine/twenty-orm/repository/workspace.repository.ts b/packages/twenty-server/src/engine/twenty-orm/repository/workspace.repository.ts index b3cb1ada77..0d135de9a0 100644 --- a/packages/twenty-server/src/engine/twenty-orm/repository/workspace.repository.ts +++ b/packages/twenty-server/src/engine/twenty-orm/repository/workspace.repository.ts @@ -595,6 +595,32 @@ export class WorkspaceRepository< ); } + // Experimental method to allow batch update and batch event emission + async updateMany( + inputs: { + criteria: string; + partialEntity: QueryDeepPartialEntity; + }[], + entityManager?: WorkspaceEntityManager, + selectedColumns?: string[], + ): Promise { + const manager = entityManager || this.manager; + + const permissionOptions = { + shouldBypassPermissionChecks: this.shouldBypassPermissionChecks, + objectRecordsPermissions: this.objectRecordsPermissions, + }; + + const results = await manager.updateMany( + this.target, + inputs, + permissionOptions, + selectedColumns, + ); + + return results; + } + override async upsert( entityOrEntities: QueryDeepPartialEntity | QueryDeepPartialEntity[], conflictPathsOrOptions: string[] | UpsertOptions, diff --git a/packages/twenty-server/src/engine/twenty-orm/utils/format-result.util.ts b/packages/twenty-server/src/engine/twenty-orm/utils/format-result.util.ts index d99983b454..b3ae019d4c 100644 --- a/packages/twenty-server/src/engine/twenty-orm/utils/format-result.util.ts +++ b/packages/twenty-server/src/engine/twenty-orm/utils/format-result.util.ts @@ -13,14 +13,13 @@ import { getCompositeFieldMetadataCollection } from 'src/engine/twenty-orm/utils import { isFieldMetadataEntityOfType } from 'src/engine/utils/is-field-metadata-of-type.util'; import { isDate } from 'src/utils/date/isDate'; import { isValidDate } from 'src/utils/date/isValidDate'; - export function formatResult( // eslint-disable-next-line @typescript-eslint/no-explicit-any data: any, objectMetadataItemWithFieldMaps: ObjectMetadataItemWithFieldMaps | undefined, objectMetadataMaps: ObjectMetadataMaps, ): T { - if (!data) { + if (!isDefined(data)) { return data; } diff --git a/packages/twenty-server/src/engine/workspace-event-emitter/workspace-event-emitter.ts b/packages/twenty-server/src/engine/workspace-event-emitter/workspace-event-emitter.ts index 496ec9df17..230e2bf785 100644 --- a/packages/twenty-server/src/engine/workspace-event-emitter/workspace-event-emitter.ts +++ b/packages/twenty-server/src/engine/workspace-event-emitter/workspace-event-emitter.ts @@ -15,6 +15,7 @@ import { ObjectRecordUpdateEvent } from 'src/engine/core-modules/event-emitter/t import { objectRecordChangedValues } from 'src/engine/core-modules/event-emitter/utils/object-record-changed-values'; import { ObjectMetadataItemWithFieldMaps } from 'src/engine/metadata-modules/types/object-metadata-item-with-field-maps'; import { CustomEventName } from 'src/engine/workspace-event-emitter/types/custom-event-name.type'; +import { STANDARD_OBJECT_IDS } from 'src/engine/workspace-manager/workspace-sync-metadata/constants/standard-object-ids'; type ActionEventMap = { [DatabaseEventAction.CREATED]: ObjectRecordCreateEvent; @@ -43,6 +44,12 @@ export class WorkspaceEventEmitter { entities: T | T[]; beforeEntities?: T | T[]; }) { + if ( + objectMetadataItem.standardId === STANDARD_OBJECT_IDS.timelineActivity + ) { + return; + } + const objectMetadataNameSingular = objectMetadataItem.nameSingular; const fields = Object.values(objectMetadataItem.fieldsById ?? {}); const entityArray = isDefined(entities) diff --git a/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/services/calendar-save-events.service.ts b/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/services/calendar-save-events.service.ts index 181442cb17..ce0e873384 100644 --- a/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/services/calendar-save-events.service.ts +++ b/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/services/calendar-save-events.service.ts @@ -199,7 +199,7 @@ export class CalendarSaveEventsService { transactionManager, ); - const participantsToSave = + const participantsToCreate = fetchedCalendarEventsWithDBEventsEnrichedWithSavedEvents .filter( ({ newlyCreatedCalendarEvent }) => @@ -240,7 +240,7 @@ export class CalendarSaveEventsService { await this.calendarEventParticipantService.upsertAndDeleteCalendarEventParticipants( { - participantsToSave, + participantsToCreate, participantsToUpdate, transactionManager, calendarChannel, diff --git a/packages/twenty-server/src/modules/calendar/calendar-event-participant-manager/calendar-event-participant-manager.module.ts b/packages/twenty-server/src/modules/calendar/calendar-event-participant-manager/calendar-event-participant-manager.module.ts index d1b68cbcce..af6f161f07 100644 --- a/packages/twenty-server/src/modules/calendar/calendar-event-participant-manager/calendar-event-participant-manager.module.ts +++ b/packages/twenty-server/src/modules/calendar/calendar-event-participant-manager/calendar-event-participant-manager.module.ts @@ -10,7 +10,6 @@ import { ObjectMetadataEntity } from 'src/engine/metadata-modules/object-metadat import { WorkspaceDataSourceModule } from 'src/engine/workspace-datasource/workspace-datasource.module'; import { CalendarCreateCompanyAndContactAfterSyncJob } from 'src/modules/calendar/calendar-event-participant-manager/jobs/calendar-create-company-and-contact-after-sync.job'; import { CalendarEventParticipantMatchParticipantJob } from 'src/modules/calendar/calendar-event-participant-manager/jobs/calendar-event-participant-match-participant.job'; -import { CalendarEventParticipantUnmatchParticipantJob } from 'src/modules/calendar/calendar-event-participant-manager/jobs/calendar-event-participant-unmatch-participant.job'; import { CalendarEventParticipantPersonListener } from 'src/modules/calendar/calendar-event-participant-manager/listeners/calendar-event-participant-person.listener'; import { CalendarEventParticipantWorkspaceMemberListener } from 'src/modules/calendar/calendar-event-participant-manager/listeners/calendar-event-participant-workspace-member.listener'; import { CalendarEventParticipantListener } from 'src/modules/calendar/calendar-event-participant-manager/listeners/calendar-event-participant.listener'; @@ -34,7 +33,6 @@ import { MatchParticipantModule } from 'src/modules/match-participant/match-part CalendarEventParticipantService, CalendarCreateCompanyAndContactAfterSyncJob, CalendarEventParticipantMatchParticipantJob, - CalendarEventParticipantUnmatchParticipantJob, CalendarEventParticipantListener, CalendarEventParticipantPersonListener, CalendarEventParticipantWorkspaceMemberListener, diff --git a/packages/twenty-server/src/modules/calendar/calendar-event-participant-manager/jobs/calendar-event-participant-match-participant.job.ts b/packages/twenty-server/src/modules/calendar/calendar-event-participant-manager/jobs/calendar-event-participant-match-participant.job.ts index 3e249bfebe..22107be791 100644 --- a/packages/twenty-server/src/modules/calendar/calendar-event-participant-manager/jobs/calendar-event-participant-match-participant.job.ts +++ b/packages/twenty-server/src/modules/calendar/calendar-event-participant-manager/jobs/calendar-event-participant-match-participant.job.ts @@ -12,10 +12,11 @@ import { MatchParticipantService } from 'src/modules/match-participant/match-par export type CalendarEventParticipantMatchParticipantJobData = { workspaceId: string; - isPrimaryEmail: boolean; - email: string; - personId?: string; - workspaceMemberId?: string; + participantMatching: { + personIds: string[]; + personEmails: string[]; + workspaceMemberIds: string[]; + }; }; @Processor({ @@ -33,8 +34,7 @@ export class CalendarEventParticipantMatchParticipantJob { async handle( data: CalendarEventParticipantMatchParticipantJobData, ): Promise { - const { workspaceId, isPrimaryEmail, email, personId, workspaceMemberId } = - data; + const { workspaceId, participantMatching } = data; const workspace = await this.workspaceRepository.findOne({ where: { @@ -46,23 +46,21 @@ export class CalendarEventParticipantMatchParticipantJob { return; } - if (personId) { - await this.matchParticipantService.matchParticipantsAfterPersonCreation({ - handle: email, - isPrimaryEmail, + if ( + participantMatching.personIds.length > 0 || + participantMatching.personEmails.length > 0 + ) { + await this.matchParticipantService.matchParticipantsForPeople({ objectMetadataName: 'calendarEventParticipant', - personId, + participantMatching, }); } - if (workspaceMemberId) { - await this.matchParticipantService.matchParticipantsAfterWorkspaceMemberCreation( - { - handle: email, - objectMetadataName: 'calendarEventParticipant', - workspaceMemberId, - }, - ); + if (participantMatching.workspaceMemberIds.length > 0) { + await this.matchParticipantService.matchParticipantsForWorkspaceMembers({ + objectMetadataName: 'calendarEventParticipant', + participantMatching, + }); } } } diff --git a/packages/twenty-server/src/modules/calendar/calendar-event-participant-manager/jobs/calendar-event-participant-unmatch-participant.job.ts b/packages/twenty-server/src/modules/calendar/calendar-event-participant-manager/jobs/calendar-event-participant-unmatch-participant.job.ts deleted file mode 100644 index 765e1286b7..0000000000 --- a/packages/twenty-server/src/modules/calendar/calendar-event-participant-manager/jobs/calendar-event-participant-unmatch-participant.job.ts +++ /dev/null @@ -1,38 +0,0 @@ -import { Scope } from '@nestjs/common'; - -import { Process } from 'src/engine/core-modules/message-queue/decorators/process.decorator'; -import { Processor } from 'src/engine/core-modules/message-queue/decorators/processor.decorator'; -import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants'; -import { CalendarEventParticipantWorkspaceEntity } from 'src/modules/calendar/common/standard-objects/calendar-event-participant.workspace-entity'; -import { MatchParticipantService } from 'src/modules/match-participant/match-participant.service'; - -export type CalendarEventParticipantUnmatchParticipantJobData = { - workspaceId: string; - email: string; - personId?: string; - workspaceMemberId?: string; -}; - -@Processor({ - queueName: MessageQueue.calendarQueue, - scope: Scope.REQUEST, -}) -export class CalendarEventParticipantUnmatchParticipantJob { - constructor( - private readonly matchParticipantService: MatchParticipantService, - ) {} - - @Process(CalendarEventParticipantUnmatchParticipantJob.name) - async handle( - data: CalendarEventParticipantUnmatchParticipantJobData, - ): Promise { - const { email, personId, workspaceMemberId } = data; - - await this.matchParticipantService.unmatchParticipants({ - handle: email, - objectMetadataName: 'calendarEventParticipant', - personId, - workspaceMemberId, - }); - } -} diff --git a/packages/twenty-server/src/modules/calendar/calendar-event-participant-manager/listeners/calendar-event-participant-person.listener.ts b/packages/twenty-server/src/modules/calendar/calendar-event-participant-manager/listeners/calendar-event-participant-person.listener.ts index d529d25b06..678659dd53 100644 --- a/packages/twenty-server/src/modules/calendar/calendar-event-participant-manager/listeners/calendar-event-participant-person.listener.ts +++ b/packages/twenty-server/src/modules/calendar/calendar-event-participant-manager/listeners/calendar-event-participant-person.listener.ts @@ -16,12 +16,6 @@ import { CalendarEventParticipantMatchParticipantJob, CalendarEventParticipantMatchParticipantJobData, } from 'src/modules/calendar/calendar-event-participant-manager/jobs/calendar-event-participant-match-participant.job'; -import { - CalendarEventParticipantUnmatchParticipantJob, - CalendarEventParticipantUnmatchParticipantJobData, -} from 'src/modules/calendar/calendar-event-participant-manager/jobs/calendar-event-participant-unmatch-participant.job'; -import { computeChangedAdditionalEmails } from 'src/modules/contact-creation-manager/utils/compute-changed-additional-emails'; -import { hasPrimaryEmailChanged } from 'src/modules/contact-creation-manager/utils/has-primary-email-changed'; import { PersonWorkspaceEntity } from 'src/modules/person/standard-objects/person.workspace-entity'; @Injectable() @@ -37,45 +31,34 @@ export class CalendarEventParticipantPersonListener { ObjectRecordCreateEvent >, ) { - for (const eventPayload of payload.events) { - const jobPromises: Promise[] = []; + const personWithEmails = payload.events.filter( + (eventPayload) => + isDefined(eventPayload.properties.after.emails?.primaryEmail) || + isDefined(eventPayload.properties.after.emails?.additionalEmails), + ); - if (isDefined(eventPayload.properties.after.emails?.primaryEmail)) { - // TODO: modify this job to take an array of participants to match - jobPromises.push( - this.messageQueueService.add( - CalendarEventParticipantMatchParticipantJob.name, - { - workspaceId: payload.workspaceId, - email: eventPayload.properties.after.emails?.primaryEmail, - isPrimaryEmail: true, - personId: eventPayload.recordId, - }, - ), - ); - } + const personIds = personWithEmails.map( + (eventPayload) => eventPayload.recordId, + ); + const personEmails = personWithEmails + .flatMap((eventPayload) => [ + eventPayload.properties.after.emails.primaryEmail, + ...((eventPayload.properties.after.emails?.additionalEmails ?? + []) as string[]), + ]) + .filter(isDefined); - const additionalEmails = - eventPayload.properties.after.emails?.additionalEmails; - - if (Array.isArray(additionalEmails)) { - const additionalEmailPromises = additionalEmails.map((email) => - this.messageQueueService.add( - CalendarEventParticipantMatchParticipantJob.name, - { - workspaceId: payload.workspaceId, - email: email, - isPrimaryEmail: false, - personId: eventPayload.recordId, - }, - ), - ); - - jobPromises.push(...additionalEmailPromises); - } - - await Promise.all(jobPromises); - } + await this.messageQueueService.add( + CalendarEventParticipantMatchParticipantJob.name, + { + workspaceId: payload.workspaceId, + participantMatching: { + personIds, + personEmails, + workspaceMemberIds: [], + }, + }, + ); } @OnDatabaseBatchEvent('person', DatabaseEventAction.UPDATED) @@ -84,81 +67,35 @@ export class CalendarEventParticipantPersonListener { ObjectRecordUpdateEvent >, ) { - for (const eventPayload of payload.events) { - if ( - objectRecordUpdateEventChangedProperties( - eventPayload.properties.before, - eventPayload.properties.after, - ).includes('emails') - ) { - if (!isDefined(eventPayload.properties.diff)) { - continue; - } + const personWithEmails = payload.events.filter((eventPayload) => + objectRecordUpdateEventChangedProperties( + eventPayload.properties.before, + eventPayload.properties.after, + ).includes('emails'), + ); - const jobPromises: Promise[] = []; + const personIds = personWithEmails.map( + (eventPayload) => eventPayload.recordId, + ); + const personEmails = personWithEmails + .flatMap((eventPayload) => [ + eventPayload.properties.after.emails.primaryEmail, + ...((eventPayload.properties.after.emails?.additionalEmails ?? + []) as string[]), + ]) + .filter(isDefined); - if (hasPrimaryEmailChanged(eventPayload.properties.diff)) { - if (eventPayload.properties.before.emails?.primaryEmail) { - jobPromises.push( - this.messageQueueService.add( - CalendarEventParticipantUnmatchParticipantJob.name, - { - workspaceId: payload.workspaceId, - email: eventPayload.properties.before.emails?.primaryEmail, - personId: eventPayload.recordId, - }, - ), - ); - } - - if (eventPayload.properties.after.emails?.primaryEmail) { - jobPromises.push( - this.messageQueueService.add( - CalendarEventParticipantMatchParticipantJob.name, - { - workspaceId: payload.workspaceId, - email: eventPayload.properties.after.emails?.primaryEmail, - isPrimaryEmail: true, - personId: eventPayload.recordId, - }, - ), - ); - } - } - - const { addedAdditionalEmails, removedAdditionalEmails } = - computeChangedAdditionalEmails(eventPayload.properties.diff); - - const removedEmailPromises = removedAdditionalEmails - ?.filter((email: string) => isDefined(email)) - .map((email) => - this.messageQueueService.add( - CalendarEventParticipantUnmatchParticipantJob.name, - { - workspaceId: payload.workspaceId, - email: email, - personId: eventPayload.recordId, - }, - ), - ); - - const addedEmailPromises = addedAdditionalEmails.map((email) => - this.messageQueueService.add( - CalendarEventParticipantMatchParticipantJob.name, - { - workspaceId: payload.workspaceId, - email: email, - isPrimaryEmail: false, - personId: eventPayload.recordId, - }, - ), - ); - - jobPromises.push(...removedEmailPromises, ...addedEmailPromises); - - await Promise.all(jobPromises); - } - } + await this.messageQueueService.add( + CalendarEventParticipantMatchParticipantJob.name, + { + workspaceId: payload.workspaceId, + participantMatching: { + personIds, + personEmails, + workspaceMemberIds: [], + }, + }, + ); } @OnDatabaseBatchEvent('person', DatabaseEventAction.DESTROYED) @@ -167,37 +104,30 @@ export class CalendarEventParticipantPersonListener { ObjectRecordDeleteEvent >, ) { - for (const eventPayload of payload.events) { - if (isDefined(eventPayload.properties.before.emails?.primaryEmail)) { - await this.messageQueueService.add( - CalendarEventParticipantUnmatchParticipantJob.name, - { - workspaceId: payload.workspaceId, - email: eventPayload.properties.before.emails?.primaryEmail, - personId: eventPayload.recordId, - }, - ); - } + const peopleHavingEmails = payload.events.filter( + (eventPayload) => + isDefined(eventPayload.properties.before.emails?.primaryEmail) || + isDefined(eventPayload.properties.before.emails?.additionalEmails), + ); - const additionalEmails = - eventPayload.properties.before.emails?.additionalEmails; + const personEmails = peopleHavingEmails + .flatMap((eventPayload) => [ + eventPayload.properties.before.emails.primaryEmail, + ...((eventPayload.properties.before.emails?.additionalEmails ?? + []) as string[]), + ]) + .filter(isDefined); - if (Array.isArray(additionalEmails)) { - const additionalEmailPromises = additionalEmails - ?.filter((email: string) => isDefined(email)) - .map((email) => - this.messageQueueService.add( - CalendarEventParticipantUnmatchParticipantJob.name, - { - workspaceId: payload.workspaceId, - email: email, - personId: eventPayload.recordId, - }, - ), - ); - - await Promise.all(additionalEmailPromises); - } - } + await this.messageQueueService.add( + CalendarEventParticipantMatchParticipantJob.name, + { + workspaceId: payload.workspaceId, + participantMatching: { + personIds: [], + personEmails, + workspaceMemberIds: [], + }, + }, + ); } } diff --git a/packages/twenty-server/src/modules/calendar/calendar-event-participant-manager/listeners/calendar-event-participant-workspace-member.listener.ts b/packages/twenty-server/src/modules/calendar/calendar-event-participant-manager/listeners/calendar-event-participant-workspace-member.listener.ts index 8792bc124e..b260462ee1 100644 --- a/packages/twenty-server/src/modules/calendar/calendar-event-participant-manager/listeners/calendar-event-participant-workspace-member.listener.ts +++ b/packages/twenty-server/src/modules/calendar/calendar-event-participant-manager/listeners/calendar-event-participant-workspace-member.listener.ts @@ -13,10 +13,6 @@ import { CalendarEventParticipantMatchParticipantJob, CalendarEventParticipantMatchParticipantJobData, } from 'src/modules/calendar/calendar-event-participant-manager/jobs/calendar-event-participant-match-participant.job'; -import { - CalendarEventParticipantUnmatchParticipantJob, - CalendarEventParticipantUnmatchParticipantJobData, -} from 'src/modules/calendar/calendar-event-participant-manager/jobs/calendar-event-participant-unmatch-participant.job'; import { WorkspaceMemberWorkspaceEntity } from 'src/modules/workspace-member/standard-objects/workspace-member.workspace-entity'; @Injectable() @@ -41,9 +37,11 @@ export class CalendarEventParticipantWorkspaceMemberListener { CalendarEventParticipantMatchParticipantJob.name, { workspaceId: payload.workspaceId, - email: eventPayload.properties.after.userEmail, - workspaceMemberId: eventPayload.recordId, - isPrimaryEmail: true, + participantMatching: { + personIds: [], + personEmails: [], + workspaceMemberIds: [eventPayload.recordId], + }, }, ); } @@ -62,22 +60,15 @@ export class CalendarEventParticipantWorkspaceMemberListener { eventPayload.properties.after, ).includes('userEmail') ) { - await this.messageQueueService.add( - CalendarEventParticipantUnmatchParticipantJob.name, - { - workspaceId: payload.workspaceId, - email: eventPayload.properties.before.userEmail, - personId: eventPayload.recordId, - }, - ); - await this.messageQueueService.add( CalendarEventParticipantMatchParticipantJob.name, { workspaceId: payload.workspaceId, - email: eventPayload.properties.after.userEmail, - workspaceMemberId: eventPayload.recordId, - isPrimaryEmail: true, + participantMatching: { + personIds: [], + personEmails: [], + workspaceMemberIds: [eventPayload.recordId], + }, }, ); } diff --git a/packages/twenty-server/src/modules/calendar/calendar-event-participant-manager/services/calendar-event-participant.service.ts b/packages/twenty-server/src/modules/calendar/calendar-event-participant-manager/services/calendar-event-participant.service.ts index f66e8e12aa..d870e1c99d 100644 --- a/packages/twenty-server/src/modules/calendar/calendar-event-participant-manager/services/calendar-event-participant.service.ts +++ b/packages/twenty-server/src/modules/calendar/calendar-event-participant-manager/services/calendar-event-participant.service.ts @@ -1,6 +1,7 @@ import { Injectable } from '@nestjs/common'; import { isDefined } from 'class-validator'; +import chunk from 'lodash.chunk'; import differenceWith from 'lodash.differencewith'; import { Any } from 'typeorm'; @@ -25,6 +26,11 @@ type FetchedCalendarEventParticipantWithCalendarEventId = calendarEventId: string; }; +type FetchedCalendarEventParticipantWithCalendarEventIdAndExistingId = + FetchedCalendarEventParticipantWithCalendarEventId & { + id: string; + }; + @Injectable() export class CalendarEventParticipantService { constructor( @@ -35,39 +41,44 @@ export class CalendarEventParticipantService { ) {} public async upsertAndDeleteCalendarEventParticipants({ - participantsToSave, + participantsToCreate, participantsToUpdate, transactionManager, calendarChannel, connectedAccount, workspaceId, }: { - participantsToSave: FetchedCalendarEventParticipantWithCalendarEventId[]; + participantsToCreate: FetchedCalendarEventParticipantWithCalendarEventId[]; participantsToUpdate: FetchedCalendarEventParticipantWithCalendarEventId[]; transactionManager?: WorkspaceEntityManager; calendarChannel: CalendarChannelWorkspaceEntity; connectedAccount: ConnectedAccountWorkspaceEntity; workspaceId: string; }): Promise { + const chunkedParticipantsToUpdate = chunk(participantsToUpdate, 200); + const calendarEventParticipantRepository = await this.twentyORMManager.getRepository( 'calendarEventParticipant', ); - const existingCalendarEventParticipants = - await calendarEventParticipantRepository.find({ - where: { - calendarEventId: Any( - participantsToUpdate - .map((participant) => participant.calendarEventId) - .filter(isDefined), - ), - }, - }); + for (const participantsToUpdateChunk of chunkedParticipantsToUpdate) { + const existingCalendarEventParticipants = + await calendarEventParticipantRepository.find({ + where: { + calendarEventId: Any( + participantsToUpdateChunk + .map((participant) => participant.calendarEventId) + .filter(isDefined), + ), + }, + }); - const { calendarEventParticipantsToUpdate, newCalendarEventParticipants } = - participantsToUpdate.reduce<{ - calendarEventParticipantsToUpdate: FetchedCalendarEventParticipantWithCalendarEventId[]; + const { + calendarEventParticipantsToUpdate, + newCalendarEventParticipants, + } = participantsToUpdateChunk.reduce<{ + calendarEventParticipantsToUpdate: FetchedCalendarEventParticipantWithCalendarEventIdAndExistingId[]; newCalendarEventParticipants: FetchedCalendarEventParticipantWithCalendarEventId[]; }>( (acc, calendarEventParticipant) => { @@ -81,9 +92,10 @@ export class CalendarEventParticipantService { ); if (existingCalendarEventParticipant) { - acc.calendarEventParticipantsToUpdate.push( - calendarEventParticipant, - ); + acc.calendarEventParticipantsToUpdate.push({ + ...calendarEventParticipant, + id: existingCalendarEventParticipant.id, + }); } else { acc.newCalendarEventParticipants.push(calendarEventParticipant); } @@ -96,47 +108,49 @@ export class CalendarEventParticipantService { }, ); - const calendarEventParticipantsToDelete = differenceWith( - existingCalendarEventParticipants, - participantsToUpdate, - (existingCalendarEventParticipant, participantToUpdate) => - existingCalendarEventParticipant.handle === - participantToUpdate.handle && - existingCalendarEventParticipant.calendarEventId === - participantToUpdate.calendarEventId, - ); + const calendarEventParticipantsToDelete = differenceWith( + existingCalendarEventParticipants, + participantsToUpdateChunk, + (existingCalendarEventParticipant, participantToUpdate) => + existingCalendarEventParticipant.handle === + participantToUpdate.handle && + existingCalendarEventParticipant.calendarEventId === + participantToUpdate.calendarEventId, + ); - await calendarEventParticipantRepository.delete( - { - id: Any( - calendarEventParticipantsToDelete.map( - (calendarEventParticipant) => calendarEventParticipant.id, + await calendarEventParticipantRepository.delete( + { + id: Any( + calendarEventParticipantsToDelete.map( + (calendarEventParticipant) => calendarEventParticipant.id, + ), ), - ), - }, - transactionManager, - ); - - for (const calendarEventParticipantToUpdate of calendarEventParticipantsToUpdate) { - await calendarEventParticipantRepository.update( - { - calendarEventId: calendarEventParticipantToUpdate.calendarEventId, - handle: calendarEventParticipantToUpdate.handle, - }, - { - ...calendarEventParticipantToUpdate, }, transactionManager, ); + + await calendarEventParticipantRepository.updateMany( + calendarEventParticipantsToUpdate.map((participant) => ({ + criteria: participant.id, + partialEntity: participant, + })), + transactionManager, + ); + participantsToCreate.push(...newCalendarEventParticipants); } - participantsToSave.push(...newCalendarEventParticipants); + const chunkedParticipantsToCreate = chunk(participantsToCreate, 200); + const savedParticipants: CalendarEventParticipantWorkspaceEntity[] = []; - const savedParticipants = await calendarEventParticipantRepository.save( - participantsToSave, - {}, - transactionManager, - ); + for (const participantsToCreateChunk of chunkedParticipantsToCreate) { + const savedParticipantsChunk = + await calendarEventParticipantRepository.insert( + participantsToCreateChunk, + transactionManager, + ); + + savedParticipants.push(...savedParticipantsChunk.raw); + } if (calendarChannel.isContactAutoCreationEnabled) { await this.messageQueueService.add( @@ -154,6 +168,7 @@ export class CalendarEventParticipantService { participants: savedParticipants, objectMetadataName: 'calendarEventParticipant', transactionManager, + matchWith: 'workspaceMemberAndPerson', }); } } diff --git a/packages/twenty-server/src/modules/connected-account/refresh-tokens-manager/services/connected-account-refresh-tokens.service.ts b/packages/twenty-server/src/modules/connected-account/refresh-tokens-manager/services/connected-account-refresh-tokens.service.ts index 169c3cb649..fc665cd17b 100644 --- a/packages/twenty-server/src/modules/connected-account/refresh-tokens-manager/services/connected-account-refresh-tokens.service.ts +++ b/packages/twenty-server/src/modules/connected-account/refresh-tokens-manager/services/connected-account-refresh-tokens.service.ts @@ -17,6 +17,7 @@ import { ConnectedAccountRefreshAccessTokenExceptionCode, } from 'src/modules/connected-account/refresh-tokens-manager/exceptions/connected-account-refresh-tokens.exception'; import { ConnectedAccountWorkspaceEntity } from 'src/modules/connected-account/standard-objects/connected-account.workspace-entity'; +import { isAxiosTemporaryError } from 'src/modules/messaging/message-import-manager/drivers/gmail/utils/is-axios-gaxios-error.util'; export type ConnectedAccountTokens = GoogleTokens | MicrosoftTokens; @@ -98,29 +99,26 @@ export class ConnectedAccountRefreshTokensService { } } catch (error) { if (error?.name === 'AggregateError') { - const firstErrorCode = error?.errors?.[0]?.code; - const networkErrorCodes = [ - 'ENETUNREACH', - 'ETIMEDOUT', - 'ECONNABORTED', - 'ERR_NETWORK', - ]; - const isTemporaryNetworkError = - networkErrorCodes.includes(firstErrorCode); + const firstError = error?.errors?.[0]; - this.logger.log(error?.message); - this.logger.log(firstErrorCode); - this.logger.log(error?.errors); + this.logger.log(firstError); - if (isTemporaryNetworkError) { + if (isAxiosTemporaryError(error)) { throw new ConnectedAccountRefreshAccessTokenException( - `Error refreshing tokens for connected account ${connectedAccount.id.slice(0, 7)} in workspace ${workspaceId.slice(0, 7)}: ${firstErrorCode}`, + `Error refreshing tokens for connected account ${connectedAccount.id.slice(0, 7)} in workspace ${workspaceId.slice(0, 7)}: ${firstError.code}`, ConnectedAccountRefreshAccessTokenExceptionCode.TEMPORARY_NETWORK_ERROR, ); } - } else { - this.logger.log(error); } + + if (isAxiosTemporaryError(error)) { + throw new ConnectedAccountRefreshAccessTokenException( + `Error refreshing tokens for connected account ${connectedAccount.id.slice(0, 7)} in workspace ${workspaceId.slice(0, 7)}: ${error.code}`, + ConnectedAccountRefreshAccessTokenExceptionCode.TEMPORARY_NETWORK_ERROR, + ); + } + + this.logger.log(error); throw new ConnectedAccountRefreshAccessTokenException( `Error refreshing tokens for connected account ${connectedAccount.id.slice(0, 7)} in workspace ${workspaceId.slice(0, 7)}: ${error.message} ${error?.response?.data?.error_description}`, ConnectedAccountRefreshAccessTokenExceptionCode.REFRESH_ACCESS_TOKEN_FAILED, diff --git a/packages/twenty-server/src/modules/match-participant/match-participant.service.spec.ts b/packages/twenty-server/src/modules/match-participant/match-participant.service.spec.ts deleted file mode 100644 index 5d2d62d92d..0000000000 --- a/packages/twenty-server/src/modules/match-participant/match-participant.service.spec.ts +++ /dev/null @@ -1,748 +0,0 @@ -import { Test, TestingModule } from '@nestjs/testing'; - -import { WorkspaceEntityManager } from 'src/engine/twenty-orm/entity-manager/workspace-entity-manager'; -import { ScopedWorkspaceContextFactory } from 'src/engine/twenty-orm/factories/scoped-workspace-context.factory'; -import { TwentyORMGlobalManager } from 'src/engine/twenty-orm/twenty-orm-global.manager'; -import { WorkspaceEventEmitter } from 'src/engine/workspace-event-emitter/workspace-event-emitter'; -import { CalendarEventParticipantWorkspaceEntity } from 'src/modules/calendar/common/standard-objects/calendar-event-participant.workspace-entity'; -import { MatchParticipantService } from 'src/modules/match-participant/match-participant.service'; -import { MessageParticipantWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message-participant.workspace-entity'; -import { PersonWorkspaceEntity } from 'src/modules/person/standard-objects/person.workspace-entity'; -import { WorkspaceMemberWorkspaceEntity } from 'src/modules/workspace-member/standard-objects/workspace-member.workspace-entity'; - -describe('MatchParticipantService', () => { - let service: MatchParticipantService; - let twentyORMGlobalManager: TwentyORMGlobalManager; - let workspaceEventEmitter: WorkspaceEventEmitter; - let scopedWorkspaceContextFactory: ScopedWorkspaceContextFactory; - - let mockMessageParticipantRepository: { - find: jest.Mock; - update: jest.Mock; - createQueryBuilder: jest.Mock; - }; - let mockCalendarEventParticipantRepository: { - find: jest.Mock; - update: jest.Mock; - createQueryBuilder: jest.Mock; - }; - let mockPersonRepository: { - find: jest.Mock; - createQueryBuilder: jest.Mock; - }; - let mockWorkspaceMemberRepository: { - find: jest.Mock; - }; - let mockTransactionManager: WorkspaceEntityManager; - - const mockWorkspaceId = 'test-workspace-id'; - - beforeEach(async () => { - mockMessageParticipantRepository = { - find: jest.fn(), - update: jest.fn(), - createQueryBuilder: jest.fn().mockReturnValue({ - select: jest.fn().mockReturnThis(), - where: jest.fn().mockReturnThis(), - andWhere: jest.fn().mockReturnThis(), - orWhere: jest.fn().mockReturnThis(), - orderBy: jest.fn().mockReturnThis(), - getMany: jest.fn(), - withDeleted: jest.fn().mockReturnThis(), - }), - }; - - mockCalendarEventParticipantRepository = { - find: jest.fn(), - update: jest.fn(), - createQueryBuilder: jest.fn().mockReturnValue({ - select: jest.fn().mockReturnThis(), - where: jest.fn().mockReturnThis(), - andWhere: jest.fn().mockReturnThis(), - orWhere: jest.fn().mockReturnThis(), - orderBy: jest.fn().mockReturnThis(), - getMany: jest.fn(), - withDeleted: jest.fn().mockReturnThis(), - }), - }; - - mockPersonRepository = { - find: jest.fn(), - createQueryBuilder: jest.fn().mockReturnValue({ - select: jest.fn().mockReturnThis(), - where: jest.fn().mockReturnThis(), - andWhere: jest.fn().mockReturnThis(), - orWhere: jest.fn().mockReturnThis(), - orderBy: jest.fn().mockReturnThis(), - getMany: jest.fn(), - withDeleted: jest.fn().mockReturnThis(), - }), - }; - - mockWorkspaceMemberRepository = { - find: jest.fn(), - }; - - mockTransactionManager = {} as WorkspaceEntityManager; - - const module: TestingModule = await Test.createTestingModule({ - providers: [ - MatchParticipantService, - { - provide: TwentyORMGlobalManager, - useValue: { - getRepositoryForWorkspace: jest - .fn() - .mockImplementation((_workspaceId, entityName) => { - switch (entityName) { - case 'messageParticipant': - return mockMessageParticipantRepository; - case 'calendarEventParticipant': - return mockCalendarEventParticipantRepository; - case 'person': - return mockPersonRepository; - case 'workspaceMember': - return mockWorkspaceMemberRepository; - default: - return {}; - } - }), - }, - }, - { - provide: WorkspaceEventEmitter, - useValue: { - emitCustomBatchEvent: jest.fn(), - }, - }, - { - provide: ScopedWorkspaceContextFactory, - useValue: { - create: jest.fn().mockReturnValue({ - workspaceId: mockWorkspaceId, - }), - }, - }, - ], - }).compile(); - - service = module.get< - MatchParticipantService - >(MatchParticipantService); - twentyORMGlobalManager = module.get( - TwentyORMGlobalManager, - ); - workspaceEventEmitter = module.get( - WorkspaceEventEmitter, - ); - scopedWorkspaceContextFactory = module.get( - ScopedWorkspaceContextFactory, - ); - }); - - afterEach(() => { - jest.clearAllMocks(); - }); - - it('should be defined', () => { - expect(service).toBeDefined(); - }); - - describe('matchParticipants', () => { - const mockParticipants = [ - { - id: 'participant-1', - handle: 'test-1@example.com', - displayName: 'Test User', - }, - { - id: 'participant-2', - handle: 'test-2@company.com', - displayName: 'Contact', - }, - ] as MessageParticipantWorkspaceEntity[]; - - const mockPeople = [ - { - id: 'person-1', - emails: { - primaryEmail: 'test-1@example.com', - additionalEmails: ['test.alias@example.com'], - }, - }, - { - id: 'person-2', - emails: { - primaryEmail: 'test-2@company.com', - additionalEmails: ['test-2.alias@company.com'], - }, - }, - ] as PersonWorkspaceEntity[]; - - const mockWorkspaceMembers = [ - { - id: 'workspace-member-1', - userEmail: 'test-1@example.com', - }, - ] as WorkspaceMemberWorkspaceEntity[]; - - beforeEach(() => { - const mockQueryBuilder = { - select: jest.fn().mockReturnThis(), - where: jest.fn().mockReturnThis(), - andWhere: jest.fn().mockReturnThis(), - orWhere: jest.fn().mockReturnThis(), - orderBy: jest.fn().mockReturnThis(), - getMany: jest.fn().mockResolvedValue(mockPeople), - withDeleted: jest.fn().mockReturnThis(), - }; - - mockPersonRepository.createQueryBuilder.mockReturnValue(mockQueryBuilder); - mockQueryBuilder.getMany.mockResolvedValue(mockPeople); - mockWorkspaceMemberRepository.find.mockResolvedValue( - mockWorkspaceMembers, - ); - mockMessageParticipantRepository.update.mockResolvedValue({ - affected: 1, - }); - mockMessageParticipantRepository.find.mockResolvedValue(mockParticipants); - }); - - it('should match participants with people by primary email', async () => { - await service.matchParticipants({ - participants: mockParticipants, - objectMetadataName: 'messageParticipant', - }); - - expect(mockMessageParticipantRepository.update).toHaveBeenCalledWith( - { - id: expect.any(Object), - handle: 'test-1@example.com', - }, - { - personId: 'person-1', - workspaceMemberId: 'workspace-member-1', - }, - undefined, - ); - }); - - it('should match participants with people by additional email', async () => { - await service.matchParticipants({ - participants: mockParticipants, - objectMetadataName: 'messageParticipant', - }); - - expect(mockMessageParticipantRepository.update).toHaveBeenCalledWith( - { - id: expect.any(Object), - handle: 'test-2@company.com', - }, - { - personId: 'person-2', - workspaceMemberId: undefined, - }, - undefined, - ); - }); - - it('should emit matched event after successful matching', async () => { - await service.matchParticipants({ - participants: mockParticipants, - objectMetadataName: 'messageParticipant', - }); - - expect(workspaceEventEmitter.emitCustomBatchEvent).toHaveBeenCalledWith( - 'messageParticipant_matched', - [ - { - workspaceMemberId: null, - participants: mockParticipants, - }, - ], - mockWorkspaceId, - ); - }); - - it('should work with calendar event participants', async () => { - const calendarParticipants = [ - { - id: 'calendar-participant-1', - handle: 'test-1@example.com', - displayName: 'Test User', - isOrganizer: false, - responseStatus: 'ACCEPTED', - }, - { - id: 'calendar-participant-2', - handle: 'test-2@company.com', - displayName: 'Contact', - isOrganizer: false, - responseStatus: 'ACCEPTED', - }, - ] as CalendarEventParticipantWorkspaceEntity[]; - - const calendarService = - new MatchParticipantService( - workspaceEventEmitter, - twentyORMGlobalManager, - scopedWorkspaceContextFactory, - ); - - mockCalendarEventParticipantRepository.update.mockResolvedValue({ - affected: 1, - }); - mockCalendarEventParticipantRepository.find.mockResolvedValue( - calendarParticipants, - ); - - await calendarService.matchParticipants({ - participants: calendarParticipants, - objectMetadataName: 'calendarEventParticipant', - }); - - expect(mockCalendarEventParticipantRepository.update).toHaveBeenCalled(); - expect(workspaceEventEmitter.emitCustomBatchEvent).toHaveBeenCalledWith( - 'calendarEventParticipant_matched', - expect.any(Array), - mockWorkspaceId, - ); - }); - - it('should handle participants with no matching people or workspace members', async () => { - const mockQueryBuilder = { - select: jest.fn().mockReturnThis(), - where: jest.fn().mockReturnThis(), - andWhere: jest.fn().mockReturnThis(), - orWhere: jest.fn().mockReturnThis(), - orderBy: jest.fn().mockReturnThis(), - getMany: jest.fn().mockResolvedValue([]), - withDeleted: jest.fn().mockReturnThis(), - }; - - mockPersonRepository.createQueryBuilder.mockReturnValue(mockQueryBuilder); - mockQueryBuilder.getMany.mockResolvedValue([]); - mockWorkspaceMemberRepository.find.mockResolvedValue([]); - - await service.matchParticipants({ - participants: mockParticipants, - objectMetadataName: 'messageParticipant', - }); - - expect(mockMessageParticipantRepository.update).toHaveBeenCalledWith( - expect.any(Object), - { - personId: undefined, - workspaceMemberId: undefined, - }, - undefined, - ); - }); - - it('should throw error when workspace ID is not found', async () => { - scopedWorkspaceContextFactory.create = jest.fn().mockReturnValue({ - workspaceId: null, - }); - - await expect( - service.matchParticipants({ - participants: mockParticipants, - objectMetadataName: 'messageParticipant', - }), - ).rejects.toThrow('Workspace ID is required'); - }); - - it('should use transaction manager when provided', async () => { - await service.matchParticipants({ - participants: mockParticipants, - objectMetadataName: 'messageParticipant', - transactionManager: mockTransactionManager, - }); - - expect(mockWorkspaceMemberRepository.find).toHaveBeenCalledWith( - expect.any(Object), - mockTransactionManager, - ); - }); - }); - - describe('matchParticipantsAfterPersonOrWorkspaceMemberCreation', () => { - const mockExistingParticipants = [ - { - id: 'participant-1', - handle: 'test-1@example.com', - person: null, - }, - { - id: 'participant-2', - handle: 'test-2@company.com', - person: { - id: 'existing-person', - emails: { - primaryEmail: 'test-2@company.com', - additionalEmails: ['test-2.alias@company.com'], - }, - }, - }, - ] as MessageParticipantWorkspaceEntity[]; - - beforeEach(() => { - mockMessageParticipantRepository.find.mockResolvedValue( - mockExistingParticipants, - ); - mockMessageParticipantRepository.update.mockResolvedValue({ - affected: 1, - }); - }); - - describe('person matching', () => { - it('should match unmatched participants to new person', async () => { - await service.matchParticipantsAfterPersonCreation({ - handle: 'test-1@example.com', - isPrimaryEmail: true, - objectMetadataName: 'messageParticipant', - personId: 'new-person-id', - }); - - expect(mockMessageParticipantRepository.update).toHaveBeenCalledWith( - { - id: expect.any(Object), - }, - { - person: { - id: 'new-person-id', - }, - }, - ); - }); - - it('should re-match participants when new person has primary email and existing person has secondary', async () => { - await service.matchParticipantsAfterPersonCreation({ - handle: 'test-2@company.com', - isPrimaryEmail: true, - objectMetadataName: 'messageParticipant', - personId: 'new-person-id', - }); - - expect(mockMessageParticipantRepository.update).toHaveBeenCalledWith( - { - id: expect.any(Object), - }, - { - person: { - id: 'new-person-id', - }, - }, - ); - }); - - it('should not re-match when existing person has primary email', async () => { - const participantsWithPrimaryEmail = [ - { - id: 'participant-1', - handle: 'test-1@example.com', - person: { - id: 'existing-person', - emails: { - primaryEmail: 'test-1@example.com', - additionalEmails: [], - }, - }, - }, - ] as MessageParticipantWorkspaceEntity[]; - - mockMessageParticipantRepository.find.mockResolvedValue( - participantsWithPrimaryEmail, - ); - - await service.matchParticipantsAfterPersonCreation({ - handle: 'test-1@example.com', - isPrimaryEmail: false, - objectMetadataName: 'messageParticipant', - personId: 'new-person-id', - }); - - expect(mockMessageParticipantRepository.update).not.toHaveBeenCalled(); - }); - - it('should not re-match when new email is secondary and existing person has secondary', async () => { - await service.matchParticipantsAfterPersonCreation({ - handle: 'test-1@example.com', - isPrimaryEmail: false, - objectMetadataName: 'messageParticipant', - personId: 'new-person-id', - }); - - expect(mockMessageParticipantRepository.update).toHaveBeenCalledTimes( - 1, - ); - }); - - it('should emit matched event when participants are updated', async () => { - const updatedParticipants = [mockExistingParticipants[0]]; - - mockMessageParticipantRepository.find - .mockResolvedValueOnce(mockExistingParticipants) - .mockResolvedValueOnce(updatedParticipants); - - await service.matchParticipantsAfterPersonCreation({ - handle: 'test-1@example.com', - isPrimaryEmail: true, - objectMetadataName: 'messageParticipant', - personId: 'new-person-id', - }); - - expect(workspaceEventEmitter.emitCustomBatchEvent).toHaveBeenCalledWith( - 'messageParticipant_matched', - [ - { - workspaceId: mockWorkspaceId, - name: 'messageParticipant_matched', - workspaceMemberId: null, - participants: updatedParticipants, - }, - ], - mockWorkspaceId, - ); - }); - }); - - describe('workspace member matching', () => { - it('should match all participants to workspace member', async () => { - await service.matchParticipantsAfterWorkspaceMemberCreation({ - handle: 'test-1@example.com', - objectMetadataName: 'messageParticipant', - workspaceMemberId: 'workspace-member-id', - }); - - expect(mockMessageParticipantRepository.update).toHaveBeenCalledWith( - { - id: expect.any(Object), - }, - { - workspaceMember: { - id: 'workspace-member-id', - }, - }, - ); - }); - }); - - it('should throw error when workspace ID is not found', async () => { - scopedWorkspaceContextFactory.create = jest.fn().mockReturnValue({ - workspaceId: null, - }); - - await expect( - service.matchParticipantsAfterPersonCreation({ - handle: 'test-1@example.com', - isPrimaryEmail: true, - objectMetadataName: 'messageParticipant', - personId: 'person-id', - }), - ).rejects.toThrow('Workspace ID is required'); - }); - }); - - describe('unmatchParticipants', () => { - beforeEach(() => { - mockMessageParticipantRepository.update.mockResolvedValue({ - affected: 1, - }); - mockMessageParticipantRepository.find.mockResolvedValue([]); - const mockQueryBuilder = { - select: jest.fn().mockReturnThis(), - where: jest.fn().mockReturnThis(), - andWhere: jest.fn().mockReturnThis(), - orWhere: jest.fn().mockReturnThis(), - orderBy: jest.fn().mockReturnThis(), - getMany: jest.fn().mockResolvedValue([]), - withDeleted: jest.fn().mockReturnThis(), - }; - - mockPersonRepository.createQueryBuilder.mockReturnValue(mockQueryBuilder); - mockQueryBuilder.getMany.mockResolvedValue([]); - }); - - describe('person unmatching', () => { - it('should unmatch participants from person', async () => { - await service.unmatchParticipants({ - handle: 'test-1@example.com', - objectMetadataName: 'messageParticipant', - personId: 'person-id', - }); - - expect(mockMessageParticipantRepository.update).toHaveBeenCalledWith( - { - handle: expect.any(Object), - }, - { - person: null, - }, - ); - }); - - it('should re-match to next best person after unmatching', async () => { - const mockAlternativePeople = [ - { - id: 'alternative-person', - emails: { - primaryEmail: 'test-1@example.com', - additionalEmails: [], - }, - }, - ] as PersonWorkspaceEntity[]; - - const mockQueryBuilder = { - select: jest.fn().mockReturnThis(), - where: jest.fn().mockReturnThis(), - andWhere: jest.fn().mockReturnThis(), - orWhere: jest.fn().mockReturnThis(), - orderBy: jest.fn().mockReturnThis(), - getMany: jest.fn().mockResolvedValue(mockAlternativePeople), - withDeleted: jest.fn().mockReturnThis(), - }; - - mockPersonRepository.createQueryBuilder.mockReturnValue( - mockQueryBuilder, - ); - mockQueryBuilder.getMany.mockResolvedValue(mockAlternativePeople); - - const rematchedParticipants = [ - { - id: 'participant-1', - handle: 'test-1@example.com', - }, - ] as MessageParticipantWorkspaceEntity[]; - - mockMessageParticipantRepository.find.mockResolvedValue( - rematchedParticipants, - ); - - await service.unmatchParticipants({ - handle: 'test-1@example.com', - objectMetadataName: 'messageParticipant', - personId: 'old-person-id', - }); - - expect(mockMessageParticipantRepository.update).toHaveBeenCalledWith( - { - handle: expect.any(Object), - }, - { - person: null, - }, - ); - - expect(mockMessageParticipantRepository.update).toHaveBeenCalledWith( - { - handle: expect.any(Object), - }, - { - personId: 'alternative-person', - }, - ); - - expect(workspaceEventEmitter.emitCustomBatchEvent).toHaveBeenCalledWith( - 'messageParticipant_matched', - [ - { - workspaceMemberId: null, - participants: rematchedParticipants, - }, - ], - mockWorkspaceId, - ); - }); - - it('should not re-match when no alternative people found', async () => { - const mockQueryBuilder = { - select: jest.fn().mockReturnThis(), - where: jest.fn().mockReturnThis(), - andWhere: jest.fn().mockReturnThis(), - orWhere: jest.fn().mockReturnThis(), - orderBy: jest.fn().mockReturnThis(), - getMany: jest.fn().mockResolvedValue([]), - withDeleted: jest.fn().mockReturnThis(), - }; - - mockPersonRepository.createQueryBuilder.mockReturnValue( - mockQueryBuilder, - ); - mockQueryBuilder.getMany.mockResolvedValue([]); - - await service.unmatchParticipants({ - handle: 'test-1@example.com', - objectMetadataName: 'messageParticipant', - personId: 'person-id', - }); - - expect(mockMessageParticipantRepository.update).toHaveBeenCalledTimes( - 1, - ); - expect( - workspaceEventEmitter.emitCustomBatchEvent, - ).not.toHaveBeenCalled(); - }); - }); - - describe('workspace member unmatching', () => { - it('should unmatch participants from workspace member', async () => { - await service.unmatchParticipants({ - handle: 'test-1@example.com', - objectMetadataName: 'messageParticipant', - workspaceMemberId: 'workspace-member-id', - }); - - expect(mockMessageParticipantRepository.update).toHaveBeenCalledWith( - { - handle: expect.any(Object), - }, - { - workspaceMember: null, - }, - ); - }); - }); - - it('should throw error when workspace ID is not found', async () => { - scopedWorkspaceContextFactory.create = jest.fn().mockReturnValue({ - workspaceId: null, - }); - - await expect( - service.unmatchParticipants({ - handle: 'test-1@example.com', - objectMetadataName: 'messageParticipant', - personId: 'person-id', - }), - ).rejects.toThrow('Workspace ID is required'); - }); - }); - - describe('getParticipantRepository', () => { - it('should return message participant repository for messageParticipant', async () => { - const repository = await (service as any).getParticipantRepository( - mockWorkspaceId, - 'messageParticipant', - ); - - expect( - twentyORMGlobalManager.getRepositoryForWorkspace, - ).toHaveBeenCalledWith(mockWorkspaceId, 'messageParticipant'); - expect(repository).toBe(mockMessageParticipantRepository); - }); - - it('should return calendar event participant repository for calendarEventParticipant', async () => { - const repository = await (service as any).getParticipantRepository( - mockWorkspaceId, - 'calendarEventParticipant', - ); - - expect( - twentyORMGlobalManager.getRepositoryForWorkspace, - ).toHaveBeenCalledWith(mockWorkspaceId, 'calendarEventParticipant'); - expect(repository).toBe(mockCalendarEventParticipantRepository); - }); - }); -}); diff --git a/packages/twenty-server/src/modules/match-participant/match-participant.service.ts b/packages/twenty-server/src/modules/match-participant/match-participant.service.ts index d8affc7d09..b026ab30b3 100644 --- a/packages/twenty-server/src/modules/match-participant/match-participant.service.ts +++ b/packages/twenty-server/src/modules/match-participant/match-participant.service.ts @@ -1,6 +1,8 @@ import { Injectable } from '@nestjs/common'; -import { Any, Equal } from 'typeorm'; +import chunk from 'lodash.chunk'; +import { isDefined } from 'twenty-shared/utils'; +import { Any, In } from 'typeorm'; import { WorkspaceEntityManager } from 'src/engine/twenty-orm/entity-manager/workspace-entity-manager'; import { ScopedWorkspaceContextFactory } from 'src/engine/twenty-orm/factories/scoped-workspace-context.factory'; @@ -13,6 +15,40 @@ import { MessageParticipantWorkspaceEntity } from 'src/modules/messaging/common/ import { PersonWorkspaceEntity } from 'src/modules/person/standard-objects/person.workspace-entity'; import { WorkspaceMemberWorkspaceEntity } from 'src/modules/workspace-member/standard-objects/workspace-member.workspace-entity'; +type ObjectMetadataName = 'messageParticipant' | 'calendarEventParticipant'; + +type MatchParticipantsForWorkspaceMembersArgs = { + participantMatching: { + workspaceMemberIds: string[]; + }; + objectMetadataName: ObjectMetadataName; +}; + +type MatchParticipantsForPeopleArgs = { + participantMatching: { + personIds: string[]; + personEmails: string[]; + }; + objectMetadataName: ObjectMetadataName; +}; + +type MatchParticipantsArgs< + ParticipantWorkspaceEntity extends + | Pick< + CalendarEventParticipantWorkspaceEntity, + 'id' | 'handle' | 'workspaceMemberId' | 'personId' | 'calendarEventId' + > + | Pick< + MessageParticipantWorkspaceEntity, + 'id' | 'handle' | 'workspaceMemberId' | 'personId' | 'messageId' + >, +> = { + participants: ParticipantWorkspaceEntity[]; + objectMetadataName: ObjectMetadataName; + transactionManager?: WorkspaceEntityManager; + matchWith: 'workspaceMemberOnly' | 'personOnly' | 'workspaceMemberAndPerson'; +}; + @Injectable() export class MatchParticipantService< ParticipantWorkspaceEntity extends @@ -46,29 +82,17 @@ export class MatchParticipantService< participants, objectMetadataName, transactionManager, - }: { - participants: ParticipantWorkspaceEntity[]; - objectMetadataName: 'messageParticipant' | 'calendarEventParticipant'; - transactionManager?: WorkspaceEntityManager; - }) { + matchWith = 'workspaceMemberAndPerson', + }: MatchParticipantsArgs) { if (participants.length === 0) { return; } const workspaceId = this.scopedWorkspaceContextFactory.create().workspaceId; - if (!workspaceId) { + if (!isDefined(workspaceId)) { throw new Error('Workspace ID is required'); } - const participantRepository = await this.getParticipantRepository( - workspaceId, - objectMetadataName, - ); - - const participantIds = participants.map((participant) => participant.id); - const uniqueParticipantsHandles = [ - ...new Set(participants.map((participant) => participant.handle)), - ]; const personRepository = await this.twentyORMGlobalManager.getRepositoryForWorkspace( @@ -77,14 +101,10 @@ export class MatchParticipantService< { shouldBypassPermissionChecks: true }, ); - const queryBuilder = addPersonEmailFiltersToQueryBuilder({ - queryBuilder: personRepository.createQueryBuilder('person'), - emails: uniqueParticipantsHandles, - }); - - const people = await queryBuilder - .orderBy('person.createdAt', 'ASC') - .getMany(); + const participantRepository = await this.getParticipantRepository( + workspaceId, + objectMetadataName, + ); const workspaceMemberRepository = await this.twentyORMGlobalManager.getRepositoryForWorkspace( @@ -92,240 +112,88 @@ export class MatchParticipantService< 'workspaceMember', ); - const workspaceMembers = await workspaceMemberRepository.find( - { - where: { - userEmail: Any(uniqueParticipantsHandles), - }, - }, - transactionManager, - ); + const chunkSize = 200; + const chunkedParticipants = chunk(participants, chunkSize); - for (const handle of uniqueParticipantsHandles) { - const person = findPersonByPrimaryOrAdditionalEmail({ - people, - email: handle, - }); - - const workspaceMember = workspaceMembers.find( - (workspaceMember) => workspaceMember.userEmail === handle, - ); - - await participantRepository.update( - { - id: Any(participantIds), - handle, - }, - { - personId: person?.id, - workspaceMemberId: workspaceMember?.id, - }, - transactionManager, - ); - } - - const matchedParticipants = await participantRepository.find( - { - where: { - id: Any(participantIds), - handle: Any(uniqueParticipantsHandles), - }, - }, - transactionManager, - ); - - this.workspaceEventEmitter.emitCustomBatchEvent( - `${objectMetadataName}_matched`, - [ - { - workspaceMemberId: null, - participants: matchedParticipants, - }, - ], - workspaceId, - ); - } - - public async unmatchParticipants({ - handle, - objectMetadataName, - personId, - workspaceMemberId, - }: { - handle: string; - objectMetadataName: 'messageParticipant' | 'calendarEventParticipant'; - personId?: string; - workspaceMemberId?: string; - }) { - const workspaceId = this.scopedWorkspaceContextFactory.create().workspaceId; - - if (!workspaceId) { - throw new Error('Workspace ID is required'); - } - const participantRepository = await this.getParticipantRepository( - workspaceId, - objectMetadataName, - ); - - if (personId) { - await participantRepository.update( - { - handle: Equal(handle), - }, - { - person: null, - }, - ); - - const personRepository = - await this.twentyORMGlobalManager.getRepositoryForWorkspace( - workspaceId, - 'person', - { shouldBypassPermissionChecks: true }, - ); + for (const participants of chunkedParticipants) { + const uniqueParticipantsHandles = [ + ...new Set(participants.map((participant) => participant.handle)), + ]; const queryBuilder = addPersonEmailFiltersToQueryBuilder({ queryBuilder: personRepository.createQueryBuilder('person'), - emails: [handle], - excludePersonIds: [personId], + emails: uniqueParticipantsHandles, }); - const peopleToMatch = await queryBuilder + const people = await queryBuilder .orderBy('person.createdAt', 'ASC') .getMany(); - if (peopleToMatch.length > 0) { - const bestMatch = findPersonByPrimaryOrAdditionalEmail({ - people: peopleToMatch, - email: handle, - }); - - if (bestMatch) { - await participantRepository.update( - { - handle: Equal(handle), - }, - { - personId: bestMatch.id, - }, - ); - - const rematchedParticipants = await participantRepository.find({ - where: { - handle: Equal(handle), - }, - }); - - this.workspaceEventEmitter.emitCustomBatchEvent( - `${objectMetadataName}_matched`, - [ - { - workspaceMemberId: null, - participants: rematchedParticipants, - }, - ], - workspaceId, - ); - } - } - } - - if (workspaceMemberId) { - await participantRepository.update( + const workspaceMembers = await workspaceMemberRepository.find( { - handle: Equal(handle), - }, - { - workspaceMember: null, - }, - ); - } - } - - public async matchParticipantsAfterPersonCreation({ - handle, - isPrimaryEmail, - personId, - objectMetadataName, - }: { - handle: string; - isPrimaryEmail: boolean; - personId: string; - objectMetadataName: 'messageParticipant' | 'calendarEventParticipant'; - }) { - const workspaceId = this.scopedWorkspaceContextFactory.create().workspaceId; - - if (!workspaceId) { - throw new Error('Workspace ID is required'); - } - - const participantRepository = await this.getParticipantRepository( - workspaceId, - objectMetadataName, - ); - - const participantsToUpdate = await participantRepository.find({ - where: { - handle: Equal(handle), - }, - relations: ['person'], - }); - - const participantIdsToMatchWithPerson: string[] = []; - - for (const participant of participantsToUpdate) { - const existingPerson = participant.person; - - if (!existingPerson) { - participantIdsToMatchWithPerson.push(participant.id); - continue; - } - - const isAssociatedToPrimaryEmail = - existingPerson.emails?.primaryEmail.toLowerCase() === - handle.toLowerCase(); - - if (isAssociatedToPrimaryEmail) { - continue; - } - - const isAssociatedToSecondaryEmail = - Array.isArray(existingPerson.emails?.additionalEmails) && - existingPerson.emails.additionalEmails.some( - (email) => email.toLowerCase() === handle.toLowerCase(), - ); - - if (isAssociatedToSecondaryEmail && isPrimaryEmail) { - participantIdsToMatchWithPerson.push(participant.id); - } - } - - if (participantIdsToMatchWithPerson.length > 0) { - await participantRepository.update( - { - id: Any(participantIdsToMatchWithPerson), - }, - { - person: { - id: personId, + where: { + userEmail: Any(uniqueParticipantsHandles), }, }, + transactionManager, ); - const updatedParticipants = await participantRepository.find({ - where: { - id: Any(participantIdsToMatchWithPerson), - }, - }); + const partipantsToBeUpdated = participants + .map((participant) => { + const person = findPersonByPrimaryOrAdditionalEmail({ + people, + email: participant.handle, + }); + + const workspaceMember = workspaceMembers.find( + (workspaceMember) => + workspaceMember.userEmail === participant.handle, + ); + + const shouldMatchWithPerson = + matchWith === 'workspaceMemberAndPerson' || + matchWith === 'personOnly'; + + const shouldMatchWithWorkspaceMember = + matchWith === 'workspaceMemberAndPerson' || + matchWith === 'workspaceMemberOnly'; + + const newParticipant = { + ...participant, + ...(shouldMatchWithPerson && { + personId: isDefined(person) ? person.id : null, + }), + ...(shouldMatchWithWorkspaceMember && { + workspaceMemberId: isDefined(workspaceMember) + ? workspaceMember.id + : null, + }), + }; + + if ( + newParticipant.personId === participant.personId && + newParticipant.workspaceMemberId === participant.workspaceMemberId + ) { + return null; + } + + return newParticipant; + }) + .filter(isDefined); + + await participantRepository.updateMany( + partipantsToBeUpdated.map((participant) => ({ + criteria: participant.id, + partialEntity: participant, + })), + ); this.workspaceEventEmitter.emitCustomBatchEvent( `${objectMetadataName}_matched`, [ { - workspaceId, - name: `${objectMetadataName}_matched`, workspaceMemberId: null, - participants: updatedParticipants, + participants: partipantsToBeUpdated, }, ], workspaceId, @@ -333,18 +201,13 @@ export class MatchParticipantService< } } - public async matchParticipantsAfterWorkspaceMemberCreation({ - handle, - workspaceMemberId, + public async matchParticipantsForWorkspaceMembers({ + participantMatching, objectMetadataName, - }: { - handle: string; - workspaceMemberId: string; - objectMetadataName: 'messageParticipant' | 'calendarEventParticipant'; - }) { + }: MatchParticipantsForWorkspaceMembersArgs) { const workspaceId = this.scopedWorkspaceContextFactory.create().workspaceId; - if (!workspaceId) { + if (!isDefined(workspaceId)) { throw new Error('Workspace ID is required'); } @@ -353,25 +216,78 @@ export class MatchParticipantService< objectMetadataName, ); - const participantsToUpdate = await participantRepository.find({ + const participants = await participantRepository.find({ where: { - handle: Equal(handle), + workspaceMemberId: In(participantMatching.workspaceMemberIds), }, }); - const participantIdsToMatchWithWorkspaceMember = participantsToUpdate.map( - (participant) => participant.id, + const tobeRematchedParticipants = participants.map((participant) => { + return { + ...participant, + workspaceMemberId: null, + }; + }); + + await this.matchParticipants({ + matchWith: 'workspaceMemberOnly', + participants: tobeRematchedParticipants as ParticipantWorkspaceEntity[], + objectMetadataName, + }); + } + + public async matchParticipantsForPeople({ + participantMatching, + objectMetadataName, + }: MatchParticipantsForPeopleArgs) { + const workspaceId = this.scopedWorkspaceContextFactory.create().workspaceId; + + if (!isDefined(workspaceId)) { + throw new Error('Workspace ID is required'); + } + + const participantRepository = await this.getParticipantRepository( + workspaceId, + objectMetadataName, ); - await participantRepository.update( - { - id: Any(participantIdsToMatchWithWorkspaceMember), - }, - { - workspaceMember: { - id: workspaceMemberId, + let participantsMatchingPersonEmails: ParticipantWorkspaceEntity[] = []; + let participantsMatchingPersonId: ParticipantWorkspaceEntity[] = []; + + if (participantMatching.personIds.length > 0) { + participantsMatchingPersonId = (await participantRepository.find({ + where: { + personId: In(participantMatching.personIds), }, - }, - ); + })) as ParticipantWorkspaceEntity[]; + } + + if (participantMatching.personEmails.length > 0) { + participantsMatchingPersonEmails = (await participantRepository.find({ + where: { + handle: In(participantMatching.personEmails), + }, + })) as ParticipantWorkspaceEntity[]; + } + + const uniqueParticipants = [ + ...new Set([ + ...participantsMatchingPersonId, + ...participantsMatchingPersonEmails, + ]), + ]; + + const tobeRematchedParticipants = uniqueParticipants.map((participant) => { + return { + ...participant, + personId: null, + }; + }); + + await this.matchParticipants({ + matchWith: 'personOnly', + participants: tobeRematchedParticipants, + objectMetadataName, + }); } } diff --git a/packages/twenty-server/src/modules/messaging/message-import-manager/crons/jobs/messaging-message-list-fetch.cron.job.ts b/packages/twenty-server/src/modules/messaging/message-import-manager/crons/jobs/messaging-message-list-fetch.cron.job.ts index e92f678146..290ff4e5df 100644 --- a/packages/twenty-server/src/modules/messaging/message-import-manager/crons/jobs/messaging-message-list-fetch.cron.job.ts +++ b/packages/twenty-server/src/modules/messaging/message-import-manager/crons/jobs/messaging-message-list-fetch.cron.job.ts @@ -19,7 +19,7 @@ import { MessagingMessageListFetchJobData, } from 'src/modules/messaging/message-import-manager/jobs/messaging-message-list-fetch.job'; -export const MESSAGING_MESSAGE_LIST_FETCH_CRON_PATTERN = '*/5 * * * *'; +export const MESSAGING_MESSAGE_LIST_FETCH_CRON_PATTERN = '*/2 * * * *'; @Processor(MessageQueue.cronQueue) export class MessagingMessageListFetchCronJob { diff --git a/packages/twenty-server/src/modules/messaging/message-import-manager/drivers/gmail/services/gmail-handle-error.service.ts b/packages/twenty-server/src/modules/messaging/message-import-manager/drivers/gmail/services/gmail-handle-error.service.ts index c821adbc83..06ef6f91c1 100644 --- a/packages/twenty-server/src/modules/messaging/message-import-manager/drivers/gmail/services/gmail-handle-error.service.ts +++ b/packages/twenty-server/src/modules/messaging/message-import-manager/drivers/gmail/services/gmail-handle-error.service.ts @@ -1,6 +1,10 @@ import { Injectable } from '@nestjs/common'; -import { parseGaxiosError } from 'src/modules/messaging/message-import-manager/drivers/gmail/utils/parse-gaxios-error.util'; +import { + MessageImportDriverException, + MessageImportDriverExceptionCode, +} from 'src/modules/messaging/message-import-manager/drivers/exceptions/message-import-driver.exception'; +import { isAxiosTemporaryError } from 'src/modules/messaging/message-import-manager/drivers/gmail/utils/is-axios-gaxios-error.util'; import { parseGmailMessageListFetchError } from 'src/modules/messaging/message-import-manager/drivers/gmail/utils/parse-gmail-message-list-fetch-error.util'; import { parseGmailMessagesImportError } from 'src/modules/messaging/message-import-manager/drivers/gmail/utils/parse-gmail-messages-import-error.util'; @@ -10,10 +14,11 @@ export class GmailHandleErrorService { // eslint-disable-next-line @typescript-eslint/no-explicit-any public handleGmailMessageListFetchError(error: any): void { - const gaxiosError = parseGaxiosError(error); - - if (gaxiosError) { - throw gaxiosError; + if (isAxiosTemporaryError(error)) { + throw new MessageImportDriverException( + error.message, + MessageImportDriverExceptionCode.TEMPORARY_ERROR, + ); } throw parseGmailMessageListFetchError(error); @@ -24,10 +29,11 @@ export class GmailHandleErrorService { error: any, messageExternalId: string, ): void { - const gaxiosError = parseGaxiosError(error); - - if (gaxiosError) { - throw gaxiosError; + if (isAxiosTemporaryError(error)) { + throw new MessageImportDriverException( + error.message, + MessageImportDriverExceptionCode.TEMPORARY_ERROR, + ); } const gmailError = parseGmailMessagesImportError(error, messageExternalId); diff --git a/packages/twenty-server/src/modules/messaging/message-import-manager/drivers/gmail/utils/__tests__/parse-gaxios-error.spec.ts b/packages/twenty-server/src/modules/messaging/message-import-manager/drivers/gmail/utils/__tests__/parse-gaxios-error.spec.ts index 81542690b0..7ee2cb3669 100644 --- a/packages/twenty-server/src/modules/messaging/message-import-manager/drivers/gmail/utils/__tests__/parse-gaxios-error.spec.ts +++ b/packages/twenty-server/src/modules/messaging/message-import-manager/drivers/gmail/utils/__tests__/parse-gaxios-error.spec.ts @@ -1,60 +1,46 @@ -import { - MessageImportDriverException, - MessageImportDriverExceptionCode, -} from 'src/modules/messaging/message-import-manager/drivers/exceptions/message-import-driver.exception'; import gaxiosErrorMocks from 'src/modules/messaging/message-import-manager/drivers/gmail/mocks/gaxios-error-mocks'; -import { parseGaxiosError } from 'src/modules/messaging/message-import-manager/drivers/gmail/utils/parse-gaxios-error.util'; +import { isAxiosTemporaryError } from 'src/modules/messaging/message-import-manager/drivers/gmail/utils/is-axios-gaxios-error.util'; describe('parseGaxiosError', () => { it('should return a MessageImportDriverException for ECONNRESET', () => { const error = gaxiosErrorMocks.getError('ECONNRESET'); - const result = parseGaxiosError(error); + const result = isAxiosTemporaryError(error); - expect(result).toBeInstanceOf(MessageImportDriverException); - expect(result?.message).toBe(error.message); - expect(result?.code).toBe(MessageImportDriverExceptionCode.TEMPORARY_ERROR); + expect(result).toBe(true); }); it('should return a MessageImportDriverException for ENOTFOUND', () => { const error = gaxiosErrorMocks.getError('ENOTFOUND'); - const result = parseGaxiosError(error); + const result = isAxiosTemporaryError(error); - expect(result).toBeInstanceOf(MessageImportDriverException); - expect(result?.message).toBe(error.message); - expect(result?.code).toBe(MessageImportDriverExceptionCode.TEMPORARY_ERROR); + expect(result).toBe(true); }); it('should return a MessageImportDriverException for ECONNABORTED', () => { const error = gaxiosErrorMocks.getError('ECONNABORTED'); - const result = parseGaxiosError(error); + const result = isAxiosTemporaryError(error); - expect(result).toBeInstanceOf(MessageImportDriverException); - expect(result?.message).toBe(error.message); - expect(result?.code).toBe(MessageImportDriverExceptionCode.TEMPORARY_ERROR); + expect(result).toBe(true); }); it('should return a MessageImportDriverException for ETIMEDOUT', () => { const error = gaxiosErrorMocks.getError('ETIMEDOUT'); - const result = parseGaxiosError(error); + const result = isAxiosTemporaryError(error); - expect(result).toBeInstanceOf(MessageImportDriverException); - expect(result?.message).toBe(error.message); - expect(result?.code).toBe(MessageImportDriverExceptionCode.TEMPORARY_ERROR); + expect(result).toBe(true); }); it('should return a MessageImportDriverException for ERR_NETWORK', () => { const error = gaxiosErrorMocks.getError('ERR_NETWORK'); - const result = parseGaxiosError(error); + const result = isAxiosTemporaryError(error); - expect(result).toBeInstanceOf(MessageImportDriverException); - expect(result?.message).toBe(error.message); - expect(result?.code).toBe(MessageImportDriverExceptionCode.TEMPORARY_ERROR); + expect(result).toBe(true); }); it('should return undefined for unknown error codes', () => { const error = { code: 'UNKNOWN_ERROR' } as any; - const result = parseGaxiosError(error); + const result = isAxiosTemporaryError(error); - expect(result).toBeUndefined(); + expect(result).toBe(false); }); }); diff --git a/packages/twenty-server/src/modules/messaging/message-import-manager/drivers/gmail/utils/parse-gaxios-error.util.ts b/packages/twenty-server/src/modules/messaging/message-import-manager/drivers/gmail/utils/is-axios-gaxios-error.util.ts similarity index 54% rename from packages/twenty-server/src/modules/messaging/message-import-manager/drivers/gmail/utils/parse-gaxios-error.util.ts rename to packages/twenty-server/src/modules/messaging/message-import-manager/drivers/gmail/utils/is-axios-gaxios-error.util.ts index 9c5986d36b..93ab673d5d 100644 --- a/packages/twenty-server/src/modules/messaging/message-import-manager/drivers/gmail/utils/parse-gaxios-error.util.ts +++ b/packages/twenty-server/src/modules/messaging/message-import-manager/drivers/gmail/utils/is-axios-gaxios-error.util.ts @@ -1,14 +1,8 @@ import { GaxiosError } from 'gaxios'; -import { - MessageImportDriverException, - MessageImportDriverExceptionCode, -} from 'src/modules/messaging/message-import-manager/drivers/exceptions/message-import-driver.exception'; import { MessageNetworkExceptionCode } from 'src/modules/messaging/message-import-manager/drivers/exceptions/message-network.exception'; -export const parseGaxiosError = ( - error: GaxiosError, -): MessageImportDriverException | undefined => { +export const isAxiosTemporaryError = (error: GaxiosError): boolean => { const { code } = error; switch (code) { @@ -18,12 +12,9 @@ export const parseGaxiosError = ( case MessageNetworkExceptionCode.ETIMEDOUT: case MessageNetworkExceptionCode.ERR_NETWORK: case MessageNetworkExceptionCode.EHOSTUNREACH: - return new MessageImportDriverException( - error.message, - MessageImportDriverExceptionCode.TEMPORARY_ERROR, - ); + return true; default: - return undefined; + return false; } }; diff --git a/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-message.service.ts b/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-message.service.ts index bdccd9ba82..077127c8e7 100644 --- a/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-message.service.ts +++ b/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-message.service.ts @@ -1,5 +1,7 @@ -import { Injectable } from '@nestjs/common'; +import { Injectable, Logger } from '@nestjs/common'; +import { isDefined } from 'twenty-shared/utils'; +import { In } from 'typeorm'; import { v4 } from 'uuid'; import { WorkspaceEntityManager } from 'src/engine/twenty-orm/entity-manager/workspace-entity-manager'; @@ -9,8 +11,33 @@ import { MessageThreadWorkspaceEntity } from 'src/modules/messaging/common/stand import { MessageWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message.workspace-entity'; import { MessageWithParticipants } from 'src/modules/messaging/message-import-manager/types/message'; +type MessageAccumulator = { + existingMessageInDB?: MessageWorkspaceEntity; + existingThreadInDB?: Pick; + existingMessageChannelMessageAssociationInDB?: MessageChannelMessageAssociationWorkspaceEntity; + messageToCreate?: Pick< + MessageWorkspaceEntity, + | 'id' + | 'headerMessageId' + | 'subject' + | 'receivedAt' + | 'text' + | 'messageThreadId' + >; + threadToCreate?: Pick; + messageChannelMessageAssociationToCreate?: Pick< + MessageChannelMessageAssociationWorkspaceEntity, + | 'messageChannelId' + | 'messageId' + | 'messageExternalId' + | 'messageThreadExternalId' + | 'direction' + >; +}; @Injectable() export class MessagingMessageService { + private readonly logger = new Logger(MessagingMessageService.name); + constructor(private readonly twentyORMManager: TwentyORMManager) {} public async saveMessagesWithinTransaction( @@ -36,125 +63,346 @@ export class MessagingMessageService { 'messageThread', ); - const messageExternalIdsAndIdsMap = new Map(); - const createdMessages: Partial[] = []; + const messageAccumulatorMap = new Map(); - for (const message of messages) { - const existingMessageChannelMessageAssociation = - await messageChannelMessageAssociationRepository.findOne( - { - where: { - messageExternalId: message.externalId, - messageChannelId: messageChannelId, - }, + const existingMessagesInDB = await messageRepository.find({ + where: { + headerMessageId: In(messages.map((message) => message.headerMessageId)), + }, + }); + + const messageChannelMessageAssociationsReferencingMessageThread = + await messageChannelMessageAssociationRepository.find( + { + where: { + messageThreadExternalId: In( + messages.map((message) => message.messageThreadExternalId), + ), + messageChannelId, }, - transactionManager, - ); + relations: ['message'], + }, + transactionManager, + ); - if (existingMessageChannelMessageAssociation) { - continue; - } - - const existingMessage = await messageRepository.findOne({ + const existingMessageChannelMessageAssociations = + await messageChannelMessageAssociationRepository.find({ where: { - headerMessageId: message.headerMessageId, + messageId: In(existingMessagesInDB.map((message) => message.id)), + messageChannelId, }, }); - if (existingMessage) { - const existingAssociation = - await messageChannelMessageAssociationRepository.findOne( - { - where: { - messageChannelId, - messageId: existingMessage.id, - }, - }, - transactionManager, - ); + await this.enrichMessageAccumulatorWithExistingMessages( + messages, + messageAccumulatorMap, + existingMessagesInDB, + ); - if (existingAssociation) { - await messageChannelMessageAssociationRepository.update( - { - id: existingAssociation.id, - }, - { - messageExternalId: message.externalId, - messageThreadExternalId: message.messageThreadExternalId, - }, - transactionManager, - ); - } else { - await messageChannelMessageAssociationRepository.insert( - { - messageChannelId, - messageId: existingMessage.id, - messageExternalId: message.externalId, - messageThreadExternalId: message.messageThreadExternalId, - }, - transactionManager, - ); - } + await this.enrichMessageAccumulatorWithExistingMessageThreadIds( + messages, + messageAccumulatorMap, + messageChannelMessageAssociationsReferencingMessageThread, + ); - continue; - } + await this.enrichMessageAccumulatorWithExistingMessageChannelMessageAssociations( + messages, + messageAccumulatorMap, + existingMessageChannelMessageAssociations, + ); - const existingThread = await messageThreadRepository.findOne( - { - where: { - messages: { - messageChannelMessageAssociations: { - messageThreadExternalId: message.messageThreadExternalId, - messageChannelId, - }, - }, - }, - }, - transactionManager, - ); + await this.enrichMessageAccumulatorWithMessageThreadToCreate( + messages, + messageAccumulatorMap, + ); - let newOrExistingMessageThreadId = existingThread?.id; + for (const message of messages) { + const messageAccumulator = messageAccumulatorMap.get(message.externalId); - if (!existingThread) { - newOrExistingMessageThreadId = v4(); - - await messageThreadRepository.insert( - { id: newOrExistingMessageThreadId }, - transactionManager, + if (!isDefined(messageAccumulator)) { + throw new Error( + `Message accumulator should reference the message, this should never happen`, ); } - const newMessageId = v4(); - const messageToCreate = { - id: newMessageId, - headerMessageId: message.headerMessageId, - subject: message.subject, - receivedAt: message.receivedAt, - text: message.text, - messageThreadId: newOrExistingMessageThreadId, - }; + const messageThreadId = + messageAccumulator.threadToCreate?.id ?? + messageAccumulator.existingThreadInDB?.id; - await messageRepository.insert(messageToCreate, transactionManager); + if (!isDefined(messageThreadId)) { + throw new Error( + `Message thread id should be defined, either in the threadToCreate or existingThreadInDB`, + ); + } - createdMessages.push(messageToCreate); + let newOrExistingMessageId: string; - messageExternalIdsAndIdsMap.set(message.externalId, newMessageId); + if (!isDefined(messageAccumulator.existingMessageInDB)) { + newOrExistingMessageId = v4(); - await messageChannelMessageAssociationRepository.insert( - { + const messageToCreate = { + id: newOrExistingMessageId, + headerMessageId: message.headerMessageId, + subject: message.subject, + receivedAt: message.receivedAt, + text: message.text, + messageThreadId, + }; + + messageAccumulator.messageToCreate = messageToCreate; + } else { + newOrExistingMessageId = messageAccumulator.existingMessageInDB.id; + } + + if ( + !isDefined( + messageAccumulator.existingMessageChannelMessageAssociationInDB, + ) + ) { + messageAccumulator.messageChannelMessageAssociationToCreate = { messageChannelId, - messageId: newMessageId, + messageId: newOrExistingMessageId, messageExternalId: message.externalId, messageThreadExternalId: message.messageThreadExternalId, direction: message.direction, - }, - transactionManager, - ); + }; + + messageAccumulatorMap.set(message.externalId, messageAccumulator); + } + } + + const messageThreadsToCreate = Array.from(messageAccumulatorMap.values()) + .map((accumulator) => accumulator.threadToCreate) + .filter(isDefined); + + await messageThreadRepository.insert( + messageThreadsToCreate, + transactionManager, + ); + + const messagesToCreate = Array.from(messageAccumulatorMap.values()) + .map((accumulator) => accumulator.messageToCreate) + .filter(isDefined); + + await messageRepository.insert(messagesToCreate, transactionManager); + + const messageChannelMessageAssociationsToCreate = Array.from( + messageAccumulatorMap.values(), + ) + .map( + (accumulator) => accumulator.messageChannelMessageAssociationToCreate, + ) + .filter(isDefined); + + await messageChannelMessageAssociationRepository.insert( + messageChannelMessageAssociationsToCreate, + transactionManager, + ); + + const messageExternalIdsAndIdsMap = new Map(); + + for (const [externalId, accumulator] of messageAccumulatorMap.entries()) { + if (isDefined(accumulator.messageToCreate)) { + messageExternalIdsAndIdsMap.set( + externalId, + accumulator.messageToCreate.id, + ); + } + + if (isDefined(accumulator.existingMessageInDB)) { + messageExternalIdsAndIdsMap.set( + externalId, + accumulator.existingMessageInDB.id, + ); + } } return { - createdMessages, + createdMessages: messagesToCreate, messageExternalIdsAndIdsMap, }; } + + private async enrichMessageAccumulatorWithExistingMessages( + messages: MessageWithParticipants[], + messageAccumulatorMap: Map, + existingMessagesInDB: MessageWorkspaceEntity[], + ) { + for (const message of messages) { + const existingMessage = existingMessagesInDB.find( + (existingMessage) => + existingMessage.headerMessageId === message.headerMessageId, + ); + + if (!isDefined(existingMessage)) { + messageAccumulatorMap.set(message.externalId, {}); + continue; + } + + messageAccumulatorMap.set(message.externalId, { + existingMessageInDB: existingMessage, + }); + } + } + + private async enrichMessageAccumulatorWithExistingMessageThreadIds( + messages: MessageWithParticipants[], + messageAccumulatorMap: Map, + messageChannelMessageAssociationsReferencingMessageThread: Pick< + MessageChannelMessageAssociationWorkspaceEntity, + 'messageThreadExternalId' | 'message' + >[], + ) { + for (const message of messages) { + const messageAccumulator = messageAccumulatorMap.get(message.externalId); + + if (!isDefined(messageAccumulator)) { + throw new Error( + `Message accumulator should reference the message, this should never happen`, + ); + } + + const messageChannelMessageAssociationReferencingMessageThread = + messageChannelMessageAssociationsReferencingMessageThread.find( + (association) => + association.messageThreadExternalId === + message.messageThreadExternalId, + ); + + const existingThreadIdInDBIfMessageIsExistingInDB = + messageAccumulator.existingMessageInDB?.messageThreadId; + const existingThreadIdInDBIfMessageIsReferencedInMessageChannelMessageAssociation = + messageChannelMessageAssociationReferencingMessageThread?.message + ?.messageThreadId; + + if (isDefined(existingThreadIdInDBIfMessageIsExistingInDB)) { + messageAccumulator.existingThreadInDB = { + id: existingThreadIdInDBIfMessageIsExistingInDB, + }; + } + + if ( + isDefined( + existingThreadIdInDBIfMessageIsReferencedInMessageChannelMessageAssociation, + ) + ) { + messageAccumulator.existingThreadInDB = { + id: existingThreadIdInDBIfMessageIsReferencedInMessageChannelMessageAssociation, + }; + } + + if ( + isDefined(existingThreadIdInDBIfMessageIsExistingInDB) && + isDefined( + existingThreadIdInDBIfMessageIsReferencedInMessageChannelMessageAssociation, + ) && + existingThreadIdInDBIfMessageIsExistingInDB !== + existingThreadIdInDBIfMessageIsReferencedInMessageChannelMessageAssociation + ) { + // TODO: this can be handled better + // If we find a messageThreadId different on the existingMessage (found by messageHeaderId which is cross channel) + // And on the the one associatied to the messageThreadExternalId (found by which is channel specific) + // this means that we have to channels that have imported messages separately and that this message is the first connection between the two channels + // we should merge messageThreads + this.logger.warn( + `Message thread id is different for the same message header id and message thread external id, this means that we have to channels that have imported messages separately and that this message is the first connection between the two channels, we should merge messageThreads`, + ); + } + + messageAccumulatorMap.set(message.externalId, messageAccumulator); + } + } + + private async enrichMessageAccumulatorWithExistingMessageChannelMessageAssociations( + messages: MessageWithParticipants[], + messageAccumulatorMap: Map, + existingMessageChannelMessageAssociations: MessageChannelMessageAssociationWorkspaceEntity[], + ) { + for (const message of messages) { + const messageAccumulator = messageAccumulatorMap.get(message.externalId); + + if (!isDefined(messageAccumulator)) { + throw new Error( + `Message accumulator should reference the message, this should never happen`, + ); + } + + const existingMessage = messageAccumulator.existingMessageInDB; + + if (!isDefined(existingMessage)) { + continue; + } + + const existingMessageChannelMessageAssociation = + existingMessageChannelMessageAssociations.find( + (association) => association.messageId === existingMessage.id, + ); + + if (existingMessageChannelMessageAssociation) { + messageAccumulatorMap.set(message.externalId, { + existingMessageInDB: existingMessage, + existingMessageChannelMessageAssociationInDB: + existingMessageChannelMessageAssociation, + }); + } + } + } + + private async enrichMessageAccumulatorWithMessageThreadToCreate( + messages: MessageWithParticipants[], + messageAccumulatorMap: Map, + ) { + for (const [index, message] of messages.entries()) { + const messageAccumulator = messageAccumulatorMap.get(message.externalId); + + if (!isDefined(messageAccumulator)) { + throw new Error( + `Message accumulator should reference the message, this should never happen`, + ); + } + + const previousMessageWithSameThreadExternalId = messages.find( + (otherMessage, otherMessageIndex) => + otherMessage.messageThreadExternalId === + message.messageThreadExternalId && otherMessageIndex < index, + ); + + let newOrExistingMessageThreadId: string | undefined; + + if (isDefined(messageAccumulator.existingThreadInDB)) { + newOrExistingMessageThreadId = messageAccumulator.existingThreadInDB.id; + } + + if (isDefined(previousMessageWithSameThreadExternalId)) { + const previousMessageAccumulator = messageAccumulatorMap.get( + previousMessageWithSameThreadExternalId.externalId, + ); + + const previousMessageThreadId = + previousMessageAccumulator?.threadToCreate?.id ?? + previousMessageAccumulator?.existingThreadInDB?.id; + + if (!isDefined(previousMessageThreadId)) { + throw new Error( + `Previous message should have a thread id, either in the messageToCreate or existingMessageInDB`, + ); + } + + newOrExistingMessageThreadId = previousMessageThreadId; + messageAccumulator.existingThreadInDB = { + id: previousMessageThreadId, + }; + } + + if (!isDefined(newOrExistingMessageThreadId)) { + newOrExistingMessageThreadId = v4(); + + messageAccumulator.threadToCreate = { + id: newOrExistingMessageThreadId, + }; + } + + messageAccumulatorMap.set(message.externalId, messageAccumulator); + } + } } diff --git a/packages/twenty-server/src/modules/messaging/message-participant-manager/jobs/message-participant-match-participant.job.ts b/packages/twenty-server/src/modules/messaging/message-participant-manager/jobs/message-participant-match-participant.job.ts index d19e9a9ff7..58f75340b5 100644 --- a/packages/twenty-server/src/modules/messaging/message-participant-manager/jobs/message-participant-match-participant.job.ts +++ b/packages/twenty-server/src/modules/messaging/message-participant-manager/jobs/message-participant-match-participant.job.ts @@ -8,10 +8,11 @@ import { MessageParticipantWorkspaceEntity } from 'src/modules/messaging/common/ export type MessageParticipantMatchParticipantJobData = { workspaceId: string; - isPrimaryEmail: boolean; - email: string; - personId?: string; - workspaceMemberId?: string; + participantMatching: { + personIds: string[]; + personEmails: string[]; + workspaceMemberIds: string[]; + }; }; @Processor({ @@ -25,25 +26,23 @@ export class MessageParticipantMatchParticipantJob { @Process(MessageParticipantMatchParticipantJob.name) async handle(data: MessageParticipantMatchParticipantJobData): Promise { - const { isPrimaryEmail, email, personId, workspaceMemberId } = data; + const { participantMatching } = data; - if (personId) { - await this.matchParticipantService.matchParticipantsAfterPersonCreation({ - handle: email, - isPrimaryEmail, + if ( + participantMatching.personIds.length > 0 || + participantMatching.personEmails.length > 0 + ) { + await this.matchParticipantService.matchParticipantsForPeople({ + participantMatching, objectMetadataName: 'messageParticipant', - personId, }); } - if (workspaceMemberId) { - await this.matchParticipantService.matchParticipantsAfterWorkspaceMemberCreation( - { - handle: email, - objectMetadataName: 'messageParticipant', - workspaceMemberId, - }, - ); + if (participantMatching.workspaceMemberIds.length > 0) { + await this.matchParticipantService.matchParticipantsForWorkspaceMembers({ + participantMatching, + objectMetadataName: 'messageParticipant', + }); } } } diff --git a/packages/twenty-server/src/modules/messaging/message-participant-manager/jobs/message-participant-unmatch-participant.job.ts b/packages/twenty-server/src/modules/messaging/message-participant-manager/jobs/message-participant-unmatch-participant.job.ts deleted file mode 100644 index 6fd9d4391c..0000000000 --- a/packages/twenty-server/src/modules/messaging/message-participant-manager/jobs/message-participant-unmatch-participant.job.ts +++ /dev/null @@ -1,38 +0,0 @@ -import { Scope } from '@nestjs/common'; - -import { Process } from 'src/engine/core-modules/message-queue/decorators/process.decorator'; -import { Processor } from 'src/engine/core-modules/message-queue/decorators/processor.decorator'; -import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants'; -import { MatchParticipantService } from 'src/modules/match-participant/match-participant.service'; -import { MessageParticipantWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message-participant.workspace-entity'; - -export type MessageParticipantUnmatchParticipantJobData = { - workspaceId: string; - email: string; - personId?: string; - workspaceMemberId?: string; -}; - -@Processor({ - queueName: MessageQueue.messagingQueue, - scope: Scope.REQUEST, -}) -export class MessageParticipantUnmatchParticipantJob { - constructor( - private readonly matchParticipantService: MatchParticipantService, - ) {} - - @Process(MessageParticipantUnmatchParticipantJob.name) - async handle( - data: MessageParticipantUnmatchParticipantJobData, - ): Promise { - const { email, personId, workspaceMemberId } = data; - - await this.matchParticipantService.unmatchParticipants({ - handle: email, - objectMetadataName: 'messageParticipant', - personId, - workspaceMemberId, - }); - } -} diff --git a/packages/twenty-server/src/modules/messaging/message-participant-manager/listeners/message-participant-person.listener.ts b/packages/twenty-server/src/modules/messaging/message-participant-manager/listeners/message-participant-person.listener.ts index e335e7da51..aac607e5bb 100644 --- a/packages/twenty-server/src/modules/messaging/message-participant-manager/listeners/message-participant-person.listener.ts +++ b/packages/twenty-server/src/modules/messaging/message-participant-manager/listeners/message-participant-person.listener.ts @@ -12,16 +12,10 @@ import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decora 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 { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type'; -import { computeChangedAdditionalEmails } from 'src/modules/contact-creation-manager/utils/compute-changed-additional-emails'; -import { hasPrimaryEmailChanged } from 'src/modules/contact-creation-manager/utils/has-primary-email-changed'; import { MessageParticipantMatchParticipantJob, MessageParticipantMatchParticipantJobData, } from 'src/modules/messaging/message-participant-manager/jobs/message-participant-match-participant.job'; -import { - MessageParticipantUnmatchParticipantJob, - MessageParticipantUnmatchParticipantJobData, -} from 'src/modules/messaging/message-participant-manager/jobs/message-participant-unmatch-participant.job'; import { PersonWorkspaceEntity } from 'src/modules/person/standard-objects/person.workspace-entity'; @Injectable() @@ -37,44 +31,34 @@ export class MessageParticipantPersonListener { ObjectRecordCreateEvent >, ) { - for (const eventPayload of payload.events) { - const jobPromises: Promise[] = []; + const personWithEmails = payload.events.filter( + (eventPayload) => + isDefined(eventPayload.properties.after.emails?.primaryEmail) || + isDefined(eventPayload.properties.after.emails?.additionalEmails), + ); - if (isDefined(eventPayload.properties.after.emails?.primaryEmail)) { - jobPromises.push( - this.messageQueueService.add( - MessageParticipantMatchParticipantJob.name, - { - workspaceId: payload.workspaceId, - email: eventPayload.properties.after.emails?.primaryEmail, - isPrimaryEmail: true, - personId: eventPayload.recordId, - }, - ), - ); - } + const personIds = personWithEmails.map( + (eventPayload) => eventPayload.recordId, + ); + const personEmails = personWithEmails + .flatMap((eventPayload) => [ + eventPayload.properties.after.emails.primaryEmail, + ...((eventPayload.properties.after.emails?.additionalEmails ?? + []) as string[]), + ]) + .filter(isDefined); - const additionalEmails = - eventPayload.properties.after.emails?.additionalEmails; - - if (Array.isArray(additionalEmails)) { - const additionalEmailPromises = additionalEmails.map((email) => - this.messageQueueService.add( - MessageParticipantMatchParticipantJob.name, - { - workspaceId: payload.workspaceId, - email: email, - isPrimaryEmail: false, - personId: eventPayload.recordId, - }, - ), - ); - - jobPromises.push(...additionalEmailPromises); - } - - await Promise.all(jobPromises); - } + await this.messageQueueService.add( + MessageParticipantMatchParticipantJob.name, + { + workspaceId: payload.workspaceId, + participantMatching: { + personIds, + personEmails, + workspaceMemberIds: [], + }, + }, + ); } @OnDatabaseBatchEvent('person', DatabaseEventAction.UPDATED) @@ -83,79 +67,35 @@ export class MessageParticipantPersonListener { ObjectRecordUpdateEvent >, ) { - for (const eventPayload of payload.events) { - if ( - objectRecordUpdateEventChangedProperties( - eventPayload.properties.before, - eventPayload.properties.after, - ).includes('emails') - ) { - if (!isDefined(eventPayload.properties.diff)) { - continue; - } + const personWithEmails = payload.events.filter((eventPayload) => + objectRecordUpdateEventChangedProperties( + eventPayload.properties.before, + eventPayload.properties.after, + ).includes('emails'), + ); - const jobPromises: Promise[] = []; + const personIds = personWithEmails.map( + (eventPayload) => eventPayload.recordId, + ); + const personEmails = personWithEmails + .flatMap((eventPayload) => [ + eventPayload.properties.after.emails.primaryEmail, + ...((eventPayload.properties.after.emails?.additionalEmails ?? + []) as string[]), + ]) + .filter(isDefined); - if (hasPrimaryEmailChanged(eventPayload.properties.diff)) { - if (eventPayload.properties.before.emails?.primaryEmail) { - jobPromises.push( - this.messageQueueService.add( - MessageParticipantUnmatchParticipantJob.name, - { - workspaceId: payload.workspaceId, - email: eventPayload.properties.before.emails?.primaryEmail, - personId: eventPayload.recordId, - }, - ), - ); - } - - if (eventPayload.properties.after.emails?.primaryEmail) { - jobPromises.push( - this.messageQueueService.add( - MessageParticipantMatchParticipantJob.name, - { - workspaceId: payload.workspaceId, - email: eventPayload.properties.after.emails?.primaryEmail, - isPrimaryEmail: true, - personId: eventPayload.recordId, - }, - ), - ); - } - } - - const { addedAdditionalEmails, removedAdditionalEmails } = - computeChangedAdditionalEmails(eventPayload.properties.diff); - - const removedEmailPromises = removedAdditionalEmails.map((email) => - this.messageQueueService.add( - MessageParticipantUnmatchParticipantJob.name, - { - workspaceId: payload.workspaceId, - email: email, - personId: eventPayload.recordId, - }, - ), - ); - - const addedEmailPromises = addedAdditionalEmails.map((email) => - this.messageQueueService.add( - MessageParticipantMatchParticipantJob.name, - { - workspaceId: payload.workspaceId, - email: email, - isPrimaryEmail: false, - personId: eventPayload.recordId, - }, - ), - ); - - jobPromises.push(...removedEmailPromises, ...addedEmailPromises); - - await Promise.all(jobPromises); - } - } + await this.messageQueueService.add( + MessageParticipantMatchParticipantJob.name, + { + workspaceId: payload.workspaceId, + participantMatching: { + personIds, + personEmails, + workspaceMemberIds: [], + }, + }, + ); } @OnDatabaseBatchEvent('person', DatabaseEventAction.DESTROYED) @@ -164,33 +104,28 @@ export class MessageParticipantPersonListener { ObjectRecordDeleteEvent >, ) { - for (const eventPayload of payload.events) { - await this.messageQueueService.add( - MessageParticipantUnmatchParticipantJob.name, - { - workspaceId: payload.workspaceId, - email: eventPayload.properties.before.emails?.primaryEmail, - personId: eventPayload.recordId, + const personWithEmails = payload.events.filter( + (eventPayload) => + isDefined(eventPayload.properties.before.emails?.primaryEmail) || + isDefined(eventPayload.properties.before.emails?.additionalEmails), + ); + + await this.messageQueueService.add( + MessageParticipantMatchParticipantJob.name, + { + workspaceId: payload.workspaceId, + participantMatching: { + personIds: [], + personEmails: personWithEmails + .flatMap((eventPayload) => [ + eventPayload.properties.before.emails.primaryEmail, + ...((eventPayload.properties.before.emails?.additionalEmails ?? + []) as string[]), + ]) + .filter(isDefined), + workspaceMemberIds: [], }, - ); - - const additionalEmails = - eventPayload.properties.before.emails?.additionalEmails; - - if (Array.isArray(additionalEmails)) { - const additionalEmailPromises = additionalEmails.map((email) => - this.messageQueueService.add( - MessageParticipantUnmatchParticipantJob.name, - { - workspaceId: payload.workspaceId, - email: email, - personId: eventPayload.recordId, - }, - ), - ); - - await Promise.all(additionalEmailPromises); - } - } + }, + ); } } diff --git a/packages/twenty-server/src/modules/messaging/message-participant-manager/listeners/message-participant-workspace-member.listener.ts b/packages/twenty-server/src/modules/messaging/message-participant-manager/listeners/message-participant-workspace-member.listener.ts index 53e706f8ca..2a6e68f6e2 100644 --- a/packages/twenty-server/src/modules/messaging/message-participant-manager/listeners/message-participant-workspace-member.listener.ts +++ b/packages/twenty-server/src/modules/messaging/message-participant-manager/listeners/message-participant-workspace-member.listener.ts @@ -18,10 +18,6 @@ import { MessageParticipantMatchParticipantJob, MessageParticipantMatchParticipantJobData, } from 'src/modules/messaging/message-participant-manager/jobs/message-participant-match-participant.job'; -import { - MessageParticipantUnmatchParticipantJob, - MessageParticipantUnmatchParticipantJobData, -} from 'src/modules/messaging/message-participant-manager/jobs/message-participant-unmatch-participant.job'; import { WorkspaceMemberWorkspaceEntity } from 'src/modules/workspace-member/standard-objects/workspace-member.workspace-entity'; @Injectable() @@ -59,9 +55,11 @@ export class MessageParticipantWorkspaceMemberListener { MessageParticipantMatchParticipantJob.name, { workspaceId: payload.workspaceId, - email: eventPayload.properties.after.userEmail, - workspaceMemberId: eventPayload.recordId, - isPrimaryEmail: true, + participantMatching: { + personIds: [], + personEmails: [], + workspaceMemberIds: [eventPayload.recordId], + }, }, ); } @@ -80,22 +78,15 @@ export class MessageParticipantWorkspaceMemberListener { eventPayload.properties.after, ).includes('userEmail') ) { - await this.messageQueueService.add( - MessageParticipantUnmatchParticipantJob.name, - { - workspaceId: payload.workspaceId, - email: eventPayload.properties.before.userEmail, - personId: eventPayload.recordId, - }, - ); - await this.messageQueueService.add( MessageParticipantMatchParticipantJob.name, { workspaceId: payload.workspaceId, - email: eventPayload.properties.after.userEmail, - workspaceMemberId: eventPayload.recordId, - isPrimaryEmail: true, + participantMatching: { + personIds: [], + personEmails: [], + workspaceMemberIds: [eventPayload.recordId], + }, }, ); } diff --git a/packages/twenty-server/src/modules/messaging/message-participant-manager/message-participant-manager.module.ts b/packages/twenty-server/src/modules/messaging/message-participant-manager/message-participant-manager.module.ts index 5df60ef8d8..c3419cd248 100644 --- a/packages/twenty-server/src/modules/messaging/message-participant-manager/message-participant-manager.module.ts +++ b/packages/twenty-server/src/modules/messaging/message-participant-manager/message-participant-manager.module.ts @@ -11,7 +11,6 @@ import { ContactCreationManagerModule } from 'src/modules/contact-creation-manag import { MatchParticipantModule } from 'src/modules/match-participant/match-participant.module'; import { MessagingCommonModule } from 'src/modules/messaging/common/messaging-common.module'; import { MessageParticipantMatchParticipantJob } from 'src/modules/messaging/message-participant-manager/jobs/message-participant-match-participant.job'; -import { MessageParticipantUnmatchParticipantJob } from 'src/modules/messaging/message-participant-manager/jobs/message-participant-unmatch-participant.job'; import { MessagingCreateCompanyAndContactAfterSyncJob } from 'src/modules/messaging/message-participant-manager/jobs/messaging-create-company-and-contact-after-sync.job'; import { MessageParticipantPersonListener } from 'src/modules/messaging/message-participant-manager/listeners/message-participant-person.listener'; import { MessageParticipantWorkspaceMemberListener } from 'src/modules/messaging/message-participant-manager/listeners/message-participant-workspace-member.listener'; @@ -37,7 +36,6 @@ import { TimelineActivityWorkspaceEntity } from 'src/modules/timeline/standard-o providers: [ MessagingMessageParticipantService, MessageParticipantMatchParticipantJob, - MessageParticipantUnmatchParticipantJob, MessagingCreateCompanyAndContactAfterSyncJob, MessageParticipantListener, MessageParticipantPersonListener, diff --git a/packages/twenty-server/src/modules/messaging/message-participant-manager/services/messaging-message-participant.service.ts b/packages/twenty-server/src/modules/messaging/message-participant-manager/services/messaging-message-participant.service.ts index 4f29a7b155..95d95a846a 100644 --- a/packages/twenty-server/src/modules/messaging/message-participant-manager/services/messaging-message-participant.service.ts +++ b/packages/twenty-server/src/modules/messaging/message-participant-manager/services/messaging-message-participant.service.ts @@ -1,5 +1,7 @@ import { Injectable } from '@nestjs/common'; +import { In } from 'typeorm'; + import { WorkspaceEntityManager } from 'src/engine/twenty-orm/entity-manager/workspace-entity-manager'; import { TwentyORMManager } from 'src/engine/twenty-orm/twenty-orm.manager'; import { MatchParticipantService } from 'src/modules/match-participant/match-participant.service'; @@ -22,23 +24,48 @@ export class MessagingMessageParticipantService { 'messageParticipant', ); - const savedParticipants = await messageParticipantRepository.save( - participants.map((participant) => { + const existingParticipantsBasedOnMessageIds = + await messageParticipantRepository.find({ + where: { + messageId: In( + participants.map((participant) => participant.messageId), + ), + }, + }); + + const participantsToCreate: Pick< + MessageParticipantWorkspaceEntity, + 'messageId' | 'handle' | 'displayName' | 'role' + >[] = participants + .filter( + (participant) => + !existingParticipantsBasedOnMessageIds.find( + (existingParticipant) => + existingParticipant.messageId === participant.messageId && + existingParticipant.handle === participant.handle && + existingParticipant.displayName === participant.displayName && + existingParticipant.role === participant.role, + ), + ) + .map((participant) => { return { messageId: participant.messageId, - role: participant.role, handle: participant.handle, displayName: participant.displayName, + role: participant.role, }; - }), - {}, + }); + + const createdParticipants = await messageParticipantRepository.insert( + participantsToCreate, transactionManager, ); await this.matchParticipantService.matchParticipants({ - participants: savedParticipants, + participants: createdParticipants.raw ?? [], objectMetadataName: 'messageParticipant', transactionManager, + matchWith: 'workspaceMemberAndPerson', }); } } diff --git a/packages/twenty-server/src/modules/workflow/workflow-trigger/automated-trigger/listeners/database-event-trigger.listener.ts b/packages/twenty-server/src/modules/workflow/workflow-trigger/automated-trigger/listeners/database-event-trigger.listener.ts index 3a2450e079..c56e7fb4a0 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-trigger/automated-trigger/listeners/database-event-trigger.listener.ts +++ b/packages/twenty-server/src/modules/workflow/workflow-trigger/automated-trigger/listeners/database-event-trigger.listener.ts @@ -1,7 +1,9 @@ import { Injectable, Logger } from '@nestjs/common'; import { isDefined } from 'twenty-shared/utils'; -import { Raw } from 'typeorm'; +import { In, Raw } from 'typeorm'; + +import { ObjectRecord } from 'src/engine/api/graphql/workspace-query-builder/interfaces/object-record.interface'; import { OnDatabaseBatchEvent } from 'src/engine/api/graphql/graphql-query-runner/decorators/on-database-batch-event.decorator'; import { DatabaseEventAction } from 'src/engine/api/graphql/graphql-query-runner/enums/database-event-action'; @@ -111,13 +113,11 @@ export class DatabaseEventTriggerListener { ) { const workspaceId = payload.workspaceId; - for (const event of payload.events) { - await this.enrichRecord({ - event, - record: event.properties.after, - workspaceId, - }); - } + await this.enrichRecordsWithRelations({ + records: payload.events.map((event) => event.properties.after), + objectMetadataNameSingular: payload.events[0].objectMetadata.nameSingular, + workspaceId, + }); } private async enrichUpdatedEvent( @@ -125,18 +125,16 @@ export class DatabaseEventTriggerListener { ) { const workspaceId = payload.workspaceId; - for (const event of payload.events) { - await this.enrichRecord({ - event, - record: event.properties.before, - workspaceId, - }); - await this.enrichRecord({ - event, - record: event.properties.after, - workspaceId, - }); - } + await this.enrichRecordsWithRelations({ + records: payload.events.map((event) => event.properties.before), + objectMetadataNameSingular: payload.events[0].objectMetadata.nameSingular, + workspaceId, + }); + await this.enrichRecordsWithRelations({ + records: payload.events.map((event) => event.properties.after), + objectMetadataNameSingular: payload.events[0].objectMetadata.nameSingular, + workspaceId, + }); } private async enrichDeletedEvent( @@ -144,13 +142,11 @@ export class DatabaseEventTriggerListener { ) { const workspaceId = payload.workspaceId; - for (const event of payload.events) { - await this.enrichRecord({ - event, - record: event.properties.before, - workspaceId, - }); - } + await this.enrichRecordsWithRelations({ + records: payload.events.map((event) => event.properties.before), + objectMetadataNameSingular: payload.events[0].objectMetadata.nameSingular, + workspaceId, + }); } private async enrichDestroyedEvent( @@ -158,28 +154,25 @@ export class DatabaseEventTriggerListener { ) { const workspaceId = payload.workspaceId; - for (const event of payload.events) { - await this.enrichRecord({ - event, - record: event.properties.before, - workspaceId, - }); - } + await this.enrichRecordsWithRelations({ + records: payload.events.map((event) => event.properties.before), + objectMetadataNameSingular: payload.events[0].objectMetadata.nameSingular, + workspaceId, + }); } - private async enrichRecord({ - event, - record, + private async enrichRecordsWithRelations({ + records, + objectMetadataNameSingular, workspaceId, }: { - event: ObjectRecordNonDestructiveEvent; - // eslint-disable-next-line @typescript-eslint/no-explicit-any - record: Record; + records: Partial[]; + objectMetadataNameSingular: string; workspaceId: string; }) { const { objectMetadataMaps, objectMetadataItemWithFieldsMaps } = await this.workflowCommonWorkspaceService.getObjectMetadataItemWithFieldsMaps( - event.objectMetadata.nameSingular, + objectMetadataNameSingular, workspaceId, ); @@ -188,9 +181,11 @@ export class DatabaseEventTriggerListener { )) { const joinField = objectMetadataItemWithFieldsMaps.fieldsById[joinFieldId]; - const joinRecordId = record[joinColumnName]; + const joinRecordIds = records + .map((record) => record[joinColumnName]) + .filter(isDefined); - if (!isDefined(joinRecordId)) { + if (joinRecordIds.length === 0) { continue; } @@ -214,9 +209,15 @@ export class DatabaseEventTriggerListener { { shouldBypassPermissionChecks: true }, ); - record[joinField.name] = await relatedObjectRepository.findOne({ - where: { id: joinRecordId }, + const relatedRecords = await relatedObjectRepository.find({ + where: { id: In(joinRecordIds) }, }); + + for (const record of records) { + record[joinField.name] = relatedRecords.find( + (relatedRecord) => relatedRecord.id === record[joinColumnName], + ); + } } }