message channel change #3 (#15757)

Co-authored-by: Charles Bochet <charles@twenty.com>
2025-11-13 18:42:20 +05:30
parent a3b78f080b
commit 3ef9be6622
32 changed files with 1116 additions and 171 deletions
@@ -0,0 +1,35 @@
+import { Command, CommandRunner } from 'nest-commander';
+
+import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator';
+import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants';
+import { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service';
+import {
+  MESSAGING_PROCESS_FOLDER_ACTIONS_CRON_PATTERN,
+  MessagingProcessFolderActionsCronJob,
+} from 'src/modules/messaging/message-import-manager/crons/jobs/messaging-process-folder-actions.cron.job';
+
+@Command({
+  name: 'cron:messaging:process-folder-actions',
+  description:
+    'Starts a cron job to process pending folder actions (deletion) for message channels',
+})
+export class MessagingProcessFolderActionsCronCommand extends CommandRunner {
+  constructor(
+    @InjectMessageQueue(MessageQueue.cronQueue)
+    private readonly messageQueueService: MessageQueueService,
+  ) {
+    super();
+  }
+
+  async run(): Promise<void> {
+    await this.messageQueueService.addCron<undefined>({
+      jobName: MessagingProcessFolderActionsCronJob.name,
+      data: undefined,
+      options: {
+        repeat: {
+          pattern: MESSAGING_PROCESS_FOLDER_ACTIONS_CRON_PATTERN,
+        },
+      },
+    });
+  }
+}
@@ -0,0 +1,35 @@
+import { Command, CommandRunner } from 'nest-commander';
+
+import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator';
+import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants';
+import { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service';
+import {
+  MESSAGING_PROCESS_GROUP_EMAIL_ACTIONS_CRON_PATTERN,
+  MessagingProcessGroupEmailActionsCronJob,
+} from 'src/modules/messaging/message-import-manager/crons/jobs/messaging-process-group-email-actions.cron.job';
+
+@Command({
+  name: 'cron:messaging:process-group-email-actions',
+  description:
+    'Starts a cron job to process pending group email actions (deletion or import) for message channels',
+})
+export class MessagingProcessGroupEmailActionsCronCommand extends CommandRunner {
+  constructor(
+    @InjectMessageQueue(MessageQueue.cronQueue)
+    private readonly messageQueueService: MessageQueueService,
+  ) {
+    super();
+  }
+
+  async run(): Promise<void> {
+    await this.messageQueueService.addCron<undefined>({
+      jobName: MessagingProcessGroupEmailActionsCronJob.name,
+      data: undefined,
+      options: {
+        repeat: {
+          pattern: MESSAGING_PROCESS_GROUP_EMAIL_ACTIONS_CRON_PATTERN,
+        },
+      },
+    });
+  }
+}
@@ -0,0 +1,76 @@
+import { InjectDataSource, InjectRepository } from '@nestjs/typeorm';
+
+import { WorkspaceActivationStatus } from 'twenty-shared/workspace';
+import { DataSource, Repository } from 'typeorm';
+
+import { SentryCronMonitor } from 'src/engine/core-modules/cron/sentry-cron-monitor.decorator';
+import { ExceptionHandlerService } from 'src/engine/core-modules/exception-handler/exception-handler.service';
+import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator';
+import { Process } from 'src/engine/core-modules/message-queue/decorators/process.decorator';
+import { Processor } from 'src/engine/core-modules/message-queue/decorators/processor.decorator';
+import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants';
+import { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service';
+import { WorkspaceEntity } from 'src/engine/core-modules/workspace/workspace.entity';
+import { getWorkspaceSchemaName } from 'src/engine/workspace-datasource/utils/get-workspace-schema-name.util';
+import { MessageFolderPendingSyncAction } from 'src/modules/messaging/common/standard-objects/message-folder.workspace-entity';
+import {
+  MessagingProcessFolderActionsJob,
+  type MessagingProcessFolderActionsJobData,
+} from 'src/modules/messaging/message-import-manager/jobs/messaging-process-folder-actions.job';
+
+export const MESSAGING_PROCESS_FOLDER_ACTIONS_CRON_PATTERN = '*/15 * * * *';
+
+@Processor(MessageQueue.cronQueue)
+export class MessagingProcessFolderActionsCronJob {
+  constructor(
+    @InjectRepository(WorkspaceEntity)
+    private readonly workspaceRepository: Repository<WorkspaceEntity>,
+    @InjectMessageQueue(MessageQueue.messagingQueue)
+    private readonly messageQueueService: MessageQueueService,
+    @InjectDataSource()
+    private readonly coreDataSource: DataSource,
+    private readonly exceptionHandlerService: ExceptionHandlerService,
+  ) {}
+
+  @Process(MessagingProcessFolderActionsCronJob.name)
+  @SentryCronMonitor(
+    MessagingProcessFolderActionsCronJob.name,
+    MESSAGING_PROCESS_FOLDER_ACTIONS_CRON_PATTERN,
+  )
+  async handle(): Promise<void> {
+    const activeWorkspaces = await this.workspaceRepository.find({
+      where: {
+        activationStatus: WorkspaceActivationStatus.ACTIVE,
+      },
+    });
+
+    for (const activeWorkspace of activeWorkspaces) {
+      try {
+        const schemaName = getWorkspaceSchemaName(activeWorkspace.id);
+
+        const messageChannels = await this.coreDataSource.query(
+          `SELECT DISTINCT mc.id
+           FROM ${schemaName}."messageChannel" mc
+           INNER JOIN ${schemaName}."messageFolder" mf ON mf."messageChannelId" = mc.id
+           WHERE mf."pendingSyncAction" = '${MessageFolderPendingSyncAction.FOLDER_DELETION}'`,
+        );
+
+        for (const messageChannel of messageChannels) {
+          await this.messageQueueService.add<MessagingProcessFolderActionsJobData>(
+            MessagingProcessFolderActionsJob.name,
+            {
+              workspaceId: activeWorkspace.id,
+              messageChannelId: messageChannel.id,
+            },
+          );
+        }
+      } catch (error) {
+        this.exceptionHandlerService.captureExceptions([error], {
+          workspace: {
+            id: activeWorkspace.id,
+          },
+        });
+      }
+    }
+  }
+}
@@ -0,0 +1,73 @@
+import { InjectDataSource, InjectRepository } from '@nestjs/typeorm';
+
+import { WorkspaceActivationStatus } from 'twenty-shared/workspace';
+import { DataSource, Repository } from 'typeorm';
+
+import { SentryCronMonitor } from 'src/engine/core-modules/cron/sentry-cron-monitor.decorator';
+import { ExceptionHandlerService } from 'src/engine/core-modules/exception-handler/exception-handler.service';
+import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator';
+import { Process } from 'src/engine/core-modules/message-queue/decorators/process.decorator';
+import { Processor } from 'src/engine/core-modules/message-queue/decorators/processor.decorator';
+import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants';
+import { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service';
+import { WorkspaceEntity } from 'src/engine/core-modules/workspace/workspace.entity';
+import { getWorkspaceSchemaName } from 'src/engine/workspace-datasource/utils/get-workspace-schema-name.util';
+import { MessageChannelPendingGroupEmailsAction } from 'src/modules/messaging/common/standard-objects/message-channel.workspace-entity';
+import {
+  MessagingProcessGroupEmailActionsJob,
+  type MessagingProcessGroupEmailActionsJobData,
+} from 'src/modules/messaging/message-import-manager/jobs/messaging-process-group-email-actions.job';
+
+export const MESSAGING_PROCESS_GROUP_EMAIL_ACTIONS_CRON_PATTERN = '0 */2 * * *';
+
+@Processor(MessageQueue.cronQueue)
+export class MessagingProcessGroupEmailActionsCronJob {
+  constructor(
+    @InjectRepository(WorkspaceEntity)
+    private readonly workspaceRepository: Repository<WorkspaceEntity>,
+    @InjectMessageQueue(MessageQueue.messagingQueue)
+    private readonly messageQueueService: MessageQueueService,
+    @InjectDataSource()
+    private readonly coreDataSource: DataSource,
+    private readonly exceptionHandlerService: ExceptionHandlerService,
+  ) {}
+
+  @Process(MessagingProcessGroupEmailActionsCronJob.name)
+  @SentryCronMonitor(
+    MessagingProcessGroupEmailActionsCronJob.name,
+    MESSAGING_PROCESS_GROUP_EMAIL_ACTIONS_CRON_PATTERN,
+  )
+  async handle(): Promise<void> {
+    const activeWorkspaces = await this.workspaceRepository.find({
+      where: {
+        activationStatus: WorkspaceActivationStatus.ACTIVE,
+      },
+    });
+
+    for (const activeWorkspace of activeWorkspaces) {
+      try {
+        const schemaName = getWorkspaceSchemaName(activeWorkspace.id);
+
+        const messageChannels = await this.coreDataSource.query(
+          `SELECT id FROM ${schemaName}."messageChannel" WHERE "pendingGroupEmailsAction" IN ('${MessageChannelPendingGroupEmailsAction.GROUP_EMAILS_DELETION}', '${MessageChannelPendingGroupEmailsAction.GROUP_EMAILS_IMPORT}')`,
+        );
+
+        for (const messageChannel of messageChannels) {
+          await this.messageQueueService.add<MessagingProcessGroupEmailActionsJobData>(
+            MessagingProcessGroupEmailActionsJob.name,
+            {
+              workspaceId: activeWorkspace.id,
+              messageChannelId: messageChannel.id,
+            },
+          );
+        }
+      } catch (error) {
+        this.exceptionHandlerService.captureExceptions([error], {
+          workspace: {
+            id: activeWorkspace.id,
+          },
+        });
+      }
+    }
+  }
+}
@@ -0,0 +1,92 @@
+import { Logger, Scope } from '@nestjs/common';
+
+import { Process } from 'src/engine/core-modules/message-queue/decorators/process.decorator';
+import { Processor } from 'src/engine/core-modules/message-queue/decorators/processor.decorator';
+import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants';
+import { TwentyORMManager } from 'src/engine/twenty-orm/twenty-orm.manager';
+import { type MessageChannelWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message-channel.workspace-entity';
+import {
+  MessageFolderPendingSyncAction,
+  type MessageFolderWorkspaceEntity,
+} from 'src/modules/messaging/common/standard-objects/message-folder.workspace-entity';
+import { MessagingProcessFolderActionsService } from 'src/modules/messaging/message-import-manager/services/messaging-process-folder-actions.service';
+
+export type MessagingProcessFolderActionsJobData = {
+  workspaceId: string;
+  messageChannelId: string;
+};
+
+@Processor({
+  queueName: MessageQueue.messagingQueue,
+  scope: Scope.REQUEST,
+})
+export class MessagingProcessFolderActionsJob {
+  private readonly logger = new Logger(MessagingProcessFolderActionsJob.name);
+
+  constructor(
+    private readonly twentyORMManager: TwentyORMManager,
+    private readonly messagingProcessFolderActionsService: MessagingProcessFolderActionsService,
+  ) {}
+
+  @Process(MessagingProcessFolderActionsJob.name)
+  async handle(data: MessagingProcessFolderActionsJobData): Promise<void> {
+    const { workspaceId, messageChannelId } = data;
+
+    this.logger.log(
+      `Processing pending folder actions for message channel ${messageChannelId} in workspace ${workspaceId}`,
+    );
+
+    const messageChannelRepository =
+      await this.twentyORMManager.getRepository<MessageChannelWorkspaceEntity>(
+        'messageChannel',
+      );
+
+    const messageChannel = await messageChannelRepository.findOne({
+      where: {
+        id: messageChannelId,
+      },
+    });
+
+    if (!messageChannel) {
+      this.logger.warn(
+        `Message channel ${messageChannelId} not found in workspace ${workspaceId}`,
+      );
+
+      return;
+    }
+
+    const messageFolderRepository =
+      await this.twentyORMManager.getRepository<MessageFolderWorkspaceEntity>(
+        'messageFolder',
+      );
+
+    const messageFolders = await messageFolderRepository.find({
+      where: {
+        messageChannelId: messageChannel.id,
+        pendingSyncAction: MessageFolderPendingSyncAction.FOLDER_DELETION,
+      },
+    });
+
+    if (messageFolders.length === 0) {
+      this.logger.log(
+        `Message channel ${messageChannelId} has no folders with pending deletion actions, skipping`,
+      );
+
+      return;
+    }
+
+    try {
+      await this.messagingProcessFolderActionsService.processFolderActions(
+        messageChannel,
+        messageFolders,
+        workspaceId,
+      );
+    } catch (error) {
+      this.logger.error(
+        `Error processing folder actions for message channel ${messageChannelId} in workspace ${workspaceId}: ${error.message}`,
+        error.stack,
+      );
+      throw error;
+    }
+  }
+}
@@ -0,0 +1,84 @@
+import { Logger, Scope } from '@nestjs/common';
+
+import { Process } from 'src/engine/core-modules/message-queue/decorators/process.decorator';
+import { Processor } from 'src/engine/core-modules/message-queue/decorators/processor.decorator';
+import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants';
+import { TwentyORMManager } from 'src/engine/twenty-orm/twenty-orm.manager';
+import {
+  MessageChannelPendingGroupEmailsAction,
+  type MessageChannelWorkspaceEntity,
+} from 'src/modules/messaging/common/standard-objects/message-channel.workspace-entity';
+import { MessagingProcessGroupEmailActionsService } from 'src/modules/messaging/message-import-manager/services/messaging-process-group-email-actions.service';
+
+export type MessagingProcessGroupEmailActionsJobData = {
+  workspaceId: string;
+  messageChannelId: string;
+};
+
+@Processor({
+  queueName: MessageQueue.messagingQueue,
+  scope: Scope.REQUEST,
+})
+export class MessagingProcessGroupEmailActionsJob {
+  private readonly logger = new Logger(
+    MessagingProcessGroupEmailActionsJob.name,
+  );
+
+  constructor(
+    private readonly twentyORMManager: TwentyORMManager,
+    private readonly messagingProcessGroupEmailActionsService: MessagingProcessGroupEmailActionsService,
+  ) {}
+
+  @Process(MessagingProcessGroupEmailActionsJob.name)
+  async handle(data: MessagingProcessGroupEmailActionsJobData): Promise<void> {
+    const { workspaceId, messageChannelId } = data;
+
+    this.logger.log(
+      `Processing pending group email action for message channel ${messageChannelId} in workspace ${workspaceId}`,
+    );
+
+    const messageChannelRepository =
+      await this.twentyORMManager.getRepository<MessageChannelWorkspaceEntity>(
+        'messageChannel',
+      );
+
+    const messageChannel = await messageChannelRepository.findOne({
+      where: {
+        id: messageChannelId,
+      },
+    });
+
+    if (!messageChannel) {
+      this.logger.warn(
+        `Message channel ${messageChannelId} not found in workspace ${workspaceId}`,
+      );
+
+      return;
+    }
+
+    if (
+      messageChannel.pendingGroupEmailsAction ===
+        MessageChannelPendingGroupEmailsAction.NONE ||
+      !messageChannel.pendingGroupEmailsAction
+    ) {
+      this.logger.log(
+        `Message channel ${messageChannelId} no longer has a pending action, skipping`,
+      );
+
+      return;
+    }
+
+    try {
+      await this.messagingProcessGroupEmailActionsService.processGroupEmailActions(
+        messageChannel,
+        workspaceId,
+      );
+    } catch (error) {
+      this.logger.error(
+        `Error processing group email actions for message channel ${messageChannelId} in workspace ${workspaceId}: ${error.message}`,
+        error.stack,
+      );
+      throw error;
+    }
+  }
+}
@@ -18,10 +18,14 @@ import { MessagingSingleMessageImportCommand } from 'src/modules/messaging/messa
 import { MessagingMessageListFetchCronCommand } from 'src/modules/messaging/message-import-manager/crons/commands/messaging-message-list-fetch.cron.command';
 import { MessagingMessagesImportCronCommand } from 'src/modules/messaging/message-import-manager/crons/commands/messaging-messages-import.cron.command';
 import { MessagingOngoingStaleCronCommand } from 'src/modules/messaging/message-import-manager/crons/commands/messaging-ongoing-stale.cron.command';
+import { MessagingProcessFolderActionsCronCommand } from 'src/modules/messaging/message-import-manager/crons/commands/messaging-process-folder-actions.cron.command';
+import { MessagingProcessGroupEmailActionsCronCommand } from 'src/modules/messaging/message-import-manager/crons/commands/messaging-process-group-email-actions.cron.command';
 import { MessagingRelaunchFailedMessageChannelsCronCommand } from 'src/modules/messaging/message-import-manager/crons/commands/messaging-relaunch-failed-message-channels.cron.command';
 import { MessagingMessageListFetchCronJob } from 'src/modules/messaging/message-import-manager/crons/jobs/messaging-message-list-fetch.cron.job';
 import { MessagingMessagesImportCronJob } from 'src/modules/messaging/message-import-manager/crons/jobs/messaging-messages-import.cron.job';
 import { MessagingOngoingStaleCronJob } from 'src/modules/messaging/message-import-manager/crons/jobs/messaging-ongoing-stale.cron.job';
+import { MessagingProcessFolderActionsCronJob } from 'src/modules/messaging/message-import-manager/crons/jobs/messaging-process-folder-actions.cron.job';
+import { MessagingProcessGroupEmailActionsCronJob } from 'src/modules/messaging/message-import-manager/crons/jobs/messaging-process-group-email-actions.cron.job';
 import { MessagingRelaunchFailedMessageChannelsCronJob } from 'src/modules/messaging/message-import-manager/crons/jobs/messaging-relaunch-failed-message-channels.cron.job';
 import { MessagingGmailDriverModule } from 'src/modules/messaging/message-import-manager/drivers/gmail/messaging-gmail-driver.module';
 import { MessagingIMAPDriverModule } from 'src/modules/messaging/message-import-manager/drivers/imap/messaging-imap-driver.module';
@@ -32,17 +36,23 @@ import { MessagingCleanCacheJob } from 'src/modules/messaging/message-import-man
 import { MessagingMessageListFetchJob } from 'src/modules/messaging/message-import-manager/jobs/messaging-message-list-fetch.job';
 import { MessagingMessagesImportJob } from 'src/modules/messaging/message-import-manager/jobs/messaging-messages-import.job';
 import { MessagingOngoingStaleJob } from 'src/modules/messaging/message-import-manager/jobs/messaging-ongoing-stale.job';
+import { MessagingProcessFolderActionsJob } from 'src/modules/messaging/message-import-manager/jobs/messaging-process-folder-actions.job';
+import { MessagingProcessGroupEmailActionsJob } from 'src/modules/messaging/message-import-manager/jobs/messaging-process-group-email-actions.job';
 import { MessagingRelaunchFailedMessageChannelJob } from 'src/modules/messaging/message-import-manager/jobs/messaging-relaunch-failed-message-channel.job';
 import { MessagingMessageImportManagerMessageChannelListener } from 'src/modules/messaging/message-import-manager/listeners/messaging-import-manager-message-channel.listener';
 import { MessagingAccountAuthenticationService } from 'src/modules/messaging/message-import-manager/services/messaging-account-authentication.service';
 import { MessagingClearCursorsModule } from 'src/modules/messaging/message-import-manager/services/messaging-clear-cursors.module';
 import { MessagingCursorService } from 'src/modules/messaging/message-import-manager/services/messaging-cursor.service';
+import { MessagingDeleteFolderMessagesService } from 'src/modules/messaging/message-import-manager/services/messaging-delete-folder-messages.service';
+import { MessagingDeleteGroupEmailMessagesService } from 'src/modules/messaging/message-import-manager/services/messaging-delete-group-email-messages.service';
 import { MessagingGetMessageListService } from 'src/modules/messaging/message-import-manager/services/messaging-get-message-list.service';
 import { MessagingGetMessagesService } from 'src/modules/messaging/message-import-manager/services/messaging-get-messages.service';
 import { MessageImportExceptionHandlerService } from 'src/modules/messaging/message-import-manager/services/messaging-import-exception-handler.service';
 import { MessagingMessageListFetchService } from 'src/modules/messaging/message-import-manager/services/messaging-message-list-fetch.service';
 import { MessagingMessageService } from 'src/modules/messaging/message-import-manager/services/messaging-message.service';
 import { MessagingMessagesImportService } from 'src/modules/messaging/message-import-manager/services/messaging-messages-import.service';
+import { MessagingProcessFolderActionsService } from 'src/modules/messaging/message-import-manager/services/messaging-process-folder-actions.service';
+import { MessagingProcessGroupEmailActionsService } from 'src/modules/messaging/message-import-manager/services/messaging-process-group-email-actions.service';
 import { MessagingSaveMessagesAndEnqueueContactCreationService } from 'src/modules/messaging/message-import-manager/services/messaging-save-messages-and-enqueue-contact-creation.service';
 import { MessagingSendMessageService } from 'src/modules/messaging/message-import-manager/services/messaging-send-message.service';
 import { MessageParticipantManagerModule } from 'src/modules/messaging/message-participant-manager/message-participant-manager.module';
@@ -76,15 +86,21 @@ import { MessagingMonitoringModule } from 'src/modules/messaging/monitoring/mess
    MessagingMessageListFetchCronCommand,
    MessagingMessagesImportCronCommand,
    MessagingOngoingStaleCronCommand,
+    MessagingProcessFolderActionsCronCommand,
+    MessagingProcessGroupEmailActionsCronCommand,
    MessagingRelaunchFailedMessageChannelsCronCommand,
    MessagingSingleMessageImportCommand,
    MessagingMessageListFetchJob,
    MessagingMessagesImportJob,
    MessagingOngoingStaleJob,
+    MessagingProcessFolderActionsJob,
+    MessagingProcessGroupEmailActionsJob,
    MessagingRelaunchFailedMessageChannelJob,
    MessagingMessageListFetchCronJob,
    MessagingMessagesImportCronJob,
    MessagingOngoingStaleCronJob,
+    MessagingProcessFolderActionsCronJob,
+    MessagingProcessGroupEmailActionsCronJob,
    MessagingRelaunchFailedMessageChannelsCronJob,
    MessagingAddSingleMessageToCacheForImportJob,
    MessagingMessageImportManagerMessageChannelListener,
@@ -99,6 +115,10 @@ import { MessagingMonitoringModule } from 'src/modules/messaging/monitoring/mess
    MessagingCursorService,
    MessagingSendMessageService,
    MessagingAccountAuthenticationService,
+    MessagingProcessFolderActionsService,
+    MessagingProcessGroupEmailActionsService,
+    MessagingDeleteFolderMessagesService,
+    MessagingDeleteGroupEmailMessagesService,
  ],
  exports: [
    MessagingSendMessageService,
@@ -106,6 +126,7 @@ import { MessagingMonitoringModule } from 'src/modules/messaging/monitoring/mess
    MessagingMessagesImportCronCommand,
    MessagingOngoingStaleCronCommand,
    MessagingRelaunchFailedMessageChannelsCronCommand,
+    MessagingProcessGroupEmailActionsService,
  ],
 })
 export class MessagingImportManagerModule {}
@@ -0,0 +1,77 @@
+import { Injectable, Logger } from '@nestjs/common';
+
+import chunk from 'lodash.chunk';
+import { isDefined } from 'twenty-shared/utils';
+
+import { type MessageChannelWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message-channel.workspace-entity';
+import { type MessageFolderWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message-folder.workspace-entity';
+import { MessagingMessageCleanerService } from 'src/modules/messaging/message-cleaner/services/messaging-message-cleaner.service';
+import { MessagingGetMessageListService } from 'src/modules/messaging/message-import-manager/services/messaging-get-message-list.service';
+
+@Injectable()
+export class MessagingDeleteFolderMessagesService {
+  private readonly logger = new Logger(
+    MessagingDeleteFolderMessagesService.name,
+  );
+
+  constructor(
+    private readonly messagingMessageCleanerService: MessagingMessageCleanerService,
+    private readonly messagingGetMessageListService: MessagingGetMessageListService,
+  ) {}
+
+  async deleteFolderMessages(
+    workspaceId: string,
+    messageChannel: MessageChannelWorkspaceEntity,
+    messageFolder: MessageFolderWorkspaceEntity,
+  ): Promise<number> {
+    this.logger.log(
+      `WorkspaceId: ${workspaceId}, MessageChannelId: ${messageChannel.id}, FolderId: ${messageFolder.id} - Deleting messages from folder: ${messageFolder.name}`,
+    );
+
+    const messageLists =
+      await this.messagingGetMessageListService.getMessageLists(
+        messageChannel,
+        [messageFolder],
+      );
+
+    let totalDeletedCount = 0;
+
+    for (const messageList of messageLists) {
+      const { messageExternalIds } = messageList;
+
+      if (messageExternalIds.length === 0) {
+        continue;
+      }
+
+      const messageExternalIdsChunks = chunk(messageExternalIds, 200);
+
+      for (const messageExternalIdsChunk of messageExternalIdsChunks) {
+        const validExternalIds = messageExternalIdsChunk.filter(isDefined);
+
+        if (validExternalIds.length === 0) {
+          continue;
+        }
+
+        await this.messagingMessageCleanerService.deleteMessagesChannelMessageAssociationsAndRelatedOrphans(
+          {
+            workspaceId,
+            messageExternalIds: validExternalIds,
+            messageChannelId: messageChannel.id,
+          },
+        );
+
+        totalDeletedCount += validExternalIds.length;
+
+        this.logger.log(
+          `WorkspaceId: ${workspaceId}, MessageChannelId: ${messageChannel.id}, FolderId: ${messageFolder.id} - Processed ${validExternalIds.length} message deletions`,
+        );
+      }
+    }
+
+    this.logger.log(
+      `WorkspaceId: ${workspaceId}, MessageChannelId: ${messageChannel.id}, FolderId: ${messageFolder.id} - Completed deleting ${totalDeletedCount} messages from folder: ${messageFolder.name}`,
+    );
+
+    return totalDeletedCount;
+  }
+}
@@ -0,0 +1,126 @@
+import { Injectable, Logger } from '@nestjs/common';
+
+import chunk from 'lodash.chunk';
+import { isDefined } from 'twenty-shared/utils';
+
+import { TwentyORMGlobalManager } from 'src/engine/twenty-orm/twenty-orm-global.manager';
+import { MessageChannelMessageAssociationWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message-channel-message-association.workspace-entity';
+import { MessagingMessageCleanerService } from 'src/modules/messaging/message-cleaner/services/messaging-message-cleaner.service';
+import { isGroupEmail } from 'src/modules/messaging/message-import-manager/utils/is-group-email';
+
+const MESSAGE_CHANNEL_MESSAGE_ASSOCIATION_BATCH_SIZE = 500;
+
+type MessageBatchRawResult = {
+  messageId: string;
+  messageExternalId: string;
+  participantHandle: string;
+};
+
+@Injectable()
+export class MessagingDeleteGroupEmailMessagesService {
+  private readonly logger = new Logger(
+    MessagingDeleteGroupEmailMessagesService.name,
+  );
+
+  constructor(
+    private readonly twentyORMGlobalManager: TwentyORMGlobalManager,
+    private readonly messagingMessageCleanerService: MessagingMessageCleanerService,
+  ) {}
+
+  async deleteGroupEmailMessages(
+    workspaceId: string,
+    messageChannelId: string,
+  ): Promise<number> {
+    this.logger.log(
+      `WorkspaceId: ${workspaceId}, MessageChannelId: ${messageChannelId} - Deleting messages from group email addresses`,
+    );
+
+    const messageChannelMessageAssociationRepository =
+      await this.twentyORMGlobalManager.getRepositoryForWorkspace<MessageChannelMessageAssociationWorkspaceEntity>(
+        workspaceId,
+        'messageChannelMessageAssociation',
+      );
+
+    let offset = 0;
+    let totalDeletedCount = 0;
+
+    while (true) {
+      const batch = await messageChannelMessageAssociationRepository
+        .createQueryBuilder('mcma')
+        .select('mcma.messageId', 'messageId')
+        .addSelect('mcma.messageExternalId', 'messageExternalId')
+        .addSelect('participant.handle', 'participantHandle')
+        .innerJoin('mcma.message', 'message')
+        .innerJoin(
+          'message.messageParticipants',
+          'participant',
+          'participant.role = :role',
+          { role: 'from' },
+        )
+        .where('mcma.messageChannelId = :messageChannelId', {
+          messageChannelId,
+        })
+        .skip(offset)
+        .take(MESSAGE_CHANNEL_MESSAGE_ASSOCIATION_BATCH_SIZE)
+        .getRawMany<MessageBatchRawResult>();
+
+      if (batch.length === 0) {
+        break;
+      }
+
+      const groupEmailRecords = batch.filter(
+        (record) =>
+          isDefined(record.participantHandle) &&
+          isGroupEmail(record.participantHandle),
+      );
+
+      if (groupEmailRecords.length > 0) {
+        const uniqueMessageIds = new Set(
+          groupEmailRecords.map((r) => r.messageId),
+        );
+
+        const messageExternalIdsToDelete = batch
+          .filter((record) => uniqueMessageIds.has(record.messageId))
+          .map((record) => record.messageExternalId)
+          .filter(isDefined);
+
+        if (messageExternalIdsToDelete.length > 0) {
+          const messageExternalIdsChunks = chunk(
+            messageExternalIdsToDelete,
+            200,
+          );
+
+          for (const messageExternalIdsChunk of messageExternalIdsChunks) {
+            await this.messagingMessageCleanerService.deleteMessagesChannelMessageAssociationsAndRelatedOrphans(
+              {
+                workspaceId,
+                messageExternalIds: messageExternalIdsChunk,
+                messageChannelId,
+              },
+            );
+
+            totalDeletedCount += messageExternalIdsChunk.length;
+
+            this.logger.log(
+              `WorkspaceId: ${workspaceId}, MessageChannelId: ${messageChannelId} - Deleted ${messageExternalIdsChunk.length} group email messages`,
+            );
+          }
+        }
+      }
+
+      if (batch.length < MESSAGE_CHANNEL_MESSAGE_ASSOCIATION_BATCH_SIZE) {
+        break;
+      }
+
+      if (groupEmailRecords.length === 0) {
+        offset += MESSAGE_CHANNEL_MESSAGE_ASSOCIATION_BATCH_SIZE;
+      }
+    }
+
+    this.logger.log(
+      `WorkspaceId: ${workspaceId}, MessageChannelId: ${messageChannelId} - Completed deleting ${totalDeletedCount} group email messages`,
+    );
+
+    return totalDeletedCount;
+  }
+}
@@ -12,10 +12,14 @@ import { TwentyORMManager } from 'src/engine/twenty-orm/twenty-orm.manager';
 import { MessageChannelSyncStatusService } from 'src/modules/messaging/common/services/message-channel-sync-status.service';
 import { type MessageChannelMessageAssociationWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message-channel-message-association.workspace-entity';
 import {
+  MessageChannelPendingGroupEmailsAction,
  MessageChannelSyncStage,
  MessageChannelWorkspaceEntity,
 } from 'src/modules/messaging/common/standard-objects/message-channel.workspace-entity';
-import { MessageFolderWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message-folder.workspace-entity';
+import {
+  MessageFolderPendingSyncAction,
+  MessageFolderWorkspaceEntity,
+} from 'src/modules/messaging/common/standard-objects/message-folder.workspace-entity';
 import { MessagingMessageCleanerService } from 'src/modules/messaging/message-cleaner/services/messaging-message-cleaner.service';
 import { SyncMessageFoldersService } from 'src/modules/messaging/message-folder-manager/services/sync-message-folders.service';
 import { MessagingAccountAuthenticationService } from 'src/modules/messaging/message-import-manager/services/messaging-account-authentication.service';
@@ -51,6 +55,19 @@ export class MessagingMessageListFetchService {
    workspaceId: string,
  ) {
    try {
+      if (
+        messageChannel.pendingGroupEmailsAction ===
+          MessageChannelPendingGroupEmailsAction.GROUP_EMAILS_DELETION ||
+        messageChannel.pendingGroupEmailsAction ===
+          MessageChannelPendingGroupEmailsAction.GROUP_EMAILS_IMPORT
+      ) {
+        this.logger.log(
+          `messageChannelId: ${messageChannel.id} Skipping message list fetch due to pending group emails action: ${messageChannel.pendingGroupEmailsAction}`,
+        );
+
+        return;
+      }
+
      await this.messageChannelSyncStatusService.markAsMessagesListFetchOngoing(
        [messageChannel.id],
      );
@@ -81,8 +98,7 @@ export class MessagingMessageListFetchService {

      await this.syncMessageFoldersService.syncMessageFolders({
        workspaceId,
-        messageChannelId: messageChannelWithFreshTokens.id,
-        connectedAccount: messageChannelWithFreshTokens.connectedAccount,
+        messageChannel: messageChannelWithFreshTokens,
        manager: datasource.manager,
      });

@@ -94,6 +110,7 @@ export class MessagingMessageListFetchService {
      const messageFolders = await messageFolderRepository.find({
        where: {
          messageChannelId: messageChannel.id,
+          pendingSyncAction: MessageFolderPendingSyncAction.NONE,
        },
      });

@@ -119,6 +119,7 @@ export class MessagingMessagesImportService {
        [...connectedAccountWithFreshTokens.handleAliases.split(',')],
        allMessages,
        blocklist.map((blocklistItem) => blocklistItem.handle),
+        messageChannel.excludeGroupEmails,
      );

      if (messagesToSave.length > 0) {
@@ -0,0 +1,124 @@
+import { Injectable, Logger } from '@nestjs/common';
+
+import { isDefined } from 'twenty-shared/utils';
+import { In } from 'typeorm';
+
+import { type WorkspaceEntityManager } from 'src/engine/twenty-orm/entity-manager/workspace-entity-manager';
+import { TwentyORMManager } from 'src/engine/twenty-orm/twenty-orm.manager';
+import { type MessageChannelWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message-channel.workspace-entity';
+import {
+  MessageFolderPendingSyncAction,
+  MessageFolderWorkspaceEntity,
+} from 'src/modules/messaging/common/standard-objects/message-folder.workspace-entity';
+import { MessagingDeleteFolderMessagesService } from 'src/modules/messaging/message-import-manager/services/messaging-delete-folder-messages.service';
+
+@Injectable()
+export class MessagingProcessFolderActionsService {
+  private readonly logger = new Logger(
+    MessagingProcessFolderActionsService.name,
+  );
+
+  constructor(
+    private readonly twentyORMManager: TwentyORMManager,
+    private readonly messagingDeleteFolderMessagesService: MessagingDeleteFolderMessagesService,
+  ) {}
+
+  async processFolderActions(
+    messageChannel: MessageChannelWorkspaceEntity,
+    messageFolders: MessageFolderWorkspaceEntity[],
+    workspaceId: string,
+  ): Promise<void> {
+    const foldersWithPendingActions = messageFolders.filter(
+      (folder) =>
+        isDefined(folder.pendingSyncAction) &&
+        folder.pendingSyncAction !== MessageFolderPendingSyncAction.NONE,
+    );
+
+    if (foldersWithPendingActions.length === 0) {
+      return;
+    }
+
+    this.logger.log(
+      `WorkspaceId: ${workspaceId}, MessageChannelId: ${messageChannel.id} - Processing ${foldersWithPendingActions.length} folders with pending actions`,
+    );
+
+    const folderIdsToDelete: string[] = [];
+    const processedFolderIds: string[] = [];
+    const failedFolderIds: Array<{ folderId: string; error: Error }> = [];
+
+    for (const folder of foldersWithPendingActions) {
+      try {
+        this.logger.log(
+          `WorkspaceId: ${workspaceId}, MessageChannelId: ${messageChannel.id}, FolderId: ${folder.id} - Processing folder action: ${folder.pendingSyncAction}`,
+        );
+
+        if (
+          folder.pendingSyncAction ===
+          MessageFolderPendingSyncAction.FOLDER_DELETION
+        ) {
+          await this.messagingDeleteFolderMessagesService.deleteFolderMessages(
+            workspaceId,
+            messageChannel,
+            folder,
+          );
+
+          folderIdsToDelete.push(folder.id);
+
+          this.logger.log(
+            `WorkspaceId: ${workspaceId}, MessageChannelId: ${messageChannel.id}, FolderId: ${folder.id} - Completed FOLDER_DELETION action`,
+          );
+        }
+
+        processedFolderIds.push(folder.id);
+      } catch (error) {
+        this.logger.error(
+          `WorkspaceId: ${workspaceId}, MessageChannelId: ${messageChannel.id}, FolderId: ${folder.id} - Error processing folder action: ${error.message}`,
+          error.stack,
+        );
+        failedFolderIds.push({ folderId: folder.id, error });
+      }
+    }
+
+    if (failedFolderIds.length > 0) {
+      this.logger.warn(
+        `WorkspaceId: ${workspaceId}, MessageChannelId: ${messageChannel.id} - Failed to process ${failedFolderIds.length} folders. They will be retried on next sync.`,
+      );
+    }
+
+    if (processedFolderIds.length > 0 || folderIdsToDelete.length > 0) {
+      const workspaceDataSource = await this.twentyORMManager.getDatasource();
+
+      await workspaceDataSource?.transaction(
+        async (transactionManager: WorkspaceEntityManager) => {
+          const messageFolderRepository =
+            await this.twentyORMManager.getRepository<MessageFolderWorkspaceEntity>(
+              'messageFolder',
+            );
+
+          if (processedFolderIds.length > 0) {
+            await messageFolderRepository.update(
+              { id: In(processedFolderIds) },
+              { pendingSyncAction: MessageFolderPendingSyncAction.NONE },
+              transactionManager,
+            );
+
+            this.logger.log(
+              `WorkspaceId: ${workspaceId}, MessageChannelId: ${messageChannel.id} - Reset pendingSyncAction to NONE for ${processedFolderIds.length} folders`,
+            );
+          }
+
+          if (folderIdsToDelete.length > 0) {
+            await messageFolderRepository.delete(
+              { id: In(folderIdsToDelete) },
+              transactionManager,
+            );
+
+            this.logger.log(
+              `WorkspaceId: ${workspaceId}, MessageChannelId: ${messageChannel.id} - Deleted ${folderIdsToDelete.length} folders`,
+            );
+          }
+        },
+      );
+    }
+  }
+}
@@ -0,0 +1,157 @@
+import { Injectable, Logger } from '@nestjs/common';
+
+import { isDefined } from 'twenty-shared/utils';
+
+import { type WorkspaceEntityManager } from 'src/engine/twenty-orm/entity-manager/workspace-entity-manager';
+import { TwentyORMManager } from 'src/engine/twenty-orm/twenty-orm.manager';
+import { MessageChannelSyncStatusService } from 'src/modules/messaging/common/services/message-channel-sync-status.service';
+import {
+  MessageChannelPendingGroupEmailsAction,
+  MessageChannelWorkspaceEntity,
+} from 'src/modules/messaging/common/standard-objects/message-channel.workspace-entity';
+import { MessagingClearCursorsService } from 'src/modules/messaging/message-import-manager/services/messaging-clear-cursors.service';
+import { MessagingDeleteGroupEmailMessagesService } from 'src/modules/messaging/message-import-manager/services/messaging-delete-group-email-messages.service';
+
+@Injectable()
+export class MessagingProcessGroupEmailActionsService {
+  private readonly logger = new Logger(
+    MessagingProcessGroupEmailActionsService.name,
+  );
+
+  constructor(
+    private readonly twentyORMManager: TwentyORMManager,
+    private readonly messagingDeleteGroupEmailMessagesService: MessagingDeleteGroupEmailMessagesService,
+    private readonly messagingClearCursorsService: MessagingClearCursorsService,
+    private readonly messageChannelSyncStatusService: MessageChannelSyncStatusService,
+  ) {}
+
+  async markMessageChannelAsPendingGroupEmailsAction(
+    messageChannel: MessageChannelWorkspaceEntity,
+    workspaceId: string,
+    pendingGroupEmailsAction: MessageChannelPendingGroupEmailsAction,
+  ): Promise<void> {
+    const messageChannelRepository =
+      await this.twentyORMManager.getRepository<MessageChannelWorkspaceEntity>(
+        'messageChannel',
+      );
+
+    await messageChannelRepository.update(
+      { id: messageChannel.id },
+      { pendingGroupEmailsAction },
+    );
+
+    this.logger.log(
+      `WorkspaceId: ${workspaceId}, MessageChannelId: ${messageChannel.id} - Marked message channel as pending group emails action: ${pendingGroupEmailsAction}`,
+    );
+  }
+
+  async processGroupEmailActions(
+    messageChannel: MessageChannelWorkspaceEntity,
+    workspaceId: string,
+  ): Promise<void> {
+    const { pendingGroupEmailsAction } = messageChannel;
+
+    if (
+      !isDefined(pendingGroupEmailsAction) ||
+      pendingGroupEmailsAction === MessageChannelPendingGroupEmailsAction.NONE
+    ) {
+      return;
+    }
+
+    this.logger.log(
+      `WorkspaceId: ${workspaceId}, MessageChannelId: ${messageChannel.id} - Processing group email action: ${pendingGroupEmailsAction}`,
+    );
+
+    const workspaceDataSource = await this.twentyORMManager.getDatasource();
+
+    await workspaceDataSource?.transaction(
+      async (transactionManager: WorkspaceEntityManager) => {
+        try {
+          const messageChannelRepository =
+            await this.twentyORMManager.getRepository<MessageChannelWorkspaceEntity>(
+              'messageChannel',
+            );
+
+          switch (pendingGroupEmailsAction) {
+            case MessageChannelPendingGroupEmailsAction.GROUP_EMAILS_DELETION:
+              await this.handleGroupEmailsDeletion(
+                workspaceId,
+                messageChannel.id,
+                transactionManager,
+              );
+              break;
+            case MessageChannelPendingGroupEmailsAction.GROUP_EMAILS_IMPORT:
+              await this.handleGroupEmailsImport(
+                workspaceId,
+                messageChannel.id,
+                transactionManager,
+              );
+              break;
+          }
+
+          await messageChannelRepository.update(
+            { id: messageChannel.id },
+            {
+              pendingGroupEmailsAction:
+                MessageChannelPendingGroupEmailsAction.NONE,
+            },
+            transactionManager,
+          );
+
+          this.logger.log(
+            `WorkspaceId: ${workspaceId}, MessageChannelId: ${messageChannel.id} - Reset pendingGroupEmailsAction to NONE`,
+          );
+        } catch (error) {
+          this.logger.error(
+            `WorkspaceId: ${workspaceId}, MessageChannelId: ${messageChannel.id} - Error processing group email action: ${error.message}`,
+            error.stack,
+          );
+          throw error;
+        }
+      },
+    );
+
+    await this.messageChannelSyncStatusService.scheduleMessageListFetch([
+      messageChannel.id,
+    ]);
+
+    this.logger.log(
+      `WorkspaceId: ${workspaceId}, MessageChannelId: ${messageChannel.id} - Scheduled message list fetch after processing group email action`,
+    );
+  }
+
+  private async handleGroupEmailsDeletion(
+    workspaceId: string,
+    messageChannelId: string,
+    transactionManager: WorkspaceEntityManager,
+  ): Promise<void> {
+    await this.messagingDeleteGroupEmailMessagesService.deleteGroupEmailMessages(
+      workspaceId,
+      messageChannelId,
+    );
+
+    await this.messagingClearCursorsService.clearAllMessageChannelCursors(
+      messageChannelId,
+      transactionManager,
+    );
+
+    this.logger.log(
+      `WorkspaceId: ${workspaceId}, MessageChannelId: ${messageChannelId} - Completed GROUP_EMAILS_DELETION action`,
+    );
+  }
+
+  private async handleGroupEmailsImport(
+    workspaceId: string,
+    messageChannelId: string,
+    transactionManager: WorkspaceEntityManager,
+  ): Promise<void> {
+    await this.messagingClearCursorsService.clearAllMessageChannelCursors(
+      messageChannelId,
+      transactionManager,
+    );
+
+    this.logger.log(
+      `WorkspaceId: ${workspaceId}, MessageChannelId: ${messageChannelId} - Completed GROUP_EMAILS_IMPORT action`,
+    );
+  }
+}
@@ -228,36 +228,6 @@ describe('MessagingSaveMessagesAndEnqueueContactCreationService', () => {
    );
  });

-  it('should not create group emails contacts', async () => {
-    await service.saveMessagesAndEnqueueContactCreation(
-      [
-        {
-          ...mockMessages[0],
-          participants: [
-            {
-              role: 'from',
-              handle: 'contact@group.com',
-              displayName: 'participant that is the Connected Account',
-            },
-          ],
-        },
-      ],
-      mockMessageChannel,
-      mockConnectedAccount,
-      workspaceId,
-    );
-
-    expect(messageQueueService.add).toHaveBeenCalledWith(
-      CreateCompanyAndContactJob.name,
-      {
-        workspaceId,
-        connectedAccount: mockConnectedAccount,
-        source: FieldActorSource.EMAIL,
-        contactsToCreate: [],
-      },
-    );
-  });
-
  it('should not create personal emails contacts', async () => {
    await service.saveMessagesAndEnqueueContactCreation(
      [
@@ -22,7 +22,6 @@ import {
 } from 'src/modules/messaging/message-import-manager/drivers/gmail/types/gmail-message.type';
 import { MessagingMessageService } from 'src/modules/messaging/message-import-manager/services/messaging-message.service';
 import { type MessageWithParticipants } from 'src/modules/messaging/message-import-manager/types/message';
-import { isGroupEmail } from 'src/modules/messaging/message-import-manager/utils/is-group-email';
 import { MessagingMessageParticipantService } from 'src/modules/messaging/message-participant-manager/services/messaging-message-participant.service';
 import { isWorkEmail } from 'src/utils/is-work-email';

@@ -79,15 +78,10 @@ export class MessagingSaveMessagesAndEnqueueContactCreationService {
                  messageChannel.excludeNonProfessionalEmails &&
                  !isWorkEmail(participant.handle);

-                const isExcludedByGroupEmails =
-                  messageChannel.excludeGroupEmails &&
-                  isGroupEmail(participant.handle);
-
                const shouldCreateContact =
                  !!participant.handle &&
                  !isParticipantConnectedAccount &&
                  !isExcludedByNonProfessionalEmails &&
-                  !isExcludedByGroupEmails &&
                  (messageChannel.contactAutoCreationPolicy ===
                    MessageChannelContactAutoCreationPolicy.SENT_AND_RECEIVED ||
                    (messageChannel.contactAutoCreationPolicy ===