4738 add listeners on person creation and workspacemember creation to update participants (#4854)

Closes #4738

- Added the logic to unmatch a participant when the email of a person or
a workspace member is updated

packages/twenty-server/src/modules/connected-account/jobs/match-participant.job.ts

@@ -0,0 +1,60 @@
+import { Injectable } from '@nestjs/common';
+import { InjectRepository } from '@nestjs/typeorm';
+
+import { Repository } from 'typeorm';
+
+import { MessageQueueJob } from 'src/engine/integrations/message-queue/interfaces/message-queue-job.interface';
+
+import {
+  FeatureFlagEntity,
+  FeatureFlagKeys,
+} from 'src/engine/core-modules/feature-flag/feature-flag.entity';
+import { CalendarEventParticipantService } from 'src/modules/calendar/services/calendar-event-participant/calendar-event-participant.service';
+import { MessageParticipantService } from 'src/modules/messaging/services/message-participant/message-participant.service';
+
+export type MatchParticipantJobData = {
+  workspaceId: string;
+  email: string;
+  personId?: string;
+  workspaceMemberId?: string;
+};
+
+@Injectable()
+export class MatchParticipantJob
+  implements MessageQueueJob<MatchParticipantJobData>
+{
+  constructor(
+    private readonly messageParticipantService: MessageParticipantService,
+    private readonly calendarEventParticipantService: CalendarEventParticipantService,
+    @InjectRepository(FeatureFlagEntity, 'core')
+    private readonly featureFlagRepository: Repository<FeatureFlagEntity>,
+  ) {}
+
+  async handle(data: MatchParticipantJobData): Promise<void> {
+    const { workspaceId, email, personId, workspaceMemberId } = data;
+
+    await this.messageParticipantService.matchMessageParticipants(
+      workspaceId,
+      email,
+      personId,
+      workspaceMemberId,
+    );
+
+    const isCalendarEnabled = await this.featureFlagRepository.findOneBy({
+      workspaceId,
+      key: FeatureFlagKeys.IsCalendarEnabled,
+      value: true,
+    });
+
+    if (!isCalendarEnabled || !isCalendarEnabled.value) {
+      return;
+    }
+
+    await this.calendarEventParticipantService.matchCalendarEventParticipants(
+      workspaceId,
+      email,
+      personId,
+      workspaceMemberId,
+    );
+  }
+}

packages/twenty-server/src/modules/connected-account/jobs/unmatch-participant.job.ts

@@ -0,0 +1,60 @@
+import { Injectable } from '@nestjs/common';
+import { InjectRepository } from '@nestjs/typeorm';
+
+import { Repository } from 'typeorm';
+
+import { MessageQueueJob } from 'src/engine/integrations/message-queue/interfaces/message-queue-job.interface';
+
+import {
+  FeatureFlagEntity,
+  FeatureFlagKeys,
+} from 'src/engine/core-modules/feature-flag/feature-flag.entity';
+import { CalendarEventParticipantService } from 'src/modules/calendar/services/calendar-event-participant/calendar-event-participant.service';
+import { MessageParticipantService } from 'src/modules/messaging/services/message-participant/message-participant.service';
+
+export type UnmatchParticipantJobData = {
+  workspaceId: string;
+  email: string;
+  personId?: string;
+  workspaceMemberId?: string;
+};
+
+@Injectable()
+export class UnmatchParticipantJob
+  implements MessageQueueJob<UnmatchParticipantJobData>
+{
+  constructor(
+    private readonly messageParticipantService: MessageParticipantService,
+    private readonly calendarEventParticipantService: CalendarEventParticipantService,
+    @InjectRepository(FeatureFlagEntity, 'core')
+    private readonly featureFlagRepository: Repository<FeatureFlagEntity>,
+  ) {}
+
+  async handle(data: UnmatchParticipantJobData): Promise<void> {
+    const { workspaceId, email, personId, workspaceMemberId } = data;
+
+    await this.messageParticipantService.unmatchMessageParticipants(
+      workspaceId,
+      email,
+      personId,
+      workspaceMemberId,
+    );
+
+    const isCalendarEnabled = await this.featureFlagRepository.findOneBy({
+      workspaceId,
+      key: FeatureFlagKeys.IsCalendarEnabled,
+      value: true,
+    });
+
+    if (!isCalendarEnabled || !isCalendarEnabled.value) {
+      return;
+    }
+
+    await this.calendarEventParticipantService.unmatchCalendarEventParticipants(
+      workspaceId,
+      email,
+      personId,
+      workspaceMemberId,
+    );
+  }
+}

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

@@ -0,0 +1,73 @@
+import { Inject, Injectable } from '@nestjs/common';
+import { OnEvent } from '@nestjs/event-emitter';
+
+import { ObjectRecordCreateEvent } from 'src/engine/integrations/event-emitter/types/object-record-create.event';
+import { ObjectRecordUpdateEvent } from 'src/engine/integrations/event-emitter/types/object-record-update.event';
+import { objectRecordChangedProperties as objectRecordUpdateEventChangedProperties } 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 {
+  MatchParticipantJob,
+  MatchParticipantJobData,
+} from 'src/modules/connected-account/jobs/match-participant.job';
+import {
+  UnmatchParticipantJobData,
+  UnmatchParticipantJob,
+} from 'src/modules/connected-account/jobs/unmatch-participant.job';
+import { PersonObjectMetadata } from 'src/modules/person/standard-objects/person.object-metadata';
+
+@Injectable()
+export class ParticipantPersonListener {
+  constructor(
+    @Inject(MessageQueue.messagingQueue)
+    private readonly messageQueueService: MessageQueueService,
+  ) {}
+
+  @OnEvent('person.created')
+  async handleCreatedEvent(
+    payload: ObjectRecordCreateEvent<PersonObjectMetadata>,
+  ) {
+    if (payload.details.after.email === null) {
+      return;
+    }
+
+    await this.messageQueueService.add<MatchParticipantJobData>(
+      MatchParticipantJob.name,
+      {
+        workspaceId: payload.workspaceId,
+        email: payload.details.after.email,
+        personId: payload.recordId,
+      },
+    );
+  }
+
+  @OnEvent('person.updated')
+  async handleUpdatedEvent(
+    payload: ObjectRecordUpdateEvent<PersonObjectMetadata>,
+  ) {
+    if (
+      objectRecordUpdateEventChangedProperties(
+        payload.details.before,
+        payload.details.after,
+      ).includes('email')
+    ) {
+      await this.messageQueueService.add<UnmatchParticipantJobData>(
+        UnmatchParticipantJob.name,
+        {
+          workspaceId: payload.workspaceId,
+          email: payload.details.before.email,
+          personId: payload.recordId,
+        },
+      );
+
+      await this.messageQueueService.add<MatchParticipantJobData>(
+        MatchParticipantJob.name,
+        {
+          workspaceId: payload.workspaceId,
+          email: payload.details.after.email,
+          personId: payload.recordId,
+        },
+      );
+    }
+  }
+}

packages/twenty-server/src/modules/connected-account/listeners/participant-workspace-member.listener.ts

@@ -0,0 +1,73 @@
+import { Inject, Injectable } from '@nestjs/common';
+import { OnEvent } from '@nestjs/event-emitter';
+
+import { ObjectRecordCreateEvent } from 'src/engine/integrations/event-emitter/types/object-record-create.event';
+import { ObjectRecordUpdateEvent } from 'src/engine/integrations/event-emitter/types/object-record-update.event';
+import { objectRecordChangedProperties as objectRecordUpdateEventChangedProperties } 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 {
+  MatchParticipantJob,
+  MatchParticipantJobData,
+} from 'src/modules/connected-account/jobs/match-participant.job';
+import {
+  UnmatchParticipantJobData,
+  UnmatchParticipantJob,
+} from 'src/modules/connected-account/jobs/unmatch-participant.job';
+import { WorkspaceMemberObjectMetadata } from 'src/modules/workspace-member/standard-objects/workspace-member.object-metadata';
+
+@Injectable()
+export class ParticipantWorkspaceMemberListener {
+  constructor(
+    @Inject(MessageQueue.messagingQueue)
+    private readonly messageQueueService: MessageQueueService,
+  ) {}
+
+  @OnEvent('workspaceMember.created')
+  async handleCreatedEvent(
+    payload: ObjectRecordCreateEvent<WorkspaceMemberObjectMetadata>,
+  ) {
+    if (payload.details.after.userEmail === null) {
+      return;
+    }
+
+    await this.messageQueueService.add<MatchParticipantJobData>(
+      MatchParticipantJob.name,
+      {
+        workspaceId: payload.workspaceId,
+        email: payload.details.after.userEmail,
+        workspaceMemberId: payload.details.after.id,
+      },
+    );
+  }
+
+  @OnEvent('workspaceMember.updated')
+  async handleUpdatedEvent(
+    payload: ObjectRecordUpdateEvent<WorkspaceMemberObjectMetadata>,
+  ) {
+    if (
+      objectRecordUpdateEventChangedProperties(
+        payload.details.before,
+        payload.details.after,
+      ).includes('userEmail')
+    ) {
+      await this.messageQueueService.add<UnmatchParticipantJobData>(
+        UnmatchParticipantJob.name,
+        {
+          workspaceId: payload.workspaceId,
+          email: payload.details.before.userEmail,
+          personId: payload.recordId,
+        },
+      );
+
+      await this.messageQueueService.add<MatchParticipantJobData>(
+        MatchParticipantJob.name,
+        {
+          workspaceId: payload.workspaceId,
+          email: payload.details.after.userEmail,
+          workspaceMemberId: payload.recordId,
+        },
+      );
+    }
+  }
+}