refactor reconnect account logic (#15584)

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

@@ -190,31 +190,35 @@ export class MessageChannelSyncStatusService {
       eventIds: messageChannelIds,
     });
 
-    const connectedAccountRepository =
-      await this.twentyORMManager.getRepository<ConnectedAccountWorkspaceEntity>(
-        'connectedAccount',
+    if (
+      syncStatus === MessageChannelSyncStatus.FAILED_INSUFFICIENT_PERMISSIONS
+    ) {
+      const connectedAccountRepository =
+        await this.twentyORMManager.getRepository<ConnectedAccountWorkspaceEntity>(
+          'connectedAccount',
+        );
+
+      const messageChannels = await messageChannelRepository.find({
+        select: ['id', 'connectedAccountId'],
+        where: { id: Any(messageChannelIds) },
+      });
+
+      const connectedAccountIds = messageChannels.map(
+        (messageChannel) => messageChannel.connectedAccountId,
       );
 
-    const messageChannels = await messageChannelRepository.find({
-      select: ['id', 'connectedAccountId'],
-      where: { id: Any(messageChannelIds) },
-    });
+      await connectedAccountRepository.update(
+        { id: Any(connectedAccountIds) },
+        {
+          authFailedAt: new Date(),
+        },
+      );
 
-    const connectedAccountIds = messageChannels.map(
-      (messageChannel) => messageChannel.connectedAccountId,
-    );
-
-    await connectedAccountRepository.update(
-      { id: Any(connectedAccountIds) },
-      {
-        authFailedAt: new Date(),
-      },
-    );
-
-    await this.addToAccountsToReconnect(
-      messageChannels.map((messageChannel) => messageChannel.id),
-      workspaceId,
-    );
+      await this.addToAccountsToReconnect(
+        messageChannels.map((messageChannel) => messageChannel.id),
+        workspaceId,
+      );
+    }
   }
 
   private async addToAccountsToReconnect(

packages/twenty-server/src/modules/messaging/message-import-manager/commands/messaging-relaunch-failed-message-channels.command.ts

@@ -1,84 +0,0 @@
-import { InjectRepository } from '@nestjs/typeorm';
-
-import { Command } from 'nest-commander';
-import { Repository } from 'typeorm';
-
-import {
-  ActiveOrSuspendedWorkspacesMigrationCommandRunner,
-  type RunOnWorkspaceArgs,
-} from 'src/database/commands/command-runners/active-or-suspended-workspaces-migration.command-runner';
-import { WorkspaceEntity } from 'src/engine/core-modules/workspace/workspace.entity';
-import { TwentyORMGlobalManager } from 'src/engine/twenty-orm/twenty-orm-global.manager';
-import { AccountsToReconnectService } from 'src/modules/connected-account/services/accounts-to-reconnect.service';
-import {
-  MessageChannelSyncStage,
-  MessageChannelSyncStatus,
-  MessageChannelWorkspaceEntity,
-} from 'src/modules/messaging/common/standard-objects/message-channel.workspace-entity';
-
-@Command({
-  name: 'messaging:relaunch-failed-message-channels',
-  description: 'Relaunch failed message channels',
-})
-export class MessagingRelaunchFailedMessageChannelsCommand extends ActiveOrSuspendedWorkspacesMigrationCommandRunner {
-  constructor(
-    @InjectRepository(WorkspaceEntity)
-    protected readonly workspaceRepository: Repository<WorkspaceEntity>,
-    protected readonly twentyORMGlobalManager: TwentyORMGlobalManager,
-    protected readonly accountsToReconnectService: AccountsToReconnectService,
-  ) {
-    super(workspaceRepository, twentyORMGlobalManager);
-  }
-
-  override async runOnWorkspace({
-    workspaceId,
-    options,
-  }: RunOnWorkspaceArgs): Promise<void> {
-    try {
-      const messageChannelRepository =
-        await this.twentyORMGlobalManager.getRepositoryForWorkspace<MessageChannelWorkspaceEntity>(
-          workspaceId,
-          'messageChannel',
-          { shouldBypassPermissionChecks: true },
-        );
-
-      const failedMessageChannels = await messageChannelRepository.find({
-        where: {
-          syncStage: MessageChannelSyncStage.FAILED,
-        },
-        relations: {
-          connectedAccount: {
-            accountOwner: true,
-          },
-        },
-      });
-
-      if (!options.dryRun && failedMessageChannels.length > 0) {
-        await messageChannelRepository.update(
-          failedMessageChannels.map(({ id }) => id),
-          {
-            syncStage: MessageChannelSyncStage.MESSAGE_LIST_FETCH_PENDING,
-            syncStatus: MessageChannelSyncStatus.ACTIVE,
-          },
-        );
-
-        for (const failedMessageChannel of failedMessageChannels) {
-          await this.accountsToReconnectService.removeAccountToReconnect(
-            failedMessageChannel.connectedAccount.accountOwner.userId,
-            failedMessageChannel.connectedAccountId,
-            workspaceId,
-          );
-        }
-      }
-
-      this.logger.log(
-        `${options.dryRun ? ' (DRY RUN) ' : ''}Relaunched ${failedMessageChannels.length} failed message channels`,
-      );
-    } catch (error) {
-      this.logger.error(
-        'Error while relaunching failed message channels',
-        error,
-      );
-    }
-  }
-}

packages/twenty-server/src/modules/messaging/message-import-manager/crons/commands/messaging-relaunch-failed-message-channels.cron.command.ts

@@ -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_RELAUNCH_FAILED_MESSAGE_CHANNELS_CRON_PATTERN,
+  MessagingRelaunchFailedMessageChannelsCronJob,
+} from 'src/modules/messaging/message-import-manager/crons/jobs/messaging-relaunch-failed-message-channels.cron.job';
+
+@Command({
+  name: 'cron:messaging:relaunch-failed-message-channels',
+  description:
+    'Starts a cron job to relaunch failed message channels every 30 minutes',
+})
+export class MessagingRelaunchFailedMessageChannelsCronCommand extends CommandRunner {
+  constructor(
+    @InjectMessageQueue(MessageQueue.cronQueue)
+    private readonly messageQueueService: MessageQueueService,
+  ) {
+    super();
+  }
+
+  async run(): Promise<void> {
+    await this.messageQueueService.addCron<undefined>({
+      jobName: MessagingRelaunchFailedMessageChannelsCronJob.name,
+      data: undefined,
+      options: {
+        repeat: {
+          pattern: MESSAGING_RELAUNCH_FAILED_MESSAGE_CHANNELS_CRON_PATTERN,
+        },
+      },
+    });
+  }
+}

packages/twenty-server/src/modules/messaging/message-import-manager/crons/jobs/messaging-relaunch-failed-message-channels.cron.job.ts

@@ -0,0 +1,74 @@
+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 { MessageChannelSyncStage } from 'src/modules/messaging/common/standard-objects/message-channel.workspace-entity';
+import {
+  MessagingRelaunchFailedMessageChannelJob,
+  type MessagingRelaunchFailedMessageChannelJobData,
+} from 'src/modules/messaging/message-import-manager/jobs/messaging-relaunch-failed-message-channel.job';
+
+export const MESSAGING_RELAUNCH_FAILED_MESSAGE_CHANNELS_CRON_PATTERN =
+  '*/30 * * * *';
+
+@Processor(MessageQueue.cronQueue)
+export class MessagingRelaunchFailedMessageChannelsCronJob {
+  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(MessagingRelaunchFailedMessageChannelsCronJob.name)
+  @SentryCronMonitor(
+    MessagingRelaunchFailedMessageChannelsCronJob.name,
+    MESSAGING_RELAUNCH_FAILED_MESSAGE_CHANNELS_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 failedMessageChannels = await this.coreDataSource.query(
+          `SELECT * FROM ${schemaName}."messageChannel" WHERE "syncStage" = '${MessageChannelSyncStage.FAILED}'`,
+        );
+
+        for (const messageChannel of failedMessageChannels) {
+          await this.messageQueueService.add<MessagingRelaunchFailedMessageChannelJobData>(
+            MessagingRelaunchFailedMessageChannelJob.name,
+            {
+              workspaceId: activeWorkspace.id,
+              messageChannelId: messageChannel.id,
+            },
+          );
+        }
+      } catch (error) {
+        this.exceptionHandlerService.captureExceptions([error], {
+          workspace: {
+            id: activeWorkspace.id,
+          },
+        });
+      }
+    }
+  }
+}

packages/twenty-server/src/modules/messaging/message-import-manager/jobs/messaging-relaunch-failed-message-channel.job.ts

@@ -0,0 +1,61 @@
+import { 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 { TwentyORMGlobalManager } from 'src/engine/twenty-orm/twenty-orm-global.manager';
+import {
+  MessageChannelSyncStage,
+  MessageChannelSyncStatus,
+  MessageChannelWorkspaceEntity,
+} from 'src/modules/messaging/common/standard-objects/message-channel.workspace-entity';
+
+export type MessagingRelaunchFailedMessageChannelJobData = {
+  workspaceId: string;
+  messageChannelId: string;
+};
+
+@Processor({
+  queueName: MessageQueue.messagingQueue,
+  scope: Scope.REQUEST,
+})
+export class MessagingRelaunchFailedMessageChannelJob {
+  constructor(
+    private readonly twentyORMGlobalManager: TwentyORMGlobalManager,
+  ) {}
+
+  @Process(MessagingRelaunchFailedMessageChannelJob.name)
+  async handle(data: MessagingRelaunchFailedMessageChannelJobData) {
+    const { workspaceId, messageChannelId } = data;
+
+    const messageChannelRepository =
+      await this.twentyORMGlobalManager.getRepositoryForWorkspace<MessageChannelWorkspaceEntity>(
+        workspaceId,
+        'messageChannel',
+        { shouldBypassPermissionChecks: true },
+      );
+
+    const messageChannel = await messageChannelRepository.findOne({
+      where: {
+        id: messageChannelId,
+      },
+      relations: {
+        connectedAccount: {
+          accountOwner: true,
+        },
+      },
+    });
+
+    if (
+      !messageChannel ||
+      messageChannel.syncStage !== MessageChannelSyncStage.FAILED
+    ) {
+      return;
+    }
+
+    await messageChannelRepository.update(messageChannelId, {
+      syncStage: MessageChannelSyncStage.MESSAGE_LIST_FETCH_PENDING,
+      syncStatus: MessageChannelSyncStatus.ACTIVE,
+    });
+  }
+}

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

@@ -14,14 +14,15 @@ import { RefreshTokensManagerModule } from 'src/modules/connected-account/refres
 import { MessagingCommonModule } from 'src/modules/messaging/common/messaging-common.module';
 import { MessagingMessageCleanerModule } from 'src/modules/messaging/message-cleaner/messaging-message-cleaner.module';
 import { MessagingFolderSyncManagerModule } from 'src/modules/messaging/message-folder-manager/messaging-folder-sync-manager.module';
-import { MessagingRelaunchFailedMessageChannelsCommand } from 'src/modules/messaging/message-import-manager/commands/messaging-relaunch-failed-message-channels.command';
 import { MessagingSingleMessageImportCommand } from 'src/modules/messaging/message-import-manager/commands/messaging-single-message-import.command';
 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 { 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 { 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';
 import { MessagingMicrosoftDriverModule } from 'src/modules/messaging/message-import-manager/drivers/microsoft/messaging-microsoft-driver.module';
@@ -31,6 +32,7 @@ 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 { 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 { MessagingCursorService } from 'src/modules/messaging/message-import-manager/services/messaging-cursor.service';
@@ -72,14 +74,16 @@ import { MessagingMonitoringModule } from 'src/modules/messaging/monitoring/mess
     MessagingMessageListFetchCronCommand,
     MessagingMessagesImportCronCommand,
     MessagingOngoingStaleCronCommand,
+    MessagingRelaunchFailedMessageChannelsCronCommand,
     MessagingSingleMessageImportCommand,
-    MessagingRelaunchFailedMessageChannelsCommand,
     MessagingMessageListFetchJob,
     MessagingMessagesImportJob,
     MessagingOngoingStaleJob,
+    MessagingRelaunchFailedMessageChannelJob,
     MessagingMessageListFetchCronJob,
     MessagingMessagesImportCronJob,
     MessagingOngoingStaleCronJob,
+    MessagingRelaunchFailedMessageChannelsCronJob,
     MessagingAddSingleMessageToCacheForImportJob,
     MessagingMessageImportManagerMessageChannelListener,
     MessagingCleanCacheJob,
@@ -99,6 +103,7 @@ import { MessagingMonitoringModule } from 'src/modules/messaging/monitoring/mess
     MessagingMessageListFetchCronCommand,
     MessagingMessagesImportCronCommand,
     MessagingOngoingStaleCronCommand,
+    MessagingRelaunchFailedMessageChannelsCronCommand,
   ],
 })
 export class MessagingImportManagerModule {}

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

@@ -98,6 +98,7 @@ export class MessagingAccountAuthenticationService {
           );
         case ConnectedAccountRefreshAccessTokenExceptionCode.REFRESH_ACCESS_TOKEN_FAILED:
         case ConnectedAccountRefreshAccessTokenExceptionCode.REFRESH_TOKEN_NOT_FOUND:
+        case ConnectedAccountRefreshAccessTokenExceptionCode.INVALID_REFRESH_TOKEN:
           await this.messagingMonitoringService.track({
             eventName: `refresh_token.error.insufficient_permissions`,
             workspaceId,