4736 add listener on calendarchannel isautocontactcreationenabled (#4913)

Closes #4736

packages/twenty-server/src/engine/integrations/message-queue/jobs.module.ts

@@ -15,7 +15,7 @@ import { UserModule } from 'src/engine/core-modules/user/user.module';
 import { EnvironmentModule } from 'src/engine/integrations/environment/environment.module';
 import { MatchParticipantJob } from 'src/modules/connected-account/jobs/match-participant.job';
 import { GmailPartialSyncCronJob } from 'src/modules/messaging/jobs/crons/gmail-partial-sync.cron.job';
-import { CreateCompaniesAndContactsAfterSyncJob } from 'src/modules/messaging/jobs/create-companies-and-contacts-after-sync.job';
+import { MessagingCreateCompanyAndContactAfterSyncJob } from 'src/modules/messaging/jobs/messaging-create-company-and-contact-after-sync.job';
 import { AutoCompaniesAndContactsCreationModule } from 'src/modules/connected-account/auto-companies-and-contacts-creation/auto-companies-and-contacts-creation.module';
 import { DataSeedDemoWorkspaceModule } from 'src/database/commands/data-seed-demo-workspace/data-seed-demo-workspace.module';
 import { DataSeedDemoWorkspaceJob } from 'src/database/commands/data-seed-demo-workspace/jobs/data-seed-demo-workspace.job';
@@ -50,6 +50,7 @@ import { GmailPartialSyncV2Module } from 'src/modules/messaging/services/gmail-p
 import { GoogleCalendarSyncCronJob } from 'src/modules/calendar/jobs/crons/google-calendar-sync.cron.job';
 import { CalendarEventParticipantModule } from 'src/modules/calendar/services/calendar-event-participant/calendar-event-participant.module';
 import { UnmatchParticipantJob } from 'src/modules/connected-account/jobs/unmatch-participant.job';
+import { CalendarCreateCompanyAndContactAfterSyncJob } from 'src/modules/calendar/jobs/calendar-create-company-and-contact-after-sync.job';
 import { DeleteConnectedAccountAssociatedCalendarDataJob } from 'src/modules/calendar/jobs/delete-connected-account-associated-calendar-data.job';
 import { FetchAllMessagesFromCacheCronJob } from 'src/modules/messaging/jobs/crons/fetch-all-messages-from-cache.cron.job';
 
@@ -116,8 +117,12 @@ import { FetchAllMessagesFromCacheCronJob } from 'src/modules/messaging/jobs/cro
       useClass: UnmatchParticipantJob,
     },
     {
-      provide: CreateCompaniesAndContactsAfterSyncJob.name,
+      provide: MessagingCreateCompanyAndContactAfterSyncJob.name,
-      useClass: CreateCompaniesAndContactsAfterSyncJob,
+      useClass: MessagingCreateCompanyAndContactAfterSyncJob,
+    },
+    {
+      provide: CalendarCreateCompanyAndContactAfterSyncJob.name,
+      useClass: CalendarCreateCompanyAndContactAfterSyncJob,
     },
     {
       provide: DataSeedDemoWorkspaceJob.name,
@@ -144,6 +149,7 @@ import { FetchAllMessagesFromCacheCronJob } from 'src/modules/messaging/jobs/cro
       provide: CreateCompanyAndContactJob.name,
       useClass: CreateCompanyAndContactJob,
     },
+
     {
       provide: SaveEventToDbJob.name,
       useClass: SaveEventToDbJob,

packages/twenty-server/src/modules/calendar/calendar.module.ts

@@ -1,8 +1,10 @@
 import { Module } from '@nestjs/common';
 
+import { CalendarChannelListener } from 'src/modules/calendar/listeners/calendar-channel.listener';
+
 @Module({
   imports: [],
-  providers: [],
+  providers: [CalendarChannelListener],
   exports: [],
 })
 export class CalendarModule {}

packages/twenty-server/src/modules/calendar/jobs/calendar-create-company-and-contact-after-sync.job.ts

@@ -0,0 +1,73 @@
+import { Injectable, Logger } from '@nestjs/common';
+
+import { MessageQueueJob } from 'src/engine/integrations/message-queue/interfaces/message-queue-job.interface';
+
+import { InjectObjectMetadataRepository } from 'src/engine/object-metadata-repository/object-metadata-repository.decorator';
+import { CalendarChannelRepository } from 'src/modules/calendar/repositories/calendar-channel.repository';
+import { CalendarEventParticipantRepository } from 'src/modules/calendar/repositories/calendar-event-participant.repository';
+import { CalendarChannelObjectMetadata } from 'src/modules/calendar/standard-objects/calendar-channel.object-metadata';
+import { CalendarEventParticipantObjectMetadata } from 'src/modules/calendar/standard-objects/calendar-event-participant.object-metadata';
+import { CreateCompanyAndContactService } from 'src/modules/connected-account/auto-companies-and-contacts-creation/services/create-company-and-contact.service';
+
+export type CalendarCreateCompanyAndContactAfterSyncJobData = {
+  workspaceId: string;
+  calendarChannelId: string;
+};
+
+@Injectable()
+export class CalendarCreateCompanyAndContactAfterSyncJob
+  implements MessageQueueJob<CalendarCreateCompanyAndContactAfterSyncJobData>
+{
+  private readonly logger = new Logger(
+    CalendarCreateCompanyAndContactAfterSyncJob.name,
+  );
+  constructor(
+    private readonly createCompanyAndContactService: CreateCompanyAndContactService,
+    @InjectObjectMetadataRepository(CalendarChannelObjectMetadata)
+    private readonly calendarChannelService: CalendarChannelRepository,
+    @InjectObjectMetadataRepository(CalendarEventParticipantObjectMetadata)
+    private readonly calendarEventParticipantRepository: CalendarEventParticipantRepository,
+  ) {}
+
+  async handle(
+    data: CalendarCreateCompanyAndContactAfterSyncJobData,
+  ): Promise<void> {
+    this.logger.log(
+      `create contacts and companies after sync for workspace ${data.workspaceId} and calendarChannel ${data.calendarChannelId}`,
+    );
+    const { workspaceId, calendarChannelId } = data;
+
+    const calendarChannels = await this.calendarChannelService.getByIds(
+      [calendarChannelId],
+      workspaceId,
+    );
+
+    if (calendarChannels.length === 0) {
+      throw new Error(
+        `Calendar channel with id ${calendarChannelId} not found in workspace ${workspaceId}`,
+      );
+    }
+
+    const { handle, isContactAutoCreationEnabled } = calendarChannels[0];
+
+    if (!isContactAutoCreationEnabled || !handle) {
+      return;
+    }
+
+    const calendarEventParticipantsWithoutPersonIdAndWorkspaceMemberId =
+      await this.calendarEventParticipantRepository.getByCalendarChannelIdWithoutPersonIdAndWorkspaceMemberId(
+        calendarChannelId,
+        workspaceId,
+      );
+
+    await this.createCompanyAndContactService.createCompaniesAndContactsAndUpdateParticipants(
+      handle,
+      calendarEventParticipantsWithoutPersonIdAndWorkspaceMemberId,
+      workspaceId,
+    );
+
+    this.logger.log(
+      `create contacts and companies after sync for workspace ${data.workspaceId} and calendarChannel ${data.calendarChannelId} done`,
+    );
+  }
+}

packages/twenty-server/src/modules/calendar/listeners/calendar-channel.listener.ts

@@ -0,0 +1,41 @@
+import { Inject, Injectable } from '@nestjs/common';
+import { OnEvent } from '@nestjs/event-emitter';
+
+import { ObjectRecordUpdateEvent } from 'src/engine/integrations/event-emitter/types/object-record-update.event';
+import { objectRecordChangedProperties } from 'src/engine/integrations/event-emitter/utils/object-record-changed-properties.util';
+import { MessageQueue } from 'src/engine/integrations/message-queue/message-queue.constants';
+import { MessageQueueService } from 'src/engine/integrations/message-queue/services/message-queue.service';
+import {
+  CalendarCreateCompanyAndContactAfterSyncJobData,
+  CalendarCreateCompanyAndContactAfterSyncJob,
+} from 'src/modules/calendar/jobs/calendar-create-company-and-contact-after-sync.job';
+import { MessageChannelObjectMetadata } from 'src/modules/messaging/standard-objects/message-channel.object-metadata';
+
+@Injectable()
+export class CalendarChannelListener {
+  constructor(
+    @Inject(MessageQueue.messagingQueue)
+    private readonly messageQueueService: MessageQueueService,
+  ) {}
+
+  @OnEvent('calendarChannel.updated')
+  async handleUpdatedEvent(
+    payload: ObjectRecordUpdateEvent<MessageChannelObjectMetadata>,
+  ) {
+    if (
+      objectRecordChangedProperties(
+        payload.details.before,
+        payload.details.after,
+      ).includes('isContactAutoCreationEnabled') &&
+      payload.details.after.isContactAutoCreationEnabled
+    ) {
+      await this.messageQueueService.add<CalendarCreateCompanyAndContactAfterSyncJobData>(
+        CalendarCreateCompanyAndContactAfterSyncJob.name,
+        {
+          workspaceId: payload.workspaceId,
+          calendarChannelId: payload.recordId,
+        },
+      );
+    }
+  }
+}

packages/twenty-server/src/modules/calendar/repositories/calendar-event-participant.repository.ts

@@ -263,4 +263,33 @@ export class CalendarEventParticipantRepository {
 
     return calendarEventParticipants;
   }
+
+  public async getByCalendarChannelIdWithoutPersonIdAndWorkspaceMemberId(
+    calendarChannelId: string,
+    workspaceId: string,
+    transactionManager?: EntityManager,
+  ): Promise<CalendarEventParticipantWithId[]> {
+    if (!workspaceId) {
+      throw new Error('WorkspaceId is required');
+    }
+
+    const dataSourceSchema =
+      this.workspaceDataSourceService.getSchemaName(workspaceId);
+
+    const calendarEventParticipants: CalendarEventParticipantWithId[] =
+      await this.workspaceDataSourceService.executeRawQuery(
+        `SELECT "calendarEventParticipant".*
+        FROM ${dataSourceSchema}."calendarEventParticipant" AS "calendarEventParticipant"
+        LEFT JOIN ${dataSourceSchema}."calendarEvent" AS "calendarEvent" ON "calendarEventParticipant"."calendarEventId" = "calendarEvent"."id"
+        LEFT JOIN ${dataSourceSchema}."calendarChannelEventAssociation" AS "calendarChannelEventAssociation" ON "calendarEvent"."id" = "calendarChannelEventAssociation"."calendarEventId"
+        WHERE "calendarChannelEventAssociation"."calendarChannelId" = $1
+        AND "calendarEventParticipant"."personId" IS NULL
+        AND "calendarEventParticipant"."workspaceMemberId" IS NULL`,
+        [calendarChannelId],
+        workspaceId,
+        transactionManager,
+      );
+
+    return calendarEventParticipants;
+  }
 }

packages/twenty-server/src/modules/calendar/services/calendar-event-participant/calendar-event-participant.service.ts

@@ -7,10 +7,7 @@ import { PersonRepository } from 'src/modules/person/repositories/person.reposit
 import { PersonObjectMetadata } from 'src/modules/person/standard-objects/person.object-metadata';
 import { WorkspaceDataSourceService } from 'src/engine/workspace-datasource/workspace-datasource.service';
 import { getFlattenedValuesAndValuesStringForBatchRawQuery } from 'src/modules/calendar/utils/getFlattenedValuesAndValuesStringForBatchRawQuery.util';
-import {
+import { CalendarEventParticipant } from 'src/modules/calendar/types/calendar-event';
-  CalendarEventParticipant,
-  CalendarEventParticipantWithId,
-} from 'src/modules/calendar/types/calendar-event';
 import { AddPersonIdAndWorkspaceMemberIdService } from 'src/modules/connected-account/services/add-person-id-and-workspace-member-id/add-person-id-and-workspace-member-id.service';
 import { CalendarEventParticipantRepository } from 'src/modules/calendar/repositories/calendar-event-participant.repository';
 import { CalendarEventParticipantObjectMetadata } from 'src/modules/calendar/standard-objects/calendar-event-participant.object-metadata';
@@ -26,11 +23,15 @@ export class CalendarEventParticipantService {
     private readonly addPersonIdAndWorkspaceMemberIdService: AddPersonIdAndWorkspaceMemberIdService,
   ) {}
 
-  public async updateCalendarEventParticipantsAfterContactCreation(
+  public async updateCalendarEventParticipantsAfterPeopleCreation(
-    participants: CalendarEventParticipantWithId[],
     workspaceId: string,
     transactionManager?: EntityManager,
   ): Promise<void> {
+    const participants =
+      await this.calendarEventParticipantRepository.getWithoutPersonIdAndWorkspaceMemberId(
+        workspaceId,
+      );
+
     if (!participants) return;
 
     const dataSourceSchema =

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

@@ -15,13 +15,9 @@ import { PersonObjectMetadata } from 'src/modules/person/standard-objects/person
 import { WorkspaceMemberObjectMetadata } from 'src/modules/workspace-member/standard-objects/workspace-member.object-metadata';
 import { getUniqueContactsAndHandles } from 'src/modules/connected-account/auto-companies-and-contacts-creation/utils/get-unique-contacts-and-handles.util';
 import { Contacts } from 'src/modules/connected-account/auto-companies-and-contacts-creation/types/contact.type';
-import { MessageParticipantRepository } from 'src/modules/messaging/repositories/message-participant.repository';
 import { WorkspaceDataSourceService } from 'src/engine/workspace-datasource/workspace-datasource.service';
 import { MessageParticipantService } from 'src/modules/messaging/services/message-participant/message-participant.service';
-import { MessageParticipantObjectMetadata } from 'src/modules/messaging/standard-objects/message-participant.object-metadata';
 import { CalendarEventParticipantService } from 'src/modules/calendar/services/calendar-event-participant/calendar-event-participant.service';
-import { CalendarEventParticipantRepository } from 'src/modules/calendar/repositories/calendar-event-participant.repository';
-import { CalendarEventParticipantObjectMetadata } from 'src/modules/calendar/standard-objects/calendar-event-participant.object-metadata';
 import { filterOutContactsFromCompanyOrWorkspace } from 'src/modules/connected-account/auto-companies-and-contacts-creation/utils/filter-out-contacts-from-company-or-workspace.util';
 import {
   FeatureFlagEntity,
@@ -37,12 +33,8 @@ export class CreateCompanyAndContactService {
     private readonly personRepository: PersonRepository,
     @InjectObjectMetadataRepository(WorkspaceMemberObjectMetadata)
     private readonly workspaceMemberRepository: WorkspaceMemberRepository,
-    @InjectObjectMetadataRepository(MessageParticipantObjectMetadata)
-    private readonly messageParticipantRepository: MessageParticipantRepository,
     private readonly workspaceDataSourceService: WorkspaceDataSourceService,
     private readonly messageParticipantService: MessageParticipantService,
-    @InjectObjectMetadataRepository(CalendarEventParticipantObjectMetadata)
-    private readonly calendarEventParticipantRepository: CalendarEventParticipantRepository,
     private readonly calendarEventParticipantService: CalendarEventParticipantService,
     @InjectRepository(FeatureFlagEntity, 'core')
     private readonly featureFlagRepository: Repository<FeatureFlagEntity>,
@@ -157,14 +149,7 @@ export class CreateCompanyAndContactService {
           transactionManager,
         );
 
-        const messageParticipantsWithoutPersonIdAndWorkspaceMemberId =
-          await this.messageParticipantRepository.getWithoutPersonIdAndWorkspaceMemberId(
-            workspaceId,
-            transactionManager,
-          );
-
         await this.messageParticipantService.updateMessageParticipantsAfterPeopleCreation(
-          messageParticipantsWithoutPersonIdAndWorkspaceMemberId,
           workspaceId,
           transactionManager,
         );
@@ -179,14 +164,7 @@ export class CreateCompanyAndContactService {
           return;
         }
 
-        const calendarEventParticipantsWithoutPersonIdAndWorkspaceMemberId =
+        await this.calendarEventParticipantService.updateCalendarEventParticipantsAfterPeopleCreation(
-          await this.calendarEventParticipantRepository.getWithoutPersonIdAndWorkspaceMemberId(
-            workspaceId,
-            transactionManager,
-          );
-
-        await this.calendarEventParticipantService.updateCalendarEventParticipantsAfterContactCreation(
-          calendarEventParticipantsWithoutPersonIdAndWorkspaceMemberId,
           workspaceId,
           transactionManager,
         );

packages/twenty-server/src/modules/messaging/jobs/messaging-create-company-and-contact-after-sync.job.ts

@@ -6,33 +6,31 @@ import { InjectObjectMetadataRepository } from 'src/engine/object-metadata-repos
 import { CreateCompanyAndContactService } from 'src/modules/connected-account/auto-companies-and-contacts-creation/services/create-company-and-contact.service';
 import { MessageChannelRepository } from 'src/modules/messaging/repositories/message-channel.repository';
 import { MessageParticipantRepository } from 'src/modules/messaging/repositories/message-participant.repository';
-import { MessageParticipantService } from 'src/modules/messaging/services/message-participant/message-participant.service';
 import { MessageChannelObjectMetadata } from 'src/modules/messaging/standard-objects/message-channel.object-metadata';
 import { MessageParticipantObjectMetadata } from 'src/modules/messaging/standard-objects/message-participant.object-metadata';
 
-export type CreateCompaniesAndContactsAfterSyncJobData = {
+export type MessagingCreateCompanyAndContactAfterSyncJobData = {
   workspaceId: string;
   messageChannelId: string;
 };
 
 @Injectable()
-export class CreateCompaniesAndContactsAfterSyncJob
+export class MessagingCreateCompanyAndContactAfterSyncJob
-  implements MessageQueueJob<CreateCompaniesAndContactsAfterSyncJobData>
+  implements MessageQueueJob<MessagingCreateCompanyAndContactAfterSyncJobData>
 {
   private readonly logger = new Logger(
-    CreateCompaniesAndContactsAfterSyncJob.name,
+    MessagingCreateCompanyAndContactAfterSyncJob.name,
   );
   constructor(
-    private readonly createCompaniesAndContactsService: CreateCompanyAndContactService,
+    private readonly createCompanyAndContactService: CreateCompanyAndContactService,
     @InjectObjectMetadataRepository(MessageChannelObjectMetadata)
     private readonly messageChannelService: MessageChannelRepository,
-    private readonly messageParticipantService: MessageParticipantService,
     @InjectObjectMetadataRepository(MessageParticipantObjectMetadata)
     private readonly messageParticipantRepository: MessageParticipantRepository,
   ) {}
 
   async handle(
-    data: CreateCompaniesAndContactsAfterSyncJobData,
+    data: MessagingCreateCompanyAndContactAfterSyncJobData,
   ): Promise<void> {
     this.logger.log(
       `create contacts and companies after sync for workspace ${data.workspaceId} and messageChannel ${data.messageChannelId}`,
@@ -50,20 +48,15 @@ export class CreateCompaniesAndContactsAfterSyncJob
       return;
     }
 
-    const messageParticipantsWithoutPersonIdAndWorkspaceMemberId =
+    const contactsToCreate =
       await this.messageParticipantRepository.getByMessageChannelIdWithoutPersonIdAndWorkspaceMemberIdAndMessageOutgoing(
         messageChannelId,
         workspaceId,
       );
 
-    await this.createCompaniesAndContactsService.createCompaniesAndContacts(
+    await this.createCompanyAndContactService.createCompaniesAndContactsAndUpdateParticipants(
       handle,
-      messageParticipantsWithoutPersonIdAndWorkspaceMemberId,
+      contactsToCreate,
-      workspaceId,
-    );
-
-    await this.messageParticipantService.updateMessageParticipantsAfterPeopleCreation(
-      messageParticipantsWithoutPersonIdAndWorkspaceMemberId,
       workspaceId,
     );
 

packages/twenty-server/src/modules/messaging/listeners/messaging-connected-account.listener.ts

@@ -42,7 +42,7 @@ export class MessagingConnectedAccountListener {
       value: true,
     });
 
-    this.messageQueueService.add<DeleteConnectedAccountAssociatedMessagingDataJobData>(
+    await this.messageQueueService.add<DeleteConnectedAccountAssociatedMessagingDataJobData>(
       DeleteConnectedAccountAssociatedMessagingDataJob.name,
       {
         workspaceId: payload.workspaceId,
@@ -51,7 +51,7 @@ export class MessagingConnectedAccountListener {
     );
 
     if (isCalendarEnabled) {
-      this.calendarQueueService.add<DeleteConnectedAccountAssociatedCalendarDataJobData>(
+      await this.calendarQueueService.add<DeleteConnectedAccountAssociatedCalendarDataJobData>(
         DeleteConnectedAccountAssociatedCalendarDataJob.name,
         {
           workspaceId: payload.workspaceId,

packages/twenty-server/src/modules/messaging/listeners/messaging-message-channel.listener.ts

@@ -6,9 +6,9 @@ import { objectRecordChangedProperties } from 'src/engine/integrations/event-emi
 import { MessageQueue } from 'src/engine/integrations/message-queue/message-queue.constants';
 import { MessageQueueService } from 'src/engine/integrations/message-queue/services/message-queue.service';
 import {
-  CreateCompaniesAndContactsAfterSyncJobData,
+  MessagingCreateCompanyAndContactAfterSyncJobData,
-  CreateCompaniesAndContactsAfterSyncJob,
+  MessagingCreateCompanyAndContactAfterSyncJob,
-} from 'src/modules/messaging/jobs/create-companies-and-contacts-after-sync.job';
+} from 'src/modules/messaging/jobs/messaging-create-company-and-contact-after-sync.job';
 import { MessageChannelObjectMetadata } from 'src/modules/messaging/standard-objects/message-channel.object-metadata';
 
 @Injectable()
@@ -19,7 +19,7 @@ export class MessagingMessageChannelListener {
   ) {}
 
   @OnEvent('messageChannel.updated')
-  handleUpdatedEvent(
+  async handleUpdatedEvent(
     payload: ObjectRecordUpdateEvent<MessageChannelObjectMetadata>,
   ) {
     if (
@@ -29,8 +29,8 @@ export class MessagingMessageChannelListener {
       ).includes('isContactAutoCreationEnabled') &&
       payload.details.after.isContactAutoCreationEnabled
     ) {
-      this.messageQueueService.add<CreateCompaniesAndContactsAfterSyncJobData>(
+      await this.messageQueueService.add<MessagingCreateCompanyAndContactAfterSyncJobData>(
-        CreateCompaniesAndContactsAfterSyncJob.name,
+        MessagingCreateCompanyAndContactAfterSyncJob.name,
         {
           workspaceId: payload.workspaceId,
           messageChannelId: payload.recordId,

packages/twenty-server/src/modules/messaging/services/message-participant/message-participant.service.ts

@@ -3,10 +3,7 @@ import { Injectable } from '@nestjs/common';
 import { EntityManager } from 'typeorm';
 
 import { InjectObjectMetadataRepository } from 'src/engine/object-metadata-repository/object-metadata-repository.decorator';
-import {
+import { ParticipantWithMessageId } from 'src/modules/messaging/types/gmail-message';
-  ParticipantWithId,
-  ParticipantWithMessageId,
-} from 'src/modules/messaging/types/gmail-message';
 import { PersonRepository } from 'src/modules/person/repositories/person.repository';
 import { PersonObjectMetadata } from 'src/modules/person/standard-objects/person.object-metadata';
 import { WorkspaceDataSourceService } from 'src/engine/workspace-datasource/workspace-datasource.service';
@@ -27,10 +24,14 @@ export class MessageParticipantService {
   ) {}
 
   public async updateMessageParticipantsAfterPeopleCreation(
-    participants: ParticipantWithId[],
     workspaceId: string,
     transactionManager?: EntityManager,
   ): Promise<void> {
+    const participants =
+      await this.messageParticipantRepository.getWithoutPersonIdAndWorkspaceMemberId(
+        workspaceId,
+      );
+
     if (!participants) return;
 
     const dataSourceSchema =