5044 Dispatch createcontact job instead of emitting an event (#5135)

Closes #5044

packages/twenty-server/src/modules/calendar/services/google-calendar-sync/google-calendar-sync.service.ts

@@ -1,6 +1,5 @@
 import { Injectable, Logger } from '@nestjs/common';
 import { InjectRepository } from '@nestjs/typeorm';
-import { EventEmitter2 } from '@nestjs/event-emitter';
 
 import { Repository } from 'typeorm';
 import { calendar_v3 as calendarV3 } from 'googleapis';
@@ -29,6 +28,13 @@ import { CalendarEventCleanerService } from 'src/modules/calendar/services/calen
 import { CalendarEventParticipantService } from 'src/modules/calendar/services/calendar-event-participant/calendar-event-participant.service';
 import { CalendarEventParticipant } from 'src/modules/calendar/types/calendar-event';
 import { filterOutBlocklistedEvents } from 'src/modules/calendar/utils/filter-out-blocklisted-events.util';
+import {
+  CreateCompanyAndContactJobData,
+  CreateCompanyAndContactJob,
+} from 'src/modules/connected-account/auto-companies-and-contacts-creation/jobs/create-company-and-contact.job';
+import { InjectMessageQueue } from 'src/engine/integrations/message-queue/decorators/message-queue.decorator';
+import { MessageQueue } from 'src/engine/integrations/message-queue/message-queue.constants';
+import { MessageQueueService } from 'src/engine/integrations/message-queue/services/message-queue.service';
 
 @Injectable()
 export class GoogleCalendarSyncService {
@@ -53,9 +59,10 @@ export class GoogleCalendarSyncService {
     @InjectRepository(FeatureFlagEntity, 'core')
     private readonly featureFlagRepository: Repository<FeatureFlagEntity>,
     private readonly workspaceDataSourceService: WorkspaceDataSourceService,
-    private readonly eventEmitter: EventEmitter2,
     private readonly calendarEventCleanerService: CalendarEventCleanerService,
     private readonly calendarEventParticipantsService: CalendarEventParticipantService,
+    @InjectMessageQueue(MessageQueue.emailQueue)
+    private readonly messageQueueService: MessageQueueService,
   ) {}
 
   public async startGoogleCalendarSync(
@@ -369,11 +376,14 @@ export class GoogleCalendarSyncService {
         );
 
         if (calendarChannel.isContactAutoCreationEnabled) {
-          this.eventEmitter.emit(`createContacts`, {
+          await this.messageQueueService.add<CreateCompanyAndContactJobData>(
-            workspaceId,
+            CreateCompanyAndContactJob.name,
-            connectedAccountHandle: connectedAccount.handle,
+            {
-            contactsToCreate: participantsToSave,
+              workspaceId,
-          });
+              connectedAccountHandle: connectedAccount.handle,
+              contactsToCreate: participantsToSave,
+            },
+          );
         }
       } catch (error) {
         this.logger.error(

packages/twenty-server/src/modules/connected-account/auto-companies-and-contacts-creation/auto-companies-and-contacts-creation.module.ts

@@ -9,7 +9,6 @@ import { PersonObjectMetadata } from 'src/modules/person/standard-objects/person
 import { WorkspaceMemberObjectMetadata } from 'src/modules/workspace-member/standard-objects/workspace-member.object-metadata';
 import { MessageParticipantModule } from 'src/modules/messaging/services/message-participant/message-participant.module';
 import { WorkspaceDataSourceModule } from 'src/engine/workspace-datasource/workspace-datasource.module';
-import { CreateCompanyAndContactListener } from 'src/modules/connected-account/auto-companies-and-contacts-creation/listeners/create-company-and-contact.listener';
 import { CalendarEventParticipantObjectMetadata } from 'src/modules/calendar/standard-objects/calendar-event-participant.object-metadata';
 import { CalendarEventParticipantModule } from 'src/modules/calendar/services/calendar-event-participant/calendar-event-participant.module';
 import { FeatureFlagEntity } from 'src/engine/core-modules/feature-flag/feature-flag.entity';
@@ -28,7 +27,7 @@ import { FeatureFlagEntity } from 'src/engine/core-modules/feature-flag/feature-
     CalendarEventParticipantModule,
     TypeOrmModule.forFeature([FeatureFlagEntity], 'core'),
   ],
-  providers: [CreateCompanyAndContactService, CreateCompanyAndContactListener],
+  providers: [CreateCompanyAndContactService],
   exports: [CreateCompanyAndContactService],
 })
 export class AutoCompaniesAndContactsCreationModule {}

packages/twenty-server/src/modules/connected-account/auto-companies-and-contacts-creation/listeners/create-company-and-contact.listener.ts

@@ -1,32 +0,0 @@
-import { Injectable, Inject } from '@nestjs/common';
-import { OnEvent } from '@nestjs/event-emitter';
-
-import { MessageQueue } from 'src/engine/integrations/message-queue/message-queue.constants';
-import { MessageQueueService } from 'src/engine/integrations/message-queue/services/message-queue.service';
-import {
-  CreateCompanyAndContactJobData,
-  CreateCompanyAndContactJob,
-} from 'src/modules/connected-account/auto-companies-and-contacts-creation/jobs/create-company-and-contact.job';
-
-@Injectable()
-export class CreateCompanyAndContactListener {
-  constructor(
-    @Inject(MessageQueue.contactCreationQueue)
-    private readonly messageQueueService: MessageQueueService,
-  ) {}
-
-  @OnEvent('createContacts')
-  async handleContactCreationEvent(payload: {
-    workspaceId: string;
-    connectedAccountHandle: string;
-    contactsToCreate: {
-      displayName: string;
-      handle: string;
-    }[];
-  }) {
-    await this.messageQueueService.add<CreateCompanyAndContactJobData>(
-      CreateCompanyAndContactJob.name,
-      payload,
-    );
-  }
-}

packages/twenty-server/src/modules/messaging/services/save-message-and-emit-contact-creation-event/save-message-and-emit-contact-creation-event.service.ts

@@ -1,5 +1,4 @@
 import { Injectable, Logger } from '@nestjs/common';
-import { EventEmitter2 } from '@nestjs/event-emitter';
 
 import { EntityManager } from 'typeorm';
 
@@ -15,6 +14,13 @@ import { InjectObjectMetadataRepository } from 'src/engine/object-metadata-repos
 import { MessageChannelObjectMetadata } from 'src/modules/messaging/standard-objects/message-channel.object-metadata';
 import { MessageService } from 'src/modules/messaging/services/message/message.service';
 import { MessageParticipantService } from 'src/modules/messaging/services/message-participant/message-participant.service';
+import {
+  CreateCompanyAndContactJobData,
+  CreateCompanyAndContactJob,
+} from 'src/modules/connected-account/auto-companies-and-contacts-creation/jobs/create-company-and-contact.job';
+import { InjectMessageQueue } from 'src/engine/integrations/message-queue/decorators/message-queue.decorator';
+import { MessageQueueService } from 'src/engine/integrations/message-queue/services/message-queue.service';
+import { MessageQueue } from 'src/engine/integrations/message-queue/message-queue.constants';
 
 @Injectable()
 export class SaveMessageAndEmitContactCreationEventService {
@@ -26,9 +32,10 @@ export class SaveMessageAndEmitContactCreationEventService {
     private readonly messageService: MessageService,
     @InjectObjectMetadataRepository(MessageChannelObjectMetadata)
     private readonly messageChannelRepository: MessageChannelRepository,
-    private readonly eventEmitter: EventEmitter2,
     private readonly workspaceDataSourceService: WorkspaceDataSourceService,
     private readonly messageParticipantService: MessageParticipantService,
+    @InjectMessageQueue(MessageQueue.emailQueue)
+    private readonly messageQueueService: MessageQueueService,
   ) {}
 
   public async saveMessagesAndEmitContactCreationEventWithinTransaction(
@@ -75,11 +82,14 @@ export class SaveMessageAndEmitContactCreationEventService {
         (participant) => participant.shouldCreateContact,
       );
 
-      this.eventEmitter.emit(`createContacts`, {
+      await this.messageQueueService.add<CreateCompanyAndContactJobData>(
-        workspaceId,
+        CreateCompanyAndContactJob.name,
-        connectedAccountHandle: connectedAccount.handle,
+        {
-        contactsToCreate,
+          workspaceId,
-      });
+          connectedAccountHandle: connectedAccount.handle,
+          contactsToCreate,
+        },
+      );
     }
   }
 
@@ -180,11 +190,14 @@ export class SaveMessageAndEmitContactCreationEventService {
           (participant) => participant.shouldCreateContact,
         );
 
-        this.eventEmitter.emit(`createContacts`, {
+        await this.messageQueueService.add<CreateCompanyAndContactJobData>(
-          workspaceId,
+          CreateCompanyAndContactJob.name,
-          connectedAccountHandle: connectedAccount.handle,
+          {
-          contactsToCreate,
+            workspaceId,
-        });
+            connectedAccountHandle: connectedAccount.handle,
+            contactsToCreate,
+          },
+        );
       }
     } catch (error) {
       this.logger.error(