Add message channel reset command (#16266)

In this PR, we are adding a new command to reset a message channel.

We are also refactoring a bit cursor reset as we had multiple
implementations at different places in the code base

---------

Co-authored-by: Charles Bochet <charles@twenty.com>

packages/twenty-server/src/modules/messaging/common/query-hooks/message/message-channel-update-one.pre-query.hook.ts

@@ -1,7 +1,7 @@
 import { Logger } from '@nestjs/common';
 
 import { msg } from '@lingui/core/macro';
-import { assertIsDefinedOrThrow } from 'twenty-shared/utils';
+import { assertIsDefinedOrThrow, isDefined } from 'twenty-shared/utils';
 import { Not } from 'typeorm';
 
 import { type WorkspacePreQueryHookInstance } from 'src/engine/api/graphql/workspace-query-runner/workspace-query-hook/interfaces/workspace-query-hook.interface';
@@ -28,7 +28,6 @@ import { MessagingProcessGroupEmailActionsService } from 'src/modules/messaging/
 
 const ONGOING_SYNC_STAGES = [
   MessageChannelSyncStage.MESSAGE_LIST_FETCH_ONGOING,
-  MessageChannelSyncStage.MESSAGES_IMPORT_ONGOING,
 ];
 
 @WorkspaceQueryHook(`messageChannel.updateOne`)
@@ -63,7 +62,7 @@ export class MessageChannelUpdateOnePreQueryHook
       where: { id: payload.id },
     });
 
-    if (!messageChannel) {
+    if (!isDefined(messageChannel)) {
       throw new WorkspaceQueryRunnerException(
         'Message channel not found',
         WorkspaceQueryRunnerExceptionCode.DATA_NOT_FOUND,
@@ -83,14 +82,15 @@ export class MessageChannelUpdateOnePreQueryHook
         'messageFolder',
       );
 
-    const folderWithPendingAction = await messageFolderRepository.findOne({
-      where: {
-        messageChannelId: messageChannel.id,
-        pendingSyncAction: Not(MessageFolderPendingSyncAction.NONE),
-      },
-    });
+    const messageFoldersWithPendingActionCount =
+      await messageFolderRepository.count({
+        where: {
+          messageChannelId: messageChannel.id,
+          pendingSyncAction: Not(MessageFolderPendingSyncAction.NONE),
+        },
+      });
 
-    const hasPendingFolderActions = !!folderWithPendingAction;
+    const hasPendingFolderActions = messageFoldersWithPendingActionCount > 0;
 
     const hasPendingGroupEmailsAction =
       messageChannel.pendingGroupEmailsAction !==
@@ -122,6 +122,7 @@ export class MessageChannelUpdateOnePreQueryHook
     }
 
     const excludeGroupEmailsChanged =
+      isDefined(payload.data.excludeGroupEmails) &&
       payload.data.excludeGroupEmails !== messageChannel.excludeGroupEmails;
 
     if (excludeGroupEmailsChanged) {

packages/twenty-server/src/modules/messaging/common/query-hooks/messaging-query-hook.module.ts

@@ -1,15 +1,18 @@
 import { Module } from '@nestjs/common';
 
 import { ApplyMessagesVisibilityRestrictionsService } from 'src/modules/messaging/common/query-hooks/message/apply-messages-visibility-restrictions.service';
+import { MessageChannelUpdateOnePreQueryHook } from 'src/modules/messaging/common/query-hooks/message/message-channel-update-one.pre-query.hook';
 import { MessageFindManyPostQueryHook } from 'src/modules/messaging/common/query-hooks/message/message-find-many.post-query.hook';
 import { MessageFindOnePostQueryHook } from 'src/modules/messaging/common/query-hooks/message/message-find-one.post-query.hook';
+import { MessagingImportManagerModule } from 'src/modules/messaging/message-import-manager/messaging-import-manager.module';
 
 @Module({
-  imports: [],
+  imports: [MessagingImportManagerModule],
   providers: [
     ApplyMessagesVisibilityRestrictionsService,
     MessageFindOnePostQueryHook,
     MessageFindManyPostQueryHook,
+    MessageChannelUpdateOnePreQueryHook,
   ],
 })
 export class MessagingQueryHookModule {}

packages/twenty-server/src/modules/messaging/common/services/message-channel-sync-status.service.ts

@@ -1,39 +1,48 @@
 import { Injectable } from '@nestjs/common';
 
-import { Any } from 'typeorm';
+import { Any, In } from 'typeorm';
 
 import { InjectCacheStorage } from 'src/engine/core-modules/cache-storage/decorators/cache-storage.decorator';
 import { CacheStorageService } from 'src/engine/core-modules/cache-storage/services/cache-storage.service';
 import { CacheStorageNamespace } from 'src/engine/core-modules/cache-storage/types/cache-storage-namespace.enum';
 import { MetricsService } from 'src/engine/core-modules/metrics/metrics.service';
 import { MetricsKeys } from 'src/engine/core-modules/metrics/types/metrics-keys.type';
-import { TwentyORMManager } from 'src/engine/twenty-orm/twenty-orm.manager';
+import { TwentyORMGlobalManager } from 'src/engine/twenty-orm/twenty-orm-global.manager';
 import { AccountsToReconnectService } from 'src/modules/connected-account/services/accounts-to-reconnect.service';
 import { type ConnectedAccountWorkspaceEntity } from 'src/modules/connected-account/standard-objects/connected-account.workspace-entity';
 import { AccountsToReconnectKeys } from 'src/modules/connected-account/types/accounts-to-reconnect-key-value.type';
 import {
+  MessageChannelPendingGroupEmailsAction,
   MessageChannelSyncStage,
   MessageChannelSyncStatus,
   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';
 
 @Injectable()
 export class MessageChannelSyncStatusService {
   constructor(
     @InjectCacheStorage(CacheStorageNamespace.ModuleMessaging)
     private readonly cacheStorage: CacheStorageService,
-    private readonly twentyORMManager: TwentyORMManager,
+    private readonly twentyORMGlobalManager: TwentyORMGlobalManager,
     private readonly accountsToReconnectService: AccountsToReconnectService,
     private readonly metricsService: MetricsService,
   ) {}
 
-  public async scheduleMessageListFetch(messageChannelIds: string[]) {
+  public async scheduleMessageListFetch(
+    messageChannelIds: string[],
+    workspaceId: string,
+  ) {
     if (!messageChannelIds.length) {
       return;
     }
 
     const messageChannelRepository =
-      await this.twentyORMManager.getRepository<MessageChannelWorkspaceEntity>(
+      await this.twentyORMGlobalManager.getRepositoryForWorkspace<MessageChannelWorkspaceEntity>(
+        workspaceId,
         'messageChannel',
       );
 
@@ -42,13 +51,17 @@ export class MessageChannelSyncStatusService {
     });
   }
 
-  public async scheduleMessagesImport(messageChannelIds: string[]) {
+  public async scheduleMessagesImport(
+    messageChannelIds: string[],
+    workspaceId: string,
+  ) {
     if (!messageChannelIds.length) {
       return;
     }
 
     const messageChannelRepository =
-      await this.twentyORMManager.getRepository<MessageChannelWorkspaceEntity>(
+      await this.twentyORMGlobalManager.getRepositoryForWorkspace<MessageChannelWorkspaceEntity>(
+        workspaceId,
         'messageChannel',
       );
 
@@ -72,26 +85,46 @@ export class MessageChannelSyncStatusService {
     }
 
     const messageChannelRepository =
-      await this.twentyORMManager.getRepository<MessageChannelWorkspaceEntity>(
+      await this.twentyORMGlobalManager.getRepositoryForWorkspace<MessageChannelWorkspaceEntity>(
+        workspaceId,
         'messageChannel',
       );
 
+    const messageFolderRepository =
+      await this.twentyORMGlobalManager.getRepositoryForWorkspace<MessageFolderWorkspaceEntity>(
+        workspaceId,
+        'messageFolder',
+      );
+
     await messageChannelRepository.update(messageChannelIds, {
       syncCursor: '',
       syncStageStartedAt: null,
       throttleFailureCount: 0,
+      pendingGroupEmailsAction: MessageChannelPendingGroupEmailsAction.NONE,
     });
 
-    await this.scheduleMessageListFetch(messageChannelIds);
+    await messageFolderRepository.update(
+      { messageChannelId: In(messageChannelIds) },
+      {
+        syncCursor: '',
+        pendingSyncAction: MessageFolderPendingSyncAction.NONE,
+      },
+    );
+
+    await this.scheduleMessageListFetch(messageChannelIds, workspaceId);
   }
 
-  public async resetSyncStageStartedAt(messageChannelIds: string[]) {
+  public async resetSyncStageStartedAt(
+    messageChannelIds: string[],
+    workspaceId: string,
+  ) {
     if (!messageChannelIds.length) {
       return;
     }
 
     const messageChannelRepository =
-      await this.twentyORMManager.getRepository<MessageChannelWorkspaceEntity>(
+      await this.twentyORMGlobalManager.getRepositoryForWorkspace<MessageChannelWorkspaceEntity>(
+        workspaceId,
         'messageChannel',
       );
 
@@ -100,13 +133,17 @@ export class MessageChannelSyncStatusService {
     });
   }
 
-  public async markAsMessagesListFetchOngoing(messageChannelIds: string[]) {
+  public async markAsMessagesListFetchOngoing(
+    messageChannelIds: string[],
+    workspaceId: string,
+  ) {
     if (!messageChannelIds.length) {
       return;
     }
 
     const messageChannelRepository =
-      await this.twentyORMManager.getRepository<MessageChannelWorkspaceEntity>(
+      await this.twentyORMGlobalManager.getRepositoryForWorkspace<MessageChannelWorkspaceEntity>(
+        workspaceId,
         'messageChannel',
       );
 
@@ -119,13 +156,15 @@ export class MessageChannelSyncStatusService {
 
   public async markAsCompletedAndScheduleMessageListFetch(
     messageChannelIds: string[],
+    workspaceId: string,
   ) {
     if (!messageChannelIds.length) {
       return;
     }
 
     const messageChannelRepository =
-      await this.twentyORMManager.getRepository<MessageChannelWorkspaceEntity>(
+      await this.twentyORMGlobalManager.getRepositoryForWorkspace<MessageChannelWorkspaceEntity>(
+        workspaceId,
         'messageChannel',
       );
 
@@ -143,13 +182,17 @@ export class MessageChannelSyncStatusService {
     });
   }
 
-  public async markAsMessagesImportOngoing(messageChannelIds: string[]) {
+  public async markAsMessagesImportOngoing(
+    messageChannelIds: string[],
+    workspaceId: string,
+  ) {
     if (!messageChannelIds.length) {
       return;
     }
 
     const messageChannelRepository =
-      await this.twentyORMManager.getRepository<MessageChannelWorkspaceEntity>(
+      await this.twentyORMGlobalManager.getRepositoryForWorkspace<MessageChannelWorkspaceEntity>(
+        workspaceId,
         'messageChannel',
       );
 
@@ -171,7 +214,8 @@ export class MessageChannelSyncStatusService {
     }
 
     const messageChannelRepository =
-      await this.twentyORMManager.getRepository<MessageChannelWorkspaceEntity>(
+      await this.twentyORMGlobalManager.getRepositoryForWorkspace<MessageChannelWorkspaceEntity>(
+        workspaceId,
         'messageChannel',
       );
 
@@ -194,7 +238,8 @@ export class MessageChannelSyncStatusService {
       syncStatus === MessageChannelSyncStatus.FAILED_INSUFFICIENT_PERMISSIONS
     ) {
       const connectedAccountRepository =
-        await this.twentyORMManager.getRepository<ConnectedAccountWorkspaceEntity>(
+        await this.twentyORMGlobalManager.getRepositoryForWorkspace<ConnectedAccountWorkspaceEntity>(
+          workspaceId,
           'connectedAccount',
         );
 
@@ -230,7 +275,8 @@ export class MessageChannelSyncStatusService {
     }
 
     const messageChannelRepository =
-      await this.twentyORMManager.getRepository<MessageChannelWorkspaceEntity>(
+      await this.twentyORMGlobalManager.getRepositoryForWorkspace<MessageChannelWorkspaceEntity>(
+        workspaceId,
         'messageChannel',
       );
 

packages/twenty-server/src/modules/messaging/message-channel-manager/query-hooks/message-channel-query-hook.module.ts

@@ -1,11 +0,0 @@
-import { Module } from '@nestjs/common';
-
-import { MessageChannelUpdateOnePreQueryHook } from 'src/modules/messaging/message-channel-manager/query-hooks/message-channel-update-one.pre-query.hook';
-import { MessagingImportManagerModule } from 'src/modules/messaging/message-import-manager/messaging-import-manager.module';
-
-@Module({
-  imports: [MessagingImportManagerModule],
-  providers: [MessageChannelUpdateOnePreQueryHook],
-  exports: [MessageChannelUpdateOnePreQueryHook],
-})
-export class MessageChannelQueryHookModule {}

packages/twenty-server/src/modules/messaging/message-cleaner/commands/messaging-reset-channel.command.ts

@@ -0,0 +1,97 @@
+import { Logger } from '@nestjs/common';
+
+import { Command, CommandRunner, Option } from 'nest-commander';
+import { isDefined } from 'twenty-shared/utils';
+
+import { TwentyORMGlobalManager } from 'src/engine/twenty-orm/twenty-orm-global.manager';
+import { MessageChannelSyncStatusService } from 'src/modules/messaging/common/services/message-channel-sync-status.service';
+import { type 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';
+
+type MessagingResetChannelCommandOptions = {
+  workspaceId: string;
+  messageChannelId?: string;
+};
+
+@Command({
+  name: 'messaging:reset-channel',
+  description:
+    'Reset message channel(s) for full resync. If no channel ID provided, resets all channels in the workspace.',
+})
+export class MessagingResetChannelCommand extends CommandRunner {
+  private readonly logger = new Logger(MessagingResetChannelCommand.name);
+
+  constructor(
+    private readonly twentyORMGlobalManager: TwentyORMGlobalManager,
+    private readonly messagingChannelSyncStatusService: MessageChannelSyncStatusService,
+    private readonly messagingMessageCleanerService: MessagingMessageCleanerService,
+  ) {
+    super();
+  }
+
+  async run(
+    _passedParam: string[],
+    options: MessagingResetChannelCommandOptions,
+  ): Promise<void> {
+    const { workspaceId, messageChannelId } = options;
+
+    const messageChannelRepository =
+      await this.twentyORMGlobalManager.getRepositoryForWorkspace<MessageChannelWorkspaceEntity>(
+        workspaceId,
+        'messageChannel',
+      );
+
+    this.logger.log(
+      `No message channel ID provided, resetting all message channels in workspace ${workspaceId}`,
+    );
+
+    const messageChannels = await messageChannelRepository.find({
+      where: {
+        ...(isDefined(messageChannelId) ? { id: messageChannelId } : {}),
+      },
+    });
+
+    if (messageChannels.length === 0) {
+      this.logger.log(`No message channels found in workspace ${workspaceId}`);
+
+      return;
+    }
+
+    this.logger.log(
+      `Found ${messageChannels.length} message channels to reset`,
+    );
+
+    for (const messageChannel of messageChannels) {
+      await this.messagingChannelSyncStatusService.resetAndScheduleMessageListFetch(
+        [messageChannel.id],
+        workspaceId,
+      );
+      await this.messagingMessageCleanerService.cleanOrphanMessagesAndThreads(
+        workspaceId,
+      );
+    }
+
+    this.logger.log(
+      `Successfully reset all ${messageChannels.length} message channels in workspace ${workspaceId}`,
+    );
+  }
+
+  @Option({
+    flags: '-w, --workspace-id <workspace_id>',
+    description: 'Workspace ID',
+    required: true,
+  })
+  parseWorkspaceId(value: string): string {
+    return value;
+  }
+
+  @Option({
+    flags: '-c, --message-channel-id [message_channel_id]',
+    description:
+      'Message Channel ID (optional - if not provided, all channels will be reset)',
+    required: false,
+  })
+  parseMessageChannelId(value: string): string {
+    return value;
+  }
+}

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

@@ -3,18 +3,25 @@ import { TypeOrmModule } from '@nestjs/typeorm';
 
 import { WorkspaceEntity } from 'src/engine/core-modules/workspace/workspace.entity';
 import { DataSourceModule } from 'src/engine/metadata-modules/data-source/data-source.module';
+import { MessagingCommonModule } from 'src/modules/messaging/common/messaging-common.module';
 import { MessagingMessageCleanerRemoveOrphansCommand } from 'src/modules/messaging/message-cleaner/commands/messaging-message-clearner-remove-orphans.command';
+import { MessagingResetChannelCommand } from 'src/modules/messaging/message-cleaner/commands/messaging-reset-channel.command';
 import { MessagingConnectedAccountDeletionCleanupJob } from 'src/modules/messaging/message-cleaner/jobs/messaging-connected-account-deletion-cleanup.job';
 import { MessagingMessageCleanerConnectedAccountListener } from 'src/modules/messaging/message-cleaner/listeners/messaging-message-cleaner-connected-account.listener';
 import { MessagingMessageCleanerService } from 'src/modules/messaging/message-cleaner/services/messaging-message-cleaner.service';
 
 @Module({
-  imports: [TypeOrmModule.forFeature([WorkspaceEntity]), DataSourceModule],
+  imports: [
+    TypeOrmModule.forFeature([WorkspaceEntity]),
+    DataSourceModule,
+    MessagingCommonModule,
+  ],
   providers: [
-    MessagingMessageCleanerService,
     MessagingConnectedAccountDeletionCleanupJob,
     MessagingMessageCleanerConnectedAccountListener,
     MessagingMessageCleanerRemoveOrphansCommand,
+    MessagingResetChannelCommand,
+    MessagingMessageCleanerService,
   ],
   exports: [MessagingMessageCleanerService],
 })

packages/twenty-server/src/modules/messaging/message-import-manager/jobs/messaging-ongoing-stale.job.ts

@@ -57,22 +57,25 @@ 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([
-          messageChannel.id,
-        ]);
+        await this.messageChannelSyncStatusService.resetSyncStageStartedAt(
+          [messageChannel.id],
+          workspaceId,
+        );
 
         switch (messageChannel.syncStage) {
           case MessageChannelSyncStage.MESSAGE_LIST_FETCH_ONGOING:
           case MessageChannelSyncStage.MESSAGE_LIST_FETCH_SCHEDULED:
             await this.messageChannelSyncStatusService.scheduleMessageListFetch(
               [messageChannel.id],
+              workspaceId,
             );
             break;
           case MessageChannelSyncStage.MESSAGES_IMPORT_ONGOING:
           case MessageChannelSyncStage.MESSAGES_IMPORT_SCHEDULED:
-            await this.messageChannelSyncStatusService.scheduleMessagesImport([
-              messageChannel.id,
-            ]);
+            await this.messageChannelSyncStatusService.scheduleMessagesImport(
+              [messageChannel.id],
+              workspaceId,
+            );
             break;
           default:
             break;

packages/twenty-server/src/modules/messaging/message-import-manager/messaging-import-manager.module.ts

@@ -35,7 +35,6 @@ import { MessagingOngoingStaleJob } from 'src/modules/messaging/message-import-m
 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';
@@ -70,7 +69,6 @@ import { MessagingMonitoringModule } from 'src/modules/messaging/monitoring/mess
     FeatureFlagModule,
     MessageParticipantManagerModule,
     MessagingFolderSyncManagerModule,
-    MessagingClearCursorsModule,
     MessagingMonitoringModule,
     MessagingMessageCleanerModule,
     WorkspaceEventEmitterModule,

packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-clear-cursors.module.ts

@@ -1,9 +0,0 @@
-import { Module } from '@nestjs/common';
-
-import { MessagingClearCursorsService } from 'src/modules/messaging/message-import-manager/services/messaging-clear-cursors.service';
-
-@Module({
-  providers: [MessagingClearCursorsService],
-  exports: [MessagingClearCursorsService],
-})
-export class MessagingClearCursorsModule {}

packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-clear-cursors.service.spec.ts

@@ -1,62 +0,0 @@
-import { Test, type TestingModule } from '@nestjs/testing';
-
-import { TwentyORMManager } from 'src/engine/twenty-orm/twenty-orm.manager';
-import { MessagingClearCursorsService } from 'src/modules/messaging/message-import-manager/services/messaging-clear-cursors.service';
-
-describe('MessagingClearCursorsService', () => {
-  let service: MessagingClearCursorsService;
-
-  const mockMessageChannelRepository = {
-    update: jest.fn(),
-  };
-
-  const mockMessageFolderRepository = {
-    update: jest.fn(),
-  };
-
-  beforeEach(async () => {
-    const module: TestingModule = await Test.createTestingModule({
-      providers: [
-        MessagingClearCursorsService,
-        {
-          provide: TwentyORMManager,
-          useValue: {
-            getRepository: jest.fn((entityName) => {
-              if (entityName === 'messageChannel') {
-                return mockMessageChannelRepository;
-              }
-              if (entityName === 'messageFolder') {
-                return mockMessageFolderRepository;
-              }
-            }),
-          },
-        },
-      ],
-    }).compile();
-
-    service = module.get(MessagingClearCursorsService);
-  });
-
-  afterEach(() => {
-    jest.clearAllMocks();
-  });
-
-  describe('clearAllMessageChannelCursors', () => {
-    const messageChannelId = 'test-channel-id';
-
-    it('should clear message channel and folder cursors', async () => {
-      await service.clearAllMessageChannelCursors(messageChannelId);
-
-      expect(mockMessageChannelRepository.update).toHaveBeenCalledWith(
-        { id: messageChannelId },
-        { syncCursor: '' },
-        undefined,
-      );
-      expect(mockMessageFolderRepository.update).toHaveBeenCalledWith(
-        { messageChannelId },
-        { syncCursor: '' },
-        undefined,
-      );
-    });
-  });
-});

packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-clear-cursors.service.ts

@@ -1,48 +0,0 @@
-import { Injectable, Logger } from '@nestjs/common';
-
-import { type WorkspaceEntityManager } from 'src/engine/twenty-orm/entity-manager/workspace-entity-manager';
-import { TwentyORMManager } from 'src/engine/twenty-orm/twenty-orm.manager';
-import { 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';
-
-@Injectable()
-export class MessagingClearCursorsService {
-  private readonly logger = new Logger(MessagingClearCursorsService.name);
-
-  constructor(private readonly twentyORMManager: TwentyORMManager) {}
-
-  async clearAllMessageChannelCursors(
-    messageChannelId: string,
-    transactionManager?: WorkspaceEntityManager,
-  ): Promise<void> {
-    this.logger.log(
-      `MessageChannelId: ${messageChannelId} - Clearing all sync cursors`,
-    );
-
-    const messageChannelRepository =
-      await this.twentyORMManager.getRepository<MessageChannelWorkspaceEntity>(
-        'messageChannel',
-      );
-
-    const messageFolderRepository =
-      await this.twentyORMManager.getRepository<MessageFolderWorkspaceEntity>(
-        'messageFolder',
-      );
-
-    await messageChannelRepository.update(
-      { id: messageChannelId },
-      { syncCursor: '' },
-      transactionManager,
-    );
-
-    await messageFolderRepository.update(
-      { messageChannelId },
-      { syncCursor: '' },
-      transactionManager,
-    );
-
-    this.logger.log(
-      `MessageChannelId: ${messageChannelId} - Cleared all sync cursors`,
-    );
-  }
-}

packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-import-exception-handler.service.ts

@@ -17,7 +17,6 @@ import {
   MessageImportDriverExceptionCode,
 } from 'src/modules/messaging/message-import-manager/drivers/exceptions/message-import-driver.exception';
 import { MessageNetworkExceptionCode } from 'src/modules/messaging/message-import-manager/drivers/exceptions/message-network.exception';
-import { MessagingClearCursorsService } from 'src/modules/messaging/message-import-manager/services/messaging-clear-cursors.service';
 
 export enum MessageImportSyncStep {
   MESSAGE_LIST_FETCH = 'MESSAGE_LIST_FETCH',
@@ -30,7 +29,6 @@ export class MessageImportExceptionHandlerService {
   constructor(
     private readonly twentyORMManager: TwentyORMManager,
     private readonly messageChannelSyncStatusService: MessageChannelSyncStatusService,
-    private readonly messagingClearCursorsService: MessagingClearCursorsService,
     private readonly exceptionHandlerService: ExceptionHandlerService,
   ) {}
 
@@ -113,10 +111,6 @@ export class MessageImportExceptionHandlerService {
       workspaceId,
       MessageChannelSyncStatus.FAILED_UNKNOWN,
     );
-
-    await this.messagingClearCursorsService.clearAllMessageChannelCursors(
-      messageChannel.id,
-    );
   }
 
   private async handleTemporaryException(
@@ -169,16 +163,18 @@ export class MessageImportExceptionHandlerService {
 
     switch (syncStep) {
       case MessageImportSyncStep.MESSAGE_LIST_FETCH:
-        await this.messageChannelSyncStatusService.scheduleMessageListFetch([
-          messageChannel.id,
-        ]);
+        await this.messageChannelSyncStatusService.scheduleMessageListFetch(
+          [messageChannel.id],
+          workspaceId,
+        );
         break;
 
       case MessageImportSyncStep.MESSAGES_IMPORT_PENDING:
       case MessageImportSyncStep.MESSAGES_IMPORT_ONGOING:
-        await this.messageChannelSyncStatusService.scheduleMessagesImport([
-          messageChannel.id,
-        ]);
+        await this.messageChannelSyncStatusService.scheduleMessagesImport(
+          [messageChannel.id],
+          workspaceId,
+        );
         break;
 
       default:

packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-message-list-fetch.service.ts

@@ -67,6 +67,7 @@ export class MessagingMessageListFetchService {
 
       await this.messageChannelSyncStatusService.markAsMessagesListFetchOngoing(
         [messageChannel.id],
+        workspaceId,
       );
 
       this.logger.log(
@@ -270,6 +271,7 @@ export class MessagingMessageListFetchService {
       if (totalMessagesToImportCount === 0) {
         await this.messageChannelSyncStatusService.markAsCompletedAndScheduleMessageListFetch(
           [messageChannelWithFreshTokens.id],
+          workspaceId,
         );
 
         return;
@@ -279,9 +281,10 @@ export class MessagingMessageListFetchService {
         `messageChannelId: ${freshMessageChannel.id} Scheduling direct messages import`,
       );
 
-      await this.messageChannelSyncStatusService.scheduleMessagesImport([
-        messageChannelWithFreshTokens.id,
-      ]);
+      await this.messageChannelSyncStatusService.scheduleMessagesImport(
+        [messageChannelWithFreshTokens.id],
+        workspaceId,
+      );
 
       await this.messagingMessagesImportService.processMessageBatchImport(
         {

packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-messages-import.service.ts

@@ -71,9 +71,10 @@ export class MessagingMessagesImportService {
         messageChannelId: messageChannel.id,
       });
 
-      await this.messageChannelSyncStatusService.markAsMessagesImportOngoing([
-        messageChannel.id,
-      ]);
+      await this.messageChannelSyncStatusService.markAsMessagesImportOngoing(
+        [messageChannel.id],
+        workspaceId,
+      );
 
       const { accessToken, refreshToken } =
         await this.messagingAccountAuthenticationService.validateAndRefreshConnectedAccountAuthentication(
@@ -102,6 +103,7 @@ export class MessagingMessagesImportService {
       if (!messageIdsToFetch?.length) {
         await this.messageChannelSyncStatusService.markAsCompletedAndScheduleMessageListFetch(
           [messageChannel.id],
+          workspaceId,
         );
 
         return await this.trackMessageImportCompleted(
@@ -158,11 +160,13 @@ export class MessagingMessagesImportService {
       ) {
         await this.messageChannelSyncStatusService.markAsCompletedAndScheduleMessageListFetch(
           [messageChannel.id],
+          workspaceId,
         );
       } else {
-        await this.messageChannelSyncStatusService.scheduleMessagesImport([
-          messageChannel.id,
-        ]);
+        await this.messageChannelSyncStatusService.scheduleMessagesImport(
+          [messageChannel.id],
+          workspaceId,
+        );
       }
 
       const messageChannelRepository =

packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-process-group-email-actions.service.ts

@@ -8,7 +8,7 @@ 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 { MessageFolderWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message-folder.workspace-entity';
 import { MessagingDeleteGroupEmailMessagesService } from 'src/modules/messaging/message-import-manager/services/messaging-delete-group-email-messages.service';
 
 @Injectable()
@@ -20,7 +20,6 @@ export class MessagingProcessGroupEmailActionsService {
   constructor(
     private readonly twentyORMManager: TwentyORMManager,
     private readonly messagingDeleteGroupEmailMessagesService: MessagingDeleteGroupEmailMessagesService,
-    private readonly messagingClearCursorsService: MessagingClearCursorsService,
   ) {}
 
   async markMessageChannelAsPendingGroupEmailsAction(
@@ -120,10 +119,10 @@ export class MessagingProcessGroupEmailActionsService {
       messageChannelId,
     );
 
-    await this.messagingClearCursorsService.clearAllMessageChannelCursors(
+    await this.resetCursors({
       messageChannelId,
       transactionManager,
-    );
+    });
 
     this.logger.log(
       `WorkspaceId: ${workspaceId}, MessageChannelId: ${messageChannelId} - Completed GROUP_EMAILS_DELETION action`,
@@ -135,13 +134,45 @@ export class MessagingProcessGroupEmailActionsService {
     messageChannelId: string,
     transactionManager: WorkspaceEntityManager,
   ): Promise<void> {
-    await this.messagingClearCursorsService.clearAllMessageChannelCursors(
+    await this.resetCursors({
       messageChannelId,
       transactionManager,
-    );
+    });
 
     this.logger.log(
       `WorkspaceId: ${workspaceId}, MessageChannelId: ${messageChannelId} - Completed GROUP_EMAILS_IMPORT action`,
     );
   }
+
+  private async resetCursors({
+    messageChannelId,
+    transactionManager,
+  }: {
+    messageChannelId: string;
+    transactionManager: WorkspaceEntityManager;
+  }) {
+    const messageChannelRepository =
+      await this.twentyORMManager.getRepository<MessageChannelWorkspaceEntity>(
+        'messageChannel',
+      );
+
+    await messageChannelRepository.update(
+      messageChannelId,
+      {
+        syncCursor: '',
+      },
+      transactionManager,
+    );
+
+    const messageFolderRepository =
+      await this.twentyORMManager.getRepository<MessageFolderWorkspaceEntity>(
+        'messageFolder',
+      );
+
+    await messageFolderRepository.update(
+      { messageChannelId },
+      { syncCursor: '' },
+      transactionManager,
+    );
+  }
 }

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

@@ -1,7 +1,6 @@
 import { Module } from '@nestjs/common';
 
 import { MessagingBlocklistManagerModule } from 'src/modules/messaging/blocklist-manager/messaging-blocklist-manager.module';
-import { MessageChannelQueryHookModule } from 'src/modules/messaging/message-channel-manager/query-hooks/message-channel-query-hook.module';
 import { MessagingMessageCleanerModule } from 'src/modules/messaging/message-cleaner/messaging-message-cleaner.module';
 import { MessageFolderQueryHookModule } from 'src/modules/messaging/message-folder-manager/query-hooks/message-folder-query-hook.module';
 import { MessagingImportManagerModule } from 'src/modules/messaging/message-import-manager/messaging-import-manager.module';
@@ -15,7 +14,6 @@ import { MessagingMonitoringModule } from 'src/modules/messaging/monitoring/mess
     MessageParticipantManagerModule,
     MessagingBlocklistManagerModule,
     MessagingMonitoringModule,
-    MessageChannelQueryHookModule,
     MessageFolderQueryHookModule,
   ],
   providers: [],