Fix messaging 404 handling (#17041)

`MessageImportExceptionHandlerService.handleSyncCursorErrorException()`
was calling `messageChannelSyncStatusService.markAsFailed()`
instead of

`messageChannelSyncStatusService.resetAndMarkAsMessagesListFetchPending()`

Also adds a new command `MessagingTriggerMessageListFetchCommand` for
faster local development speed

packages/twenty-server/src/modules/connected-account/refresh-tokens-manager/services/connected-account-refresh-tokens.service.ts

@@ -66,8 +66,8 @@ export class ConnectedAccountRefreshTokensService {
       };
     }
 
-    this.logger.log(
-      `Access token expired for connected account ${connectedAccount.id.slice(0, 7)} in workspace ${workspaceId.slice(0, 7)}, refreshing...`,
+    this.logger.debug(
+      `Access token expired for connected account ${connectedAccount.id} in workspace ${workspaceId}, refreshing...`,
     );
 
     const connectedAccountTokens = await this.refreshTokens(
@@ -159,13 +159,13 @@ export class ConnectedAccountRefreshTokensService {
     } catch (error) {
       if (isGmailNetworkError(error)) {
         throw new ConnectedAccountRefreshAccessTokenException(
-          `Error refreshing tokens for connected account ${connectedAccount.id.slice(0, 7)} in workspace ${workspaceId.slice(0, 7)}: ${error.code}`,
+          `Error refreshing tokens for connected account ${connectedAccount.id} in workspace ${workspaceId}: ${error.code}`,
           ConnectedAccountRefreshAccessTokenExceptionCode.TEMPORARY_NETWORK_ERROR,
         );
       }
 
       this.logger.log(
-        `Error while refreshing tokens on connected account ${connectedAccount.id.slice(0, 7)} in workspace ${workspaceId.slice(0, 7)}`,
+        `Error while refreshing tokens on connected account ${connectedAccount.id} in workspace ${workspaceId}`,
         error,
       );
       throw error;

packages/twenty-server/src/modules/messaging/message-import-manager/commands/messaging-trigger-message-list-fetch.command.ts

@@ -0,0 +1,131 @@
+import { Logger } from '@nestjs/common';
+
+import { Command, CommandRunner, Option } 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 { GlobalWorkspaceOrmManager } from 'src/engine/twenty-orm/global-workspace-datasource/global-workspace-orm.manager';
+import { buildSystemAuthContext } from 'src/engine/twenty-orm/utils/build-system-auth-context.util';
+import {
+  MessagingMessageListFetchJob,
+  type MessagingMessageListFetchJobData,
+} from 'src/modules/messaging/message-import-manager/jobs/messaging-message-list-fetch.job';
+import {
+  MessageChannelSyncStage,
+  type MessageChannelWorkspaceEntity,
+} from 'src/modules/messaging/common/standard-objects/message-channel.workspace-entity';
+
+type MessagingTriggerMessageListFetchCommandOptions = {
+  workspaceId: string;
+  messageChannelId?: string;
+};
+
+@Command({
+  name: 'messaging:trigger-message-list-fetch',
+  description:
+    'Trigger message list fetch immediately without waiting for cron',
+})
+export class MessagingTriggerMessageListFetchCommand extends CommandRunner {
+  private readonly logger = new Logger(
+    MessagingTriggerMessageListFetchCommand.name,
+  );
+
+  constructor(
+    private readonly globalWorkspaceOrmManager: GlobalWorkspaceOrmManager,
+    @InjectMessageQueue(MessageQueue.messagingQueue)
+    private readonly messageQueueService: MessageQueueService,
+  ) {
+    super();
+  }
+
+  async run(
+    _passedParam: string[],
+    options: MessagingTriggerMessageListFetchCommandOptions,
+  ): Promise<void> {
+    const { workspaceId, messageChannelId } = options;
+
+    this.logger.log(
+      `Triggering message list fetch for workspace ${workspaceId}${messageChannelId ? ` and channel ${messageChannelId}` : ' (all pending channels)'}`,
+    );
+
+    const authContext = buildSystemAuthContext(workspaceId);
+
+    await this.globalWorkspaceOrmManager.executeInWorkspaceContext(
+      authContext,
+      async () => {
+        const messageChannelRepository =
+          await this.globalWorkspaceOrmManager.getRepository<MessageChannelWorkspaceEntity>(
+            workspaceId,
+            'messageChannel',
+          );
+
+        const whereCondition: Record<string, unknown> = {
+          isSyncEnabled: true,
+          syncStage: MessageChannelSyncStage.MESSAGE_LIST_FETCH_PENDING,
+        };
+
+        if (messageChannelId) {
+          whereCondition.id = messageChannelId;
+        }
+
+        const messageChannels =
+          await messageChannelRepository.find(whereCondition);
+
+        if (messageChannels.length === 0) {
+          this.logger.warn(
+            'No message channels found with MESSAGE_LIST_FETCH_PENDING status',
+          );
+
+          return;
+        }
+
+        this.logger.log(
+          `Found ${messageChannels.length} message channel(s) to process`,
+        );
+
+        for (const messageChannel of messageChannels) {
+          await messageChannelRepository.update(messageChannel.id, {
+            syncStage: MessageChannelSyncStage.MESSAGE_LIST_FETCH_SCHEDULED,
+            syncStageStartedAt: new Date().toISOString(),
+          });
+
+          await this.messageQueueService.add<MessagingMessageListFetchJobData>(
+            MessagingMessageListFetchJob.name,
+            {
+              messageChannelId: messageChannel.id,
+              workspaceId,
+            },
+          );
+
+          this.logger.log(
+            `Triggered fetch for message channel ${messageChannel.id}`,
+          );
+        }
+
+        this.logger.log(
+          `Successfully triggered ${messageChannels.length} message list fetch job(s)`,
+        );
+      },
+    );
+  }
+
+  @Option({
+    flags: '-w, --workspace-id <workspace_id>',
+    description: 'Workspace ID',
+    required: true,
+  })
+  parseWorkspaceId(value: string): string {
+    return value;
+  }
+
+  @Option({
+    flags: '-m, --message-channel-id [message_channel_id]',
+    description:
+      'Message Channel ID (optional - if not provided, triggers for all pending channels)',
+    required: false,
+  })
+  parseMessageChannelId(value: string): string {
+    return value;
+  }
+}

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

@@ -15,6 +15,7 @@ import { MessagingCommonModule } from 'src/modules/messaging/common/messaging-co
 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 { MessagingSingleMessageImportCommand } from 'src/modules/messaging/message-import-manager/commands/messaging-single-message-import.command';
+import { MessagingTriggerMessageListFetchCommand } from 'src/modules/messaging/message-import-manager/commands/messaging-trigger-message-list-fetch.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';
@@ -80,6 +81,7 @@ import { MessagingMonitoringModule } from 'src/modules/messaging/monitoring/mess
     MessagingOngoingStaleCronCommand,
     MessagingRelaunchFailedMessageChannelsCronCommand,
     MessagingSingleMessageImportCommand,
+    MessagingTriggerMessageListFetchCommand,
     MessagingMessageListFetchJob,
     MessagingMessagesImportJob,
     MessagingOngoingStaleJob,

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

@@ -107,10 +107,9 @@ export class MessageImportExceptionHandlerService {
     messageChannel: Pick<MessageChannelWorkspaceEntity, 'id'>,
     workspaceId: string,
   ): Promise<void> {
-    await this.messageChannelSyncStatusService.markAsFailed(
+    await this.messageChannelSyncStatusService.resetAndMarkAsMessagesListFetchPending(
       [messageChannel.id],
       workspaceId,
-      MessageChannelSyncStatus.FAILED_UNKNOWN,
     );
   }