6657 Refactor and fix blocklist (#6803)

Closes #6657 - Fix listeners - Refactor jobs to take array of events - Fix calendar events and messages deletion --------- Co-authored-by: Charles Bochet <charles@twenty.com>
2024-08-31 16:38:47 +02:00
parent d9650fd5cf
commit cd66ea74a2
37 changed files with 799 additions and 699 deletions
--- a/packages/twenty-server/src/modules/messaging/blocklist-manager/jobs/messaging-blocklist-item-delete-messages.job.ts
+++ b/packages/twenty-server/src/modules/messaging/blocklist-manager/jobs/messaging-blocklist-item-delete-messages.job.ts
@ -1,21 +1,21 @@
 import { Logger, Scope } from '@nestjs/common';

-import { Any } from 'typeorm';
+import { And, Any, ILike, In, Not, Or } from 'typeorm';

+import { ObjectRecordCreateEvent } from 'src/engine/integrations/event-emitter/types/object-record-create.event';
 import { Process } from 'src/engine/integrations/message-queue/decorators/process.decorator';
 import { Processor } from 'src/engine/integrations/message-queue/decorators/processor.decorator';
 import { MessageQueue } from 'src/engine/integrations/message-queue/message-queue.constants';
-import { InjectObjectMetadataRepository } from 'src/engine/object-metadata-repository/object-metadata-repository.decorator';
 import { TwentyORMManager } from 'src/engine/twenty-orm/twenty-orm.manager';
-import { BlocklistRepository } from 'src/modules/blocklist/repositories/blocklist.repository';
+import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/workspace-event.type';
 import { BlocklistWorkspaceEntity } from 'src/modules/blocklist/standard-objects/blocklist.workspace-entity';
 import { MessageChannelMessageAssociationWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message-channel-message-association.workspace-entity';
+import { MessageChannelWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message-channel.workspace-entity';
 import { MessagingMessageCleanerService } from 'src/modules/messaging/message-cleaner/services/messaging-message-cleaner.service';

-export type BlocklistItemDeleteMessagesJobData = {
-  workspaceId: string;
-  blocklistItemId: string;
-};
+export type BlocklistItemDeleteMessagesJobData = WorkspaceEventBatch<
+  ObjectRecordCreateEvent<BlocklistWorkspaceEntity>
+>;

@Processor({
  queueName: MessageQueue.messagingQueue,
@ -25,66 +25,135 @@ export class BlocklistItemDeleteMessagesJob {
  private readonly logger = new Logger(BlocklistItemDeleteMessagesJob.name);

  constructor(
-    @InjectObjectMetadataRepository(BlocklistWorkspaceEntity)
-    private readonly blocklistRepository: BlocklistRepository,
    private readonly threadCleanerService: MessagingMessageCleanerService,
    private readonly twentyORMManager: TwentyORMManager,
  ) {}

  @Process(BlocklistItemDeleteMessagesJob.name)
  async handle(data: BlocklistItemDeleteMessagesJobData): Promise<void> {
-    const { workspaceId, blocklistItemId } = data;
+    const workspaceId = data.workspaceId;

-    const blocklistItem = await this.blocklistRepository.getById(
-      blocklistItemId,
-      workspaceId,
+    const blocklistItemIds = data.events.map(
+      (eventPayload) => eventPayload.recordId,
    );

-    if (!blocklistItem) {
-      this.logger.log(
-        `Blocklist item with id ${blocklistItemId} not found in workspace ${workspaceId}`,
+    const blocklistRepository =
+      await this.twentyORMManager.getRepository<BlocklistWorkspaceEntity>(
+        'blocklist',
      );

-      return;
-    }
+    const blocklist = await blocklistRepository.find({
+      where: {
+        id: Any(blocklistItemIds),
+      },
+    });

-    const { handle, workspaceMemberId } = blocklistItem;
+    const handlesToDeleteByWorkspaceMemberIdMap = blocklist.reduce(
+      (acc, blocklistItem) => {
+        const { handle, workspaceMemberId } = blocklistItem;

-    this.logger.log(
-      `Deleting messages from ${handle} in workspace ${workspaceId} for workspace member ${workspaceMemberId}`,
+        if (!acc.has(workspaceMemberId)) {
+          acc.set(workspaceMemberId, []);
+        }
+
+        acc.get(workspaceMemberId)?.push(handle);
+
+        return acc;
+      },
+      new Map<string, string[]>(),
    );

-    if (!workspaceMemberId) {
-      throw new Error(
-        `Workspace member ID is not defined for blocklist item ${blocklistItemId} in workspace ${workspaceId}`,
+    const messageChannelRepository =
+      await this.twentyORMManager.getRepository<MessageChannelWorkspaceEntity>(
+        'messageChannel',
      );
-    }

    const messageChannelMessageAssociationRepository =
      await this.twentyORMManager.getRepository<MessageChannelMessageAssociationWorkspaceEntity>(
        'messageChannelMessageAssociation',
      );

-    const rolesToDelete: ('from' | 'to')[] = ['from', 'to'];
+    for (const workspaceMemberId of handlesToDeleteByWorkspaceMemberIdMap.keys()) {
+      const handles =
+        handlesToDeleteByWorkspaceMemberIdMap.get(workspaceMemberId);

-    await messageChannelMessageAssociationRepository.delete({
-      messageChannel: {
-        connectedAccount: {
-          accountOwnerId: workspaceMemberId,
+      if (!handles) {
+        continue;
+      }
+
+      this.logger.log(
+        `Deleting messages from ${handles.join(
+          ', ',
+        )} in workspace ${workspaceId} for workspace member ${workspaceMemberId}`,
+      );
+
+      const rolesToDelete: ('from' | 'to')[] = ['from', 'to'];
+
+      const messageChannels = await messageChannelRepository.find({
+        select: {
+          id: true,
+          handle: true,
+          connectedAccount: {
+            handleAliases: true,
+          },
        },
-      },
-      message: {
-        messageParticipants: {
-          handle,
-          role: Any(rolesToDelete),
+        where: {
+          connectedAccount: {
+            accountOwnerId: workspaceMemberId,
+          },
        },
-      },
-    });
+        relations: ['connectedAccount'],
+      });
+
+      for (const messageChannel of messageChannels) {
+        const messageChannelHandles = [messageChannel.handle];
+
+        if (messageChannel.connectedAccount.handleAliases) {
+          messageChannelHandles.push(
+            ...messageChannel.connectedAccount.handleAliases.split(','),
+          );
+        }
+
+        const handleConditions = handles.map((handle) => {
+          const isHandleDomain = handle.startsWith('@');
+
+          return isHandleDomain
+            ? {
+                handle: And(
+                  Or(ILike(`%${handle}`), ILike(`%.${handle.slice(1)}`)),
+                  Not(In(messageChannelHandles)),
+                ),
+                role: In(rolesToDelete),
+              }
+            : { handle, role: In(rolesToDelete) };
+        });
+
+        const messageChannelMessageAssociationsToDelete =
+          await messageChannelMessageAssociationRepository.find({
+            where: {
+              messageChannelId: messageChannel.id,
+              message: {
+                messageParticipants: handleConditions,
+              },
+            },
+          });
+
+        if (messageChannelMessageAssociationsToDelete.length === 0) {
+          continue;
+        }
+
+        await messageChannelMessageAssociationRepository.delete(
+          messageChannelMessageAssociationsToDelete.map(({ id }) => id),
+        );
+      }
+
+      this.logger.log(
+        `Deleted messages from handle ${handles.join(
+          ', ',
+        )} in workspace ${workspaceId} for workspace member ${workspaceMemberId}`,
+      );
+    }

    await this.threadCleanerService.cleanWorkspaceThreads(workspaceId);
-
-    this.logger.log(
-      `Deleted messages from handle ${handle} in workspace ${workspaceId} for workspace member ${workspaceMemberId}`,
-    );
  }
 }
--- a/packages/twenty-server/src/modules/messaging/blocklist-manager/jobs/messaging-blocklist-reimport-messages.job.ts
+++ b/packages/twenty-server/src/modules/messaging/blocklist-manager/jobs/messaging-blocklist-reimport-messages.job.ts
@ -0,0 +1,63 @@
+import { Scope } from '@nestjs/common';
+
+import { Not } from 'typeorm';
+
+import { ObjectRecordDeleteEvent } from 'src/engine/integrations/event-emitter/types/object-record-delete.event';
+import { Process } from 'src/engine/integrations/message-queue/decorators/process.decorator';
+import { Processor } from 'src/engine/integrations/message-queue/decorators/processor.decorator';
+import { MessageQueue } from 'src/engine/integrations/message-queue/message-queue.constants';
+import { TwentyORMManager } from 'src/engine/twenty-orm/twenty-orm.manager';
+import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/workspace-event.type';
+import { BlocklistWorkspaceEntity } from 'src/modules/blocklist/standard-objects/blocklist.workspace-entity';
+import { MessageChannelSyncStatusService } from 'src/modules/messaging/common/services/message-channel-sync-status.service';
+import {
+  MessageChannelSyncStage,
+  MessageChannelWorkspaceEntity,
+} from 'src/modules/messaging/common/standard-objects/message-channel.workspace-entity';
+
+export type BlocklistReimportMessagesJobData = WorkspaceEventBatch<
+  ObjectRecordDeleteEvent<BlocklistWorkspaceEntity>
+>;
+
+@Processor({
+  queueName: MessageQueue.messagingQueue,
+  scope: Scope.REQUEST,
+})
+export class BlocklistReimportMessagesJob {
+  constructor(
+    private readonly twentyORMManager: TwentyORMManager,
+    private readonly messagingChannelSyncStatusService: MessageChannelSyncStatusService,
+  ) {}
+
+  @Process(BlocklistReimportMessagesJob.name)
+  async handle(data: BlocklistReimportMessagesJobData): Promise<void> {
+    const workspaceId = data.workspaceId;
+
+    const messageChannelRepository =
+      await this.twentyORMManager.getRepository<MessageChannelWorkspaceEntity>(
+        'messageChannel',
+      );
+
+    for (const eventPayload of data.events) {
+      const workspaceMemberId =
+        eventPayload.properties.before.workspaceMemberId;
+
+      const messageChannels = await messageChannelRepository.find({
+        select: ['id'],
+        where: {
+          connectedAccount: {
+            accountOwnerId: workspaceMemberId,
+          },
+          syncStage: Not(
+            MessageChannelSyncStage.FULL_MESSAGE_LIST_FETCH_PENDING,
+          ),
+        },
+      });
+
+      await this.messagingChannelSyncStatusService.resetAndScheduleFullMessageListFetch(
+        messageChannels.map((messageChannel) => messageChannel.id),
+        workspaceId,
+      );
+    }
+  }
+}
--- a/packages/twenty-server/src/modules/messaging/blocklist-manager/listeners/messaging-blocklist.listener.ts
+++ b/packages/twenty-server/src/modules/messaging/blocklist-manager/listeners/messaging-blocklist.listener.ts
@ -1,4 +1,4 @@
-import { Injectable } from '@nestjs/common';
+import { Injectable, Scope } from '@nestjs/common';
 import { OnEvent } from '@nestjs/event-emitter';

 import { ObjectRecordCreateEvent } from 'src/engine/integrations/event-emitter/types/object-record-create.event';
@ -7,28 +7,22 @@ import { ObjectRecordUpdateEvent } from 'src/engine/integrations/event-emitter/t
 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';
-import { InjectObjectMetadataRepository } from 'src/engine/object-metadata-repository/object-metadata-repository.decorator';
-import { TwentyORMManager } from 'src/engine/twenty-orm/twenty-orm.manager';
 import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/workspace-event.type';
 import { BlocklistWorkspaceEntity } from 'src/modules/blocklist/standard-objects/blocklist.workspace-entity';
-import { ConnectedAccountRepository } from 'src/modules/connected-account/repositories/connected-account.repository';
-import { ConnectedAccountWorkspaceEntity } from 'src/modules/connected-account/standard-objects/connected-account.workspace-entity';
 import {
  BlocklistItemDeleteMessagesJob,
  BlocklistItemDeleteMessagesJobData,
 } from 'src/modules/messaging/blocklist-manager/jobs/messaging-blocklist-item-delete-messages.job';
-import { MessageChannelSyncStatusService } from 'src/modules/messaging/common/services/message-channel-sync-status.service';
-import { MessageChannelWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message-channel.workspace-entity';
+import {
+  BlocklistReimportMessagesJob,
+  BlocklistReimportMessagesJobData,
+} from 'src/modules/messaging/blocklist-manager/jobs/messaging-blocklist-reimport-messages.job';

-@Injectable()
+@Injectable({ scope: Scope.REQUEST })
 export class MessagingBlocklistListener {
  constructor(
    @InjectMessageQueue(MessageQueue.messagingQueue)
    private readonly messageQueueService: MessageQueueService,
-    @InjectObjectMetadataRepository(ConnectedAccountWorkspaceEntity)
-    private readonly connectedAccountRepository: ConnectedAccountRepository,
-    private readonly messagingChannelSyncStatusService: MessageChannelSyncStatusService,
-    private readonly twentyORMManager: TwentyORMManager,
  ) {}

  @OnEvent('blocklist.created')
@ -37,17 +31,9 @@ export class MessagingBlocklistListener {
      ObjectRecordCreateEvent<BlocklistWorkspaceEntity>
    >,
  ) {
-    await Promise.all(
-      payload.events.map((eventPayload) =>
-        // TODO: modify to pass an array of blocklist items
-        this.messageQueueService.add<BlocklistItemDeleteMessagesJobData>(
-          BlocklistItemDeleteMessagesJob.name,
-          {
-            workspaceId: payload.workspaceId,
-            blocklistItemId: eventPayload.recordId,
-          },
-        ),
-      ),
+    await this.messageQueueService.add<BlocklistItemDeleteMessagesJobData>(
+      BlocklistItemDeleteMessagesJob.name,
+      payload,
    );
  }

@ -57,38 +43,10 @@ export class MessagingBlocklistListener {
      ObjectRecordDeleteEvent<BlocklistWorkspaceEntity>
    >,
  ) {
-    const workspaceId = payload.workspaceId;
-
-    for (const eventPayload of payload.events) {
-      const workspaceMemberId =
-        eventPayload.properties.before.workspaceMember.id;
-
-      const connectedAccount =
-        await this.connectedAccountRepository.getAllByWorkspaceMemberId(
-          workspaceMemberId,
-          workspaceId,
-        );
-
-      if (!connectedAccount || connectedAccount.length === 0) {
-        return;
-      }
-
-      const messageChannelRepository =
-        await this.twentyORMManager.getRepository<MessageChannelWorkspaceEntity>(
-          'messageChannel',
-        );
-
-      const messageChannel = await messageChannelRepository.findOneOrFail({
-        where: {
-          connectedAccountId: connectedAccount[0].id,
-        },
-      });
-
-      await this.messagingChannelSyncStatusService.resetAndScheduleFullMessageListFetch(
-        messageChannel.id,
-        workspaceId,
-      );
-    }
+    await this.messageQueueService.add<BlocklistReimportMessagesJobData>(
+      BlocklistReimportMessagesJob.name,
+      payload,
+    );
  }

  @OnEvent('blocklist.updated')
@ -97,45 +55,14 @@ export class MessagingBlocklistListener {
      ObjectRecordUpdateEvent<BlocklistWorkspaceEntity>
    >,
  ) {
-    const workspaceId = payload.workspaceId;
+    await this.messageQueueService.add<BlocklistItemDeleteMessagesJobData>(
+      BlocklistItemDeleteMessagesJob.name,
+      payload,
+    );

-    for (const eventPayload of payload.events) {
-      const workspaceMemberId =
-        eventPayload.properties.before.workspaceMember.id;
-
-      await this.messageQueueService.add<BlocklistItemDeleteMessagesJobData>(
-        BlocklistItemDeleteMessagesJob.name,
-        {
-          workspaceId,
-          blocklistItemId: eventPayload.recordId,
-        },
-      );
-
-      const connectedAccount =
-        await this.connectedAccountRepository.getAllByWorkspaceMemberId(
-          workspaceMemberId,
-          workspaceId,
-        );
-
-      if (!connectedAccount || connectedAccount.length === 0) {
-        continue;
-      }
-
-      const messageChannelRepository =
-        await this.twentyORMManager.getRepository<MessageChannelWorkspaceEntity>(
-          'messageChannel',
-        );
-
-      const messageChannel = await messageChannelRepository.findOneOrFail({
-        where: {
-          connectedAccountId: connectedAccount[0].id,
-        },
-      });
-
-      await this.messagingChannelSyncStatusService.resetAndScheduleFullMessageListFetch(
-        messageChannel.id,
-        workspaceId,
-      );
-    }
+    await this.messageQueueService.add<BlocklistReimportMessagesJobData>(
+      BlocklistReimportMessagesJob.name,
+      payload,
+    );
  }
 }
--- a/packages/twenty-server/src/modules/messaging/blocklist-manager/messaging-blocklist-manager.module.ts
+++ b/packages/twenty-server/src/modules/messaging/blocklist-manager/messaging-blocklist-manager.module.ts
@ -1,6 +1,7 @@
 import { Module } from '@nestjs/common';

 import { BlocklistItemDeleteMessagesJob } from 'src/modules/messaging/blocklist-manager/jobs/messaging-blocklist-item-delete-messages.job';
+import { BlocklistReimportMessagesJob } from 'src/modules/messaging/blocklist-manager/jobs/messaging-blocklist-reimport-messages.job';
 import { MessagingBlocklistListener } from 'src/modules/messaging/blocklist-manager/listeners/messaging-blocklist.listener';
 import { MessagingCommonModule } from 'src/modules/messaging/common/messaging-common.module';
 import { MessagingMessageCleanerModule } from 'src/modules/messaging/message-cleaner/messaging-message-cleaner.module';
@ -9,10 +10,8 @@ import { MessagingMessageCleanerModule } from 'src/modules/messaging/message-cle
  imports: [MessagingCommonModule, MessagingMessageCleanerModule],
  providers: [
    MessagingBlocklistListener,
-    {
-      provide: BlocklistItemDeleteMessagesJob.name,
-      useClass: BlocklistItemDeleteMessagesJob,
-    },
+    BlocklistItemDeleteMessagesJob,
+    BlocklistReimportMessagesJob,
  ],
  exports: [],
 })
--- a/packages/twenty-server/src/modules/messaging/common/services/message-channel-sync-status.service.ts
+++ b/packages/twenty-server/src/modules/messaging/common/services/message-channel-sync-status.service.ts
@ -1,5 +1,7 @@
 import { Injectable } from '@nestjs/common';

+import { Any } from 'typeorm';
+
 import { CacheStorageService } from 'src/engine/integrations/cache-storage/cache-storage.service';
 import { InjectCacheStorage } from 'src/engine/integrations/cache-storage/decorators/cache-storage.decorator';
 import { CacheStorageNamespace } from 'src/engine/integrations/cache-storage/types/cache-storage-namespace.enum';
@ -12,10 +14,6 @@ import {
  MessageChannelSyncStatus,
  MessageChannelWorkspaceEntity,
 } from 'src/modules/messaging/common/standard-objects/message-channel.workspace-entity';
-import {
-  MessageImportException,
-  MessageImportExceptionCode,
-} from 'src/modules/messaging/message-import-manager/exceptions/message-import.exception';

@Injectable()
 export class MessageChannelSyncStatusService {
@ -26,216 +24,235 @@ export class MessageChannelSyncStatusService {
    private readonly accountsToReconnectService: AccountsToReconnectService,
  ) {}

-  public async scheduleFullMessageListFetch(messageChannelId: string) {
+  public async scheduleFullMessageListFetch(messageChannelIds: string[]) {
+    if (!messageChannelIds.length) {
+      return;
+    }
+
    const messageChannelRepository =
      await this.twentyORMManager.getRepository<MessageChannelWorkspaceEntity>(
        'messageChannel',
      );

-    await messageChannelRepository.update(
-      { id: messageChannelId },
-      {
-        syncStage: MessageChannelSyncStage.FULL_MESSAGE_LIST_FETCH_PENDING,
-      },
-    );
+    await messageChannelRepository.update(messageChannelIds, {
+      syncStage: MessageChannelSyncStage.FULL_MESSAGE_LIST_FETCH_PENDING,
+    });
  }

-  public async schedulePartialMessageListFetch(messageChannelId: string) {
+  public async schedulePartialMessageListFetch(messageChannelIds: string[]) {
+    if (!messageChannelIds.length) {
+      return;
+    }
+
    const messageChannelRepository =
      await this.twentyORMManager.getRepository<MessageChannelWorkspaceEntity>(
        'messageChannel',
      );

-    await messageChannelRepository.update(
-      { id: messageChannelId },
-      {
-        syncStage: MessageChannelSyncStage.PARTIAL_MESSAGE_LIST_FETCH_PENDING,
-      },
-    );
+    await messageChannelRepository.update(messageChannelIds, {
+      syncStage: MessageChannelSyncStage.PARTIAL_MESSAGE_LIST_FETCH_PENDING,
+    });
  }

-  public async scheduleMessagesImport(messageChannelId: string) {
+  public async scheduleMessagesImport(messageChannelIds: string[]) {
+    if (!messageChannelIds.length) {
+      return;
+    }
+
    const messageChannelRepository =
      await this.twentyORMManager.getRepository<MessageChannelWorkspaceEntity>(
        'messageChannel',
      );

-    await messageChannelRepository.update(
-      { id: messageChannelId },
-      {
-        syncStage: MessageChannelSyncStage.MESSAGES_IMPORT_PENDING,
-      },
-    );
+    await messageChannelRepository.update(messageChannelIds, {
+      syncStage: MessageChannelSyncStage.MESSAGES_IMPORT_PENDING,
+    });
  }

  public async resetAndScheduleFullMessageListFetch(
-    messageChannelId: string,
+    messageChannelIds: string[],
    workspaceId: string,
  ) {
-    await this.cacheStorage.del(
-      `messages-to-import:${workspaceId}:gmail:${messageChannelId}`,
-    );
+    if (!messageChannelIds.length) {
+      return;
+    }
+
+    for (const messageChannelId of messageChannelIds) {
+      await this.cacheStorage.del(
+        `messages-to-import:${workspaceId}:gmail:${messageChannelId}`,
+      );
+    }

    const messageChannelRepository =
      await this.twentyORMManager.getRepository<MessageChannelWorkspaceEntity>(
        'messageChannel',
      );

-    await messageChannelRepository.update(
-      { id: messageChannelId },
-      {
-        syncCursor: '',
-        syncStageStartedAt: null,
-        throttleFailureCount: 0,
-      },
-    );
+    await messageChannelRepository.update(messageChannelIds, {
+      syncCursor: '',
+      syncStageStartedAt: null,
+      throttleFailureCount: 0,
+    });

-    await this.scheduleFullMessageListFetch(messageChannelId);
+    await this.scheduleFullMessageListFetch(messageChannelIds);
  }

-  public async resetSyncStageStartedAt(messageChannelId: string) {
+  public async resetSyncStageStartedAt(messageChannelIds: string[]) {
+    if (!messageChannelIds.length) {
+      return;
+    }
+
    const messageChannelRepository =
      await this.twentyORMManager.getRepository<MessageChannelWorkspaceEntity>(
        'messageChannel',
      );

-    await messageChannelRepository.update(
-      { id: messageChannelId },
-      {
-        syncStageStartedAt: null,
-      },
-    );
+    await messageChannelRepository.update(messageChannelIds, {
+      syncStageStartedAt: null,
+    });
  }

-  public async markAsMessagesListFetchOngoing(messageChannelId: string) {
+  public async markAsMessagesListFetchOngoing(messageChannelIds: string[]) {
+    if (!messageChannelIds.length) {
+      return;
+    }
+
    const messageChannelRepository =
      await this.twentyORMManager.getRepository<MessageChannelWorkspaceEntity>(
        'messageChannel',
      );

-    await messageChannelRepository.update(
-      { id: messageChannelId },
-      {
-        syncStage: MessageChannelSyncStage.MESSAGE_LIST_FETCH_ONGOING,
-        syncStatus: MessageChannelSyncStatus.ONGOING,
-      },
-    );
+    await messageChannelRepository.update(messageChannelIds, {
+      syncStage: MessageChannelSyncStage.MESSAGE_LIST_FETCH_ONGOING,
+      syncStatus: MessageChannelSyncStatus.ONGOING,
+    });
  }

  public async markAsCompletedAndSchedulePartialMessageListFetch(
-    messageChannelId: string,
+    messageChannelIds: string[],
  ) {
+    if (!messageChannelIds.length) {
+      return;
+    }
+
    const messageChannelRepository =
      await this.twentyORMManager.getRepository<MessageChannelWorkspaceEntity>(
        'messageChannel',
      );

-    await messageChannelRepository.update(
-      { id: messageChannelId },
-      {
-        syncStatus: MessageChannelSyncStatus.ACTIVE,
-      },
-    );
+    await messageChannelRepository.update(messageChannelIds, {
+      syncStatus: MessageChannelSyncStatus.ACTIVE,
+    });

-    await this.schedulePartialMessageListFetch(messageChannelId);
+    await this.schedulePartialMessageListFetch(messageChannelIds);
  }

-  public async markAsMessagesImportOngoing(messageChannelId: string) {
+  public async markAsMessagesImportOngoing(messageChannelIds: string[]) {
+    if (!messageChannelIds.length) {
+      return;
+    }
+
    const messageChannelRepository =
      await this.twentyORMManager.getRepository<MessageChannelWorkspaceEntity>(
        'messageChannel',
      );

-    await messageChannelRepository.update(
-      { id: messageChannelId },
-      {
-        syncStage: MessageChannelSyncStage.MESSAGES_IMPORT_ONGOING,
-      },
-    );
+    await messageChannelRepository.update(messageChannelIds, {
+      syncStage: MessageChannelSyncStage.MESSAGES_IMPORT_ONGOING,
+    });
  }

  public async markAsFailedUnknownAndFlushMessagesToImport(
-    messageChannelId: string,
+    messageChannelIds: string[],
    workspaceId: string,
  ) {
-    await this.cacheStorage.del(
-      `messages-to-import:${workspaceId}:gmail:${messageChannelId}`,
-    );
+    if (!messageChannelIds.length) {
+      return;
+    }
+
+    for (const messageChannelId of messageChannelIds) {
+      await this.cacheStorage.del(
+        `messages-to-import:${workspaceId}:gmail:${messageChannelId}`,
+      );
+    }

    const messageChannelRepository =
      await this.twentyORMManager.getRepository<MessageChannelWorkspaceEntity>(
        'messageChannel',
      );

-    await messageChannelRepository.update(
-      { id: messageChannelId },
-      {
-        syncStage: MessageChannelSyncStage.FAILED,
-        syncStatus: MessageChannelSyncStatus.FAILED_UNKNOWN,
-      },
-    );
+    await messageChannelRepository.update(messageChannelIds, {
+      syncStage: MessageChannelSyncStage.FAILED,
+      syncStatus: MessageChannelSyncStatus.FAILED_UNKNOWN,
+    });
  }

  public async markAsFailedInsufficientPermissionsAndFlushMessagesToImport(
-    messageChannelId: string,
+    messageChannelIds: string[],
    workspaceId: string,
  ) {
-    await this.cacheStorage.del(
-      `messages-to-import:${workspaceId}:gmail:${messageChannelId}`,
-    );
+    if (!messageChannelIds.length) {
+      return;
+    }
+
+    for (const messageChannelId of messageChannelIds) {
+      await this.cacheStorage.del(
+        `messages-to-import:${workspaceId}:gmail:${messageChannelId}`,
+      );
+    }

    const messageChannelRepository =
      await this.twentyORMManager.getRepository<MessageChannelWorkspaceEntity>(
        'messageChannel',
      );

-    await messageChannelRepository.update(
-      { id: messageChannelId },
-      {
-        syncStage: MessageChannelSyncStage.FAILED,
-        syncStatus: MessageChannelSyncStatus.FAILED_INSUFFICIENT_PERMISSIONS,
-      },
-    );
+    await messageChannelRepository.update(messageChannelIds, {
+      syncStage: MessageChannelSyncStage.FAILED,
+      syncStatus: MessageChannelSyncStatus.FAILED_INSUFFICIENT_PERMISSIONS,
+    });

    const connectedAccountRepository =
      await this.twentyORMManager.getRepository<ConnectedAccountWorkspaceEntity>(
        'connectedAccount',
      );

-    const messageChannel = await messageChannelRepository.findOne({
-      where: { id: messageChannelId },
+    const messageChannels = await messageChannelRepository.find({
+      select: ['id', 'connectedAccountId'],
+      where: { id: Any(messageChannelIds) },
    });

-    if (!messageChannel) {
-      throw new MessageImportException(
-        `Message channel ${messageChannelId} not found in workspace ${workspaceId}`,
-        MessageImportExceptionCode.MESSAGE_CHANNEL_NOT_FOUND,
-      );
-    }
-
-    const connectedAccountId = messageChannel.connectedAccountId;
+    const connectedAccountIds = messageChannels.map(
+      (messageChannel) => messageChannel.connectedAccountId,
+    );

    await connectedAccountRepository.update(
-      { id: connectedAccountId },
+      { id: Any(connectedAccountIds) },
      {
        authFailedAt: new Date(),
      },
    );

-    await this.addToAccountsToReconnect(messageChannelId, workspaceId);
+    await this.addToAccountsToReconnect(
+      messageChannels.map((messageChannel) => messageChannel.id),
+      workspaceId,
+    );
  }

  private async addToAccountsToReconnect(
-    messageChannelId: string,
+    messageChannelIds: string[],
    workspaceId: string,
  ) {
+    if (!messageChannelIds.length) {
+      return;
+    }
+
    const messageChannelRepository =
      await this.twentyORMManager.getRepository<MessageChannelWorkspaceEntity>(
        'messageChannel',
      );

-    const messageChannel = await messageChannelRepository.findOne({
-      where: { id: messageChannelId },
+    const messageChannels = await messageChannelRepository.find({
+      where: { id: Any(messageChannelIds) },
      relations: {
        connectedAccount: {
          accountOwner: true,
@ -243,18 +260,16 @@ export class MessageChannelSyncStatusService {
      },
    });

-    if (!messageChannel) {
-      return;
+    for (const messageChannel of messageChannels) {
+      const userId = messageChannel.connectedAccount.accountOwner.userId;
+      const connectedAccountId = messageChannel.connectedAccount.id;
+
+      await this.accountsToReconnectService.addAccountToReconnectByKey(
+        AccountsToReconnectKeys.ACCOUNTS_TO_RECONNECT_INSUFFICIENT_PERMISSIONS,
+        userId,
+        workspaceId,
+        connectedAccountId,
+      );
    }
-
-    const userId = messageChannel.connectedAccount.accountOwner.userId;
-    const connectedAccountId = messageChannel.connectedAccount.id;
-
-    await this.accountsToReconnectService.addAccountToReconnectByKey(
-      AccountsToReconnectKeys.ACCOUNTS_TO_RECONNECT_INSUFFICIENT_PERMISSIONS,
-      userId,
-      workspaceId,
-      connectedAccountId,
-    );
  }
 }
--- a/packages/twenty-server/src/modules/messaging/message-cleaner/services/messaging-message-cleaner.service.ts
+++ b/packages/twenty-server/src/modules/messaging/message-cleaner/services/messaging-message-cleaner.service.ts
@ -1,6 +1,6 @@
 import { Injectable } from '@nestjs/common';

-import { EntityManager } from 'typeorm';
+import { EntityManager, IsNull } from 'typeorm';

 import { TwentyORMManager } from 'src/engine/twenty-orm/twenty-orm.manager';
 import { MessageThreadWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message-thread.workspace-entity';
@ -22,67 +22,77 @@ export class MessagingMessageCleanerService {
        'message',
      );

-    await deleteUsingPagination(
-      workspaceId,
-      500,
-      async (
-        limit: number,
-        offset: number,
-        workspaceId: string,
-        transactionManager?: EntityManager,
-      ) => {
-        const nonAssociatedMessages = await messageRepository.find(
-          {
-            where: {
-              messageChannelMessageAssociations: [],
+    const workspaceDataSource = await this.twentyORMManager.getDatasource();
+
+    await workspaceDataSource.transaction(async (transactionManager) => {
+      await deleteUsingPagination(
+        workspaceId,
+        500,
+        async (
+          limit: number,
+          offset: number,
+          workspaceId: string,
+          transactionManager: EntityManager,
+        ) => {
+          const nonAssociatedMessages = await messageRepository.find(
+            {
+              where: {
+                messageChannelMessageAssociations: {
+                  id: IsNull(),
+                },
+              },
+              take: limit,
+              skip: offset,
+              relations: ['messageChannelMessageAssociations'],
            },
-            take: limit,
-            skip: offset,
-            relations: ['messageChannelMessageAssociations'],
-          },
-          transactionManager,
-        );
+            transactionManager,
+          );

-        return nonAssociatedMessages.map(({ id }) => id);
-      },
-      async (
-        ids: string[],
-        workspaceId: string,
-        transactionManager?: EntityManager,
-      ) => {
-        await messageRepository.delete(ids, transactionManager);
-      },
-    );
+          return nonAssociatedMessages.map(({ id }) => id);
+        },
+        async (
+          ids: string[],
+          workspaceId: string,
+          transactionManager?: EntityManager,
+        ) => {
+          await messageRepository.delete(ids, transactionManager);
+        },
+        transactionManager,
+      );

-    await deleteUsingPagination(
-      workspaceId,
-      500,
-      async (
-        limit: number,
-        offset: number,
-        workspaceId: string,
-        transactionManager?: EntityManager,
-      ) => {
-        const orphanThreads = await messageThreadRepository.find(
-          {
-            where: {
-              messages: [],
+      await deleteUsingPagination(
+        workspaceId,
+        500,
+        async (
+          limit: number,
+          offset: number,
+          workspaceId: string,
+          transactionManager?: EntityManager,
+        ) => {
+          const orphanThreads = await messageThreadRepository.find(
+            {
+              where: {
+                messages: {
+                  id: IsNull(),
+                },
+              },
+              take: limit,
+              skip: offset,
            },
-            take: limit,
-            skip: offset,
-          },
-          transactionManager,
-        );
+            transactionManager,
+          );

-        return orphanThreads.map(({ id }) => id);
-      },
-      async (
-        ids: string[],
-        workspaceId: string,
-        transactionManager?: EntityManager,
-      ) => {
-        await messageThreadRepository.delete(ids, transactionManager);
-      },
-    );
+          return orphanThreads.map(({ id }) => id);
+        },
+        async (
+          ids: string[],
+          workspaceId: string,
+          transactionManager?: EntityManager,
+        ) => {
+          await messageThreadRepository.delete(ids, transactionManager);
+        },
+        transactionManager,
+      );
+    });
  }
 }
--- a/packages/twenty-server/src/modules/messaging/message-import-manager/jobs/messaging-ongoing-stale.job.ts
+++ b/packages/twenty-server/src/modules/messaging/message-import-manager/jobs/messaging-ongoing-stale.job.ts
@ -55,20 +55,20 @@ export class MessagingOngoingStaleJob {
          `Sync for message channel ${messageChannel.id} and workspace ${workspaceId} is stale. Setting sync stage to MESSAGES_IMPORT_PENDING`,
        );

-        await this.messageChannelSyncStatusService.resetSyncStageStartedAt(
+        await this.messageChannelSyncStatusService.resetSyncStageStartedAt([
          messageChannel.id,
-        );
+        ]);

        switch (messageChannel.syncStage) {
          case MessageChannelSyncStage.MESSAGE_LIST_FETCH_ONGOING:
            await this.messageChannelSyncStatusService.schedulePartialMessageListFetch(
-              messageChannel.id,
+              [messageChannel.id],
            );
            break;
          case MessageChannelSyncStage.MESSAGES_IMPORT_ONGOING:
-            await this.messageChannelSyncStatusService.scheduleMessagesImport(
+            await this.messageChannelSyncStatusService.scheduleMessagesImport([
              messageChannel.id,
-            );
+            ]);
            break;
          default:
            break;
--- a/packages/twenty-server/src/modules/messaging/message-import-manager/services/message-import-exception-handler.service.ts
+++ b/packages/twenty-server/src/modules/messaging/message-import-manager/services/message-import-exception-handler.service.ts
@ -79,7 +79,7 @@ export class MessageImportExceptionHandlerService {
  ): Promise<void> {
    if (messageChannel.throttleFailureCount >= CALENDAR_THROTTLE_MAX_ATTEMPTS) {
      await this.messageChannelSyncStatusService.markAsFailedUnknownAndFlushMessagesToImport(
-        messageChannel.id,
+        [messageChannel.id],
        workspaceId,
      );

@ -92,9 +92,7 @@ export class MessageImportExceptionHandlerService {
      );

    await messageChannelRepository.increment(
-      {
-        id: messageChannel.id,
-      },
+      { id: messageChannel.id },
      'throttleFailureCount',
      1,
    );
@ -102,20 +100,20 @@ export class MessageImportExceptionHandlerService {
    switch (syncStep) {
      case MessageImportSyncStep.FULL_MESSAGE_LIST_FETCH:
        await this.messageChannelSyncStatusService.scheduleFullMessageListFetch(
-          messageChannel.id,
+          [messageChannel.id],
        );
        break;

      case MessageImportSyncStep.PARTIAL_MESSAGE_LIST_FETCH:
        await this.messageChannelSyncStatusService.schedulePartialMessageListFetch(
-          messageChannel.id,
+          [messageChannel.id],
        );
        break;

      case MessageImportSyncStep.MESSAGES_IMPORT:
-        await this.messageChannelSyncStatusService.scheduleMessagesImport(
+        await this.messageChannelSyncStatusService.scheduleMessagesImport([
          messageChannel.id,
-        );
+        ]);
        break;

      default:
@ -128,7 +126,7 @@ export class MessageImportExceptionHandlerService {
    workspaceId: string,
  ): Promise<void> {
    await this.messageChannelSyncStatusService.markAsFailedInsufficientPermissionsAndFlushMessagesToImport(
-      messageChannel.id,
+      [messageChannel.id],
      workspaceId,
    );
  }
@ -139,7 +137,7 @@ export class MessageImportExceptionHandlerService {
    workspaceId: string,
  ): Promise<void> {
    await this.messageChannelSyncStatusService.markAsFailedUnknownAndFlushMessagesToImport(
-      messageChannel.id,
+      [messageChannel.id],
      workspaceId,
    );

@ -159,7 +157,7 @@ export class MessageImportExceptionHandlerService {
    }

    await this.messageChannelSyncStatusService.resetAndScheduleFullMessageListFetch(
-      messageChannel.id,
+      [messageChannel.id],
      workspaceId,
    );
  }
--- a/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-full-message-list-fetch.service.ts
+++ b/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-full-message-list-fetch.service.ts
@ -34,7 +34,7 @@ export class MessagingFullMessageListFetchService {
  ) {
    try {
      await this.messageChannelSyncStatusService.markAsMessagesListFetchOngoing(
-        messageChannel.id,
+        [messageChannel.id],
      );

      const { messageExternalIds, nextSyncCursor } =
@ -95,9 +95,9 @@ export class MessagingFullMessageListFetchService {
        },
      );

-      await this.messageChannelSyncStatusService.scheduleMessagesImport(
+      await this.messageChannelSyncStatusService.scheduleMessagesImport([
        messageChannel.id,
-      );
+      ]);
    } catch (error) {
      await this.messageImportErrorHandlerService.handleDriverException(
        error,
--- a/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-messages-import.service.ts
+++ b/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-messages-import.service.ts
@ -1,7 +1,5 @@
 import { Injectable, Logger } from '@nestjs/common';

-import { FeatureFlagKey } from 'src/engine/core-modules/feature-flag/enums/feature-flag-key.enum';
-import { FeatureFlagService } from 'src/engine/core-modules/feature-flag/services/feature-flag.service';
 import { CacheStorageService } from 'src/engine/integrations/cache-storage/cache-storage.service';
 import { InjectCacheStorage } from 'src/engine/integrations/cache-storage/decorators/cache-storage.decorator';
 import { CacheStorageNamespace } from 'src/engine/integrations/cache-storage/types/cache-storage-namespace.enum';
@ -44,7 +42,6 @@ export class MessagingMessagesImportService {
    @InjectObjectMetadataRepository(BlocklistWorkspaceEntity)
    private readonly blocklistRepository: BlocklistRepository,
    private readonly emailAliasManagerService: EmailAliasManagerService,
-    private readonly isFeatureEnabledService: FeatureFlagService,
    private readonly twentyORMManager: TwentyORMManager,
    private readonly messagingGetMessagesService: MessagingGetMessagesService,
    private readonly messageImportErrorHandlerService: MessageImportExceptionHandlerService,
@ -76,9 +73,9 @@ export class MessagingMessagesImportService {
        `Messaging import for workspace ${workspaceId} and account ${connectedAccount.id} starting...`,
      );

-      await this.messageChannelSyncStatusService.markAsMessagesImportOngoing(
+      await this.messageChannelSyncStatusService.markAsMessagesImportOngoing([
        messageChannel.id,
-      );
+      ]);

      try {
        connectedAccount.accessToken =
@ -111,17 +108,10 @@ export class MessagingMessagesImportService {
        }
      }

-      if (
-        await this.isFeatureEnabledService.isFeatureEnabled(
-          FeatureFlagKey.IsMessagingAliasFetchingEnabled,
-          workspaceId,
-        )
-      ) {
-        await this.emailAliasManagerService.refreshHandleAliases(
-          connectedAccount,
-          workspaceId,
-        );
-      }
+      await this.emailAliasManagerService.refreshHandleAliases(
+        connectedAccount,
+        workspaceId,
+      );

      messageIdsToFetch = await this.cacheStorage.setPop(
        `messages-to-import:${workspaceId}:gmail:${messageChannel.id}`,
@ -130,7 +120,7 @@ export class MessagingMessagesImportService {

      if (!messageIdsToFetch?.length) {
        await this.messageChannelSyncStatusService.markAsCompletedAndSchedulePartialMessageListFetch(
-          messageChannel.id,
+          [messageChannel.id],
        );

        return await this.trackMessageImportCompleted(
@ -151,7 +141,7 @@ export class MessagingMessagesImportService {
      );

      const messagesToSave = filterEmails(
-        messageChannel.handle,
+        [messageChannel.handle, ...connectedAccount.handleAliases.split(',')],
        allMessages,
        blocklist.map((blocklistItem) => blocklistItem.handle),
      );
@ -167,12 +157,12 @@ export class MessagingMessagesImportService {
        messageIdsToFetch.length < MESSAGING_GMAIL_USERS_MESSAGES_GET_BATCH_SIZE
      ) {
        await this.messageChannelSyncStatusService.markAsCompletedAndSchedulePartialMessageListFetch(
-          messageChannel.id,
+          [messageChannel.id],
        );
      } else {
-        await this.messageChannelSyncStatusService.scheduleMessagesImport(
+        await this.messageChannelSyncStatusService.scheduleMessagesImport([
          messageChannel.id,
-        );
+        ]);
      }

      const messageChannelRepository =
--- a/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-partial-message-list-fetch.service.ts
+++ b/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-partial-message-list-fetch.service.ts
@ -38,7 +38,7 @@ export class MessagingPartialMessageListFetchService {
  ): Promise<void> {
    try {
      await this.messageChannelSyncStatusService.markAsMessagesListFetchOngoing(
-        messageChannel.id,
+        [messageChannel.id],
      );

      const messageChannelRepository =
@ -70,7 +70,7 @@ export class MessagingPartialMessageListFetchService {
        );

        await this.messageChannelSyncStatusService.markAsCompletedAndSchedulePartialMessageListFetch(
-          messageChannel.id,
+          [messageChannel.id],
        );

        return;
@ -110,9 +110,9 @@ export class MessagingPartialMessageListFetchService {
        );
      }

-      await this.messageChannelSyncStatusService.scheduleMessagesImport(
+      await this.messageChannelSyncStatusService.scheduleMessagesImport([
        messageChannel.id,
-      );
+      ]);
    } catch (error) {
      await this.messageImportErrorHandlerService.handleDriverException(
        error,
--- a/packages/twenty-server/src/modules/messaging/message-import-manager/utils/filter-emails.util.ts
+++ b/packages/twenty-server/src/modules/messaging/message-import-manager/utils/filter-emails.util.ts
@ -3,19 +3,19 @@ import { MessageWithParticipants } from 'src/modules/messaging/message-import-ma

 // Todo: refactor this into several utils
 export const filterEmails = (
-  messageChannelHandle: string,
+  messageChannelHandles: string[],
  messages: MessageWithParticipants[],
  blocklist: string[],
 ) => {
  return filterOutBlocklistedMessages(
-    messageChannelHandle,
+    messageChannelHandles,
    filterOutIcsAttachments(messages),
    blocklist,
  );
 };

 const filterOutBlocklistedMessages = (
-  messageChannelHandle: string,
+  messageChannelHandles: string[],
  messages: MessageWithParticipants[],
  blocklist: string[],
 ) => {
@ -27,7 +27,7 @@ const filterOutBlocklistedMessages = (
    return message.participants.every(
      (participant) =>
        !isEmailBlocklisted(
-          messageChannelHandle,
+          messageChannelHandles,
          participant.handle,
          blocklist,
        ),