From a0b2aa207f0e08183a931d22a4e2b1ffa5cba9ee Mon Sep 17 00:00:00 2001 From: neo773 <62795688+neo773@users.noreply.github.com> Date: Fri, 9 Jan 2026 22:05:12 +0530 Subject: [PATCH] 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 --- ...onnected-account-refresh-tokens.service.ts | 8 +- ...ging-trigger-message-list-fetch.command.ts | 131 ++++++++++++++++++ .../messaging-import-manager.module.ts | 2 + ...saging-import-exception-handler.service.ts | 3 +- 4 files changed, 138 insertions(+), 6 deletions(-) create mode 100644 packages/twenty-server/src/modules/messaging/message-import-manager/commands/messaging-trigger-message-list-fetch.command.ts diff --git a/packages/twenty-server/src/modules/connected-account/refresh-tokens-manager/services/connected-account-refresh-tokens.service.ts b/packages/twenty-server/src/modules/connected-account/refresh-tokens-manager/services/connected-account-refresh-tokens.service.ts index 9b11dd16da..610deaf9cc 100644 --- a/packages/twenty-server/src/modules/connected-account/refresh-tokens-manager/services/connected-account-refresh-tokens.service.ts +++ b/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; diff --git a/packages/twenty-server/src/modules/messaging/message-import-manager/commands/messaging-trigger-message-list-fetch.command.ts b/packages/twenty-server/src/modules/messaging/message-import-manager/commands/messaging-trigger-message-list-fetch.command.ts new file mode 100644 index 0000000000..10cf4ad08d --- /dev/null +++ b/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 { + 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( + workspaceId, + 'messageChannel', + ); + + const whereCondition: Record = { + 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( + 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 ', + 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; + } +} diff --git a/packages/twenty-server/src/modules/messaging/message-import-manager/messaging-import-manager.module.ts b/packages/twenty-server/src/modules/messaging/message-import-manager/messaging-import-manager.module.ts index 51ba0f753e..191da7cacd 100644 --- a/packages/twenty-server/src/modules/messaging/message-import-manager/messaging-import-manager.module.ts +++ b/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, diff --git a/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-import-exception-handler.service.ts b/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-import-exception-handler.service.ts index 9fc65f051d..48881a1734 100644 --- a/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-import-exception-handler.service.ts +++ b/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, workspaceId: string, ): Promise { - await this.messageChannelSyncStatusService.markAsFailed( + await this.messageChannelSyncStatusService.resetAndMarkAsMessagesListFetchPending( [messageChannel.id], workspaceId, - MessageChannelSyncStatus.FAILED_UNKNOWN, ); }