4746 create created listener on blocklist (#5031)

Closes #4746 for messaging.

I will create another PR to implement the listener on calendar.

packages/twenty-front/src/modules/settings/accounts/components/SettingsAccountsEmailsBlocklistTableRow.tsx

@@ -19,7 +19,9 @@ export const SettingsAccountsEmailsBlocklistTableRow = ({
     <TableRow key={blocklistItem.id}>
       <TableCell>{blocklistItem.handle}</TableCell>
       <TableCell>
-        {formatToHumanReadableDate(blocklistItem.createdAt)}
+        {blocklistItem.createdAt
+          ? formatToHumanReadableDate(blocklistItem.createdAt)
+          : ''}
       </TableCell>
       <TableCell align="right">
         <IconButton

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

@@ -45,6 +45,7 @@ import { EventObjectMetadata } from 'src/modules/event/standard-objects/event.ob
 import { GmailFetchMessagesFromCacheCronJob } from 'src/modules/messaging/crons/jobs/gmail-fetch-messages-from-cache.cron.job';
 import { GmailPartialSyncCronJob } from 'src/modules/messaging/crons/jobs/gmail-partial-sync.cron.job';
 import { DeleteConnectedAccountAssociatedMessagingDataJob } from 'src/modules/messaging/jobs/delete-connected-account-associated-messaging-data.job';
+import { DeleteMessagesFromHandleJob } from 'src/modules/messaging/jobs/delete-messages-from-handle.job';
 import { GmailFullSyncJob } from 'src/modules/messaging/jobs/gmail-full-sync.job';
 import { GmailPartialSyncJob } from 'src/modules/messaging/jobs/gmail-partial-sync.job';
 import { MessagingCreateCompanyAndContactAfterSyncJob } from 'src/modules/messaging/jobs/messaging-create-company-and-contact-after-sync.job';
@@ -53,6 +54,7 @@ import { GmailFullSyncModule } from 'src/modules/messaging/services/gmail-full-s
 import { GmailPartialSyncModule } from 'src/modules/messaging/services/gmail-partial-sync/gmail-partial-sync.module';
 import { MessageParticipantModule } from 'src/modules/messaging/services/message-participant/message-participant.module';
 import { ThreadCleanerModule } from 'src/modules/messaging/services/thread-cleaner/thread-cleaner.module';
+import { MessageChannelMessageAssociationObjectMetadata } from 'src/modules/messaging/standard-objects/message-channel-message-association.object-metadata';
 import { MessageChannelObjectMetadata } from 'src/modules/messaging/standard-objects/message-channel.object-metadata';
 
 @Module({
@@ -82,6 +84,7 @@ import { MessageChannelObjectMetadata } from 'src/modules/messaging/standard-obj
       ConnectedAccountObjectMetadata,
       MessageChannelObjectMetadata,
       EventObjectMetadata,
+      MessageChannelMessageAssociationObjectMetadata,
     ]),
     GmailFullSyncModule,
     GmailFetchMessageContentFromCacheModule,
@@ -172,6 +175,10 @@ import { MessageChannelObjectMetadata } from 'src/modules/messaging/standard-obj
       provide: GoogleCalendarSyncCronJob.name,
       useClass: GoogleCalendarSyncCronJob,
     },
+    {
+      provide: DeleteMessagesFromHandleJob.name,
+      useClass: DeleteMessagesFromHandleJob,
+    },
   ],
 })
 export class JobsModule {

packages/twenty-server/src/modules/calendar/utils/google-calendar-search-filter.util.ts

@@ -5,5 +5,5 @@ export const googleCalendarSearchFilterExcludeEmails = (
     return undefined;
   }
 
-  return `email=-${emails.join(', -')}`;
+  return `email=-(${emails.join(', -')})`;
 };

packages/twenty-server/src/modules/connected-account/repositories/blocklist.repository.ts

@@ -12,6 +12,29 @@ export class BlocklistRepository {
     private readonly workspaceDataSourceService: WorkspaceDataSourceService,
   ) {}
 
+  public async getById(
+    id: string,
+    workspaceId: string,
+    transactionManager?: EntityManager,
+  ): Promise<ObjectRecord<BlocklistObjectMetadata> | null> {
+    const dataSourceSchema =
+      this.workspaceDataSourceService.getSchemaName(workspaceId);
+
+    const blocklistItems =
+      await this.workspaceDataSourceService.executeRawQuery(
+        `SELECT * FROM ${dataSourceSchema}."blocklist" WHERE "id" = $1`,
+        [id],
+        workspaceId,
+        transactionManager,
+      );
+
+    if (!blocklistItems || blocklistItems.length === 0) {
+      return null;
+    }
+
+    return blocklistItems[0];
+  }
+
   public async getByWorkspaceMemberId(
     workspaceMemberId: string,
     workspaceId: string,

packages/twenty-server/src/modules/messaging/jobs/delete-messages-from-handle.job.ts

@@ -0,0 +1,79 @@
+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 { BlocklistRepository } from 'src/modules/connected-account/repositories/blocklist.repository';
+import { BlocklistObjectMetadata } from 'src/modules/connected-account/standard-objects/blocklist.object-metadata';
+import { MessageChannelMessageAssociationRepository } from 'src/modules/messaging/repositories/message-channel-message-association.repository';
+import { MessageChannelRepository } from 'src/modules/messaging/repositories/message-channel.repository';
+import { ThreadCleanerService } from 'src/modules/messaging/services/thread-cleaner/thread-cleaner.service';
+import { MessageChannelMessageAssociationObjectMetadata } from 'src/modules/messaging/standard-objects/message-channel-message-association.object-metadata';
+import { MessageChannelObjectMetadata } from 'src/modules/messaging/standard-objects/message-channel.object-metadata';
+
+export type DeleteMessagesFromHandleJobData = {
+  workspaceId: string;
+  blocklistItemId: string;
+};
+
+@Injectable()
+export class DeleteMessagesFromHandleJob
+  implements MessageQueueJob<DeleteMessagesFromHandleJobData>
+{
+  private readonly logger = new Logger(DeleteMessagesFromHandleJob.name);
+
+  constructor(
+    @InjectObjectMetadataRepository(MessageChannelObjectMetadata)
+    private readonly messageChannelRepository: MessageChannelRepository,
+    @InjectObjectMetadataRepository(
+      MessageChannelMessageAssociationObjectMetadata,
+    )
+    private readonly messageChannelMessageAssociationRepository: MessageChannelMessageAssociationRepository,
+    @InjectObjectMetadataRepository(BlocklistObjectMetadata)
+    private readonly blocklistRepository: BlocklistRepository,
+    private readonly threadCleanerService: ThreadCleanerService,
+  ) {}
+
+  async handle(data: DeleteMessagesFromHandleJobData): Promise<void> {
+    const { workspaceId, blocklistItemId } = data;
+
+    const blocklistItem = await this.blocklistRepository.getById(
+      blocklistItemId,
+      workspaceId,
+    );
+
+    if (!blocklistItem) {
+      this.logger.log(
+        `Blocklist item with id ${blocklistItemId} not found in workspace ${workspaceId}`,
+      );
+
+      return;
+    }
+
+    const { handle, workspaceMemberId } = blocklistItem;
+
+    this.logger.log(
+      `Deleting messages from ${handle} in workspace ${workspaceId} for workspace member ${workspaceMemberId}`,
+    );
+
+    const messageChannels =
+      await this.messageChannelRepository.getIdsByWorkspaceMemberId(
+        workspaceMemberId,
+        workspaceId,
+      );
+
+    const messageChannelIds = messageChannels.map(({ id }) => id);
+
+    await this.messageChannelMessageAssociationRepository.deleteByMessageParticipantHandleAndMessageChannelIds(
+      handle,
+      messageChannelIds,
+      workspaceId,
+    );
+
+    await this.threadCleanerService.cleanWorkspaceThreads(workspaceId);
+
+    this.logger.log(
+      `Deleted messages from handle ${handle} in workspace ${workspaceId} for workspace member ${workspaceMemberId}`,
+    );
+  }
+}

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

@@ -0,0 +1,32 @@
+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 { MessageQueue } from 'src/engine/integrations/message-queue/message-queue.constants';
+import { MessageQueueService } from 'src/engine/integrations/message-queue/services/message-queue.service';
+import { BlocklistObjectMetadata } from 'src/modules/connected-account/standard-objects/blocklist.object-metadata';
+import {
+  DeleteMessagesFromHandleJobData,
+  DeleteMessagesFromHandleJob,
+} from 'src/modules/messaging/jobs/delete-messages-from-handle.job';
+
+@Injectable()
+export class MessagingBlocklistListener {
+  constructor(
+    @Inject(MessageQueue.messagingQueue)
+    private readonly messageQueueService: MessageQueueService,
+  ) {}
+
+  @OnEvent('blocklist.created')
+  handleCreatedEvent(
+    payload: ObjectRecordCreateEvent<BlocklistObjectMetadata>,
+  ) {
+    this.messageQueueService.add<DeleteMessagesFromHandleJobData>(
+      DeleteMessagesFromHandleJob.name,
+      {
+        workspaceId: payload.workspaceId,
+        blocklistItemId: payload.recordId,
+      },
+    );
+  }
+}

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

@@ -6,6 +6,7 @@ import { MessagingConnectedAccountListener } from 'src/modules/messaging/listene
 import { FeatureFlagEntity } from 'src/engine/core-modules/feature-flag/feature-flag.entity';
 import { ParticipantPersonListener } from 'src/modules/calendar-messaging-participant/listeners/participant-person.listener';
 import { ParticipantWorkspaceMemberListener } from 'src/modules/calendar-messaging-participant/listeners/participant-workspace-member.listener';
+import { MessagingBlocklistListener } from 'src/modules/messaging/listeners/messaging-blocklist.listener';
 
 @Module({
   imports: [TypeOrmModule.forFeature([FeatureFlagEntity], 'core')],
@@ -14,6 +15,7 @@ import { ParticipantWorkspaceMemberListener } from 'src/modules/calendar-messagi
     ParticipantWorkspaceMemberListener,
     MessagingMessageChannelListener,
     MessagingConnectedAccountListener,
+    MessagingBlocklistListener,
   ],
   exports: [],
 })

packages/twenty-server/src/modules/messaging/repositories/message-channel-message-association.repository.ts

@@ -67,6 +67,40 @@ export class MessageChannelMessageAssociationRepository {
     );
   }
 
+  public async deleteByMessageParticipantHandleAndMessageChannelIds(
+    messageParticipantHandle: string,
+    messageChannelIds: string[],
+    workspaceId: string,
+    transactionManager?: EntityManager,
+  ) {
+    const dataSourceSchema =
+      this.workspaceDataSourceService.getSchemaName(workspaceId);
+
+    const messageChannelMessageAssociationIdsToDelete =
+      await this.workspaceDataSourceService.executeRawQuery(
+        `SELECT "messageChannelMessageAssociation".id
+      FROM ${dataSourceSchema}."messageChannelMessageAssociation" "messageChannelMessageAssociation"
+      JOIN ${dataSourceSchema}."message" ON "messageChannelMessageAssociation"."messageId" = ${dataSourceSchema}."message"."id"
+      JOIN ${dataSourceSchema}."messageParticipant" "messageParticipant" ON ${dataSourceSchema}."message"."id" = "messageParticipant"."messageId"
+      WHERE "messageParticipant"."handle" = $1 AND "messageParticipant"."role"= ANY($2) AND "messageChannelMessageAssociation"."messageChannelId" = ANY($3)`,
+        [messageParticipantHandle, ['from', 'to'], messageChannelIds],
+        workspaceId,
+        transactionManager,
+      );
+
+    const messageChannelMessageAssociationIdsToDeleteArray =
+      messageChannelMessageAssociationIdsToDelete.map(
+        (messageChannelMessageAssociation: { id: string }) =>
+          messageChannelMessageAssociation.id,
+      );
+
+    await this.deleteByIds(
+      messageChannelMessageAssociationIdsToDeleteArray,
+      workspaceId,
+      transactionManager,
+    );
+  }
+
   public async getByMessageChannelIds(
     messageChannelIds: string[],
     workspaceId: string,

packages/twenty-server/src/modules/messaging/repositories/message-channel.repository.ts

@@ -137,6 +137,27 @@ export class MessageChannelRepository {
     );
   }
 
+  public async getIdsByWorkspaceMemberId(
+    workspaceMemberId: string,
+    workspaceId: string,
+    transactionManager?: EntityManager,
+  ): Promise<ObjectRecord<MessageChannelObjectMetadata>[]> {
+    const dataSourceSchema =
+      this.workspaceDataSourceService.getSchemaName(workspaceId);
+
+    const messageChannelIds =
+      await this.workspaceDataSourceService.executeRawQuery(
+        `SELECT "messageChannel".id FROM ${dataSourceSchema}."messageChannel" "messageChannel"
+        JOIN ${dataSourceSchema}."connectedAccount" ON "messageChannel"."connectedAccountId" = ${dataSourceSchema}."connectedAccount"."id"
+        WHERE ${dataSourceSchema}."connectedAccount"."accountOwnerId" = $1`,
+        [workspaceMemberId],
+        workspaceId,
+        transactionManager,
+      );
+
+    return messageChannelIds;
+  }
+
   public async updateSyncStatus(
     id: string,
     syncStatus: MessageChannelSyncStatus,