From 4baabd80ef94af615f08eb5e4ccc7c8d82900611 Mon Sep 17 00:00:00 2001 From: Charles Bochet Date: Thu, 7 Aug 2025 20:21:18 +0200 Subject: [PATCH] Improve messaging behavior (#13746) This PR contains multiple small enhancements --- ...google-api-refresh-access-token.service.ts | 6 +- .../microsoft-api-refresh-tokens.service.ts | 3 +- ...onnected-account-refresh-tokens.service.ts | 41 +++--- .../jobs/messaging-message-list-fetch.job.ts | 5 - ...essaging-account-authentication.service.ts | 117 +++++++----------- ...ssaging-message-list-fetch.service.spec.ts | 92 +++++++++++++- .../messaging-message-list-fetch.service.ts | 53 +++++--- .../services/messaging-message.service.ts | 14 ++- .../messaging-messages-import.service.spec.ts | 21 +++- .../messaging-messages-import.service.ts | 29 +++-- ...d-enqueue-contact-creation.service.spec.ts | 1 + ...es-and-enqueue-contact-creation.service.ts | 1 + 12 files changed, 236 insertions(+), 147 deletions(-) diff --git a/packages/twenty-server/src/modules/connected-account/refresh-tokens-manager/drivers/google/services/google-api-refresh-access-token.service.ts b/packages/twenty-server/src/modules/connected-account/refresh-tokens-manager/drivers/google/services/google-api-refresh-access-token.service.ts index 2843ce491a..1698a2e85f 100644 --- a/packages/twenty-server/src/modules/connected-account/refresh-tokens-manager/drivers/google/services/google-api-refresh-access-token.service.ts +++ b/packages/twenty-server/src/modules/connected-account/refresh-tokens-manager/drivers/google/services/google-api-refresh-access-token.service.ts @@ -4,6 +4,7 @@ import axios from 'axios'; import { z } from 'zod'; import { TwentyConfigService } from 'src/engine/core-modules/twenty-config/twenty-config.service'; +import { ConnectedAccountTokens } from 'src/modules/connected-account/refresh-tokens-manager/services/connected-account-refresh-tokens.service'; export type GoogleTokens = { accessToken: string; @@ -20,7 +21,9 @@ interface GoogleRefreshTokenResponse { export class GoogleAPIRefreshAccessTokenService { constructor(private readonly twentyConfigService: TwentyConfigService) {} - async refreshAccessToken(refreshToken: string): Promise { + async refreshAccessToken( + refreshToken: string, + ): Promise { const response = await axios.post( 'https://oauth2.googleapis.com/token', { @@ -42,6 +45,7 @@ export class GoogleAPIRefreshAccessTokenService { return { accessToken: response.data.access_token, + refreshToken, }; } } diff --git a/packages/twenty-server/src/modules/connected-account/refresh-tokens-manager/drivers/microsoft/services/microsoft-api-refresh-tokens.service.ts b/packages/twenty-server/src/modules/connected-account/refresh-tokens-manager/drivers/microsoft/services/microsoft-api-refresh-tokens.service.ts index 64054b23b0..e7f72b937a 100644 --- a/packages/twenty-server/src/modules/connected-account/refresh-tokens-manager/drivers/microsoft/services/microsoft-api-refresh-tokens.service.ts +++ b/packages/twenty-server/src/modules/connected-account/refresh-tokens-manager/drivers/microsoft/services/microsoft-api-refresh-tokens.service.ts @@ -4,6 +4,7 @@ import axios from 'axios'; import { z } from 'zod'; import { TwentyConfigService } from 'src/engine/core-modules/twenty-config/twenty-config.service'; +import { ConnectedAccountTokens } from 'src/modules/connected-account/refresh-tokens-manager/services/connected-account-refresh-tokens.service'; export type MicrosoftTokens = { accessToken: string; @@ -22,7 +23,7 @@ interface MicrosoftRefreshTokenResponse { export class MicrosoftAPIRefreshAccessTokenService { constructor(private readonly twentyConfigService: TwentyConfigService) {} - async refreshTokens(refreshToken: string): Promise { + async refreshTokens(refreshToken: string): Promise { const response = await axios.post( 'https://login.microsoftonline.com/common/oauth2/v2.0/token', new URLSearchParams({ 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 a7098776a3..38b5e173af 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 @@ -4,14 +4,8 @@ import { ConnectedAccountProvider } from 'twenty-shared/types'; import { assertUnreachable } from 'twenty-shared/utils'; import { TwentyORMManager } from 'src/engine/twenty-orm/twenty-orm.manager'; -import { - GoogleAPIRefreshAccessTokenService, - GoogleTokens, -} from 'src/modules/connected-account/refresh-tokens-manager/drivers/google/services/google-api-refresh-access-token.service'; -import { - MicrosoftAPIRefreshAccessTokenService, - MicrosoftTokens, -} from 'src/modules/connected-account/refresh-tokens-manager/drivers/microsoft/services/microsoft-api-refresh-tokens.service'; +import { GoogleAPIRefreshAccessTokenService } from 'src/modules/connected-account/refresh-tokens-manager/drivers/google/services/google-api-refresh-access-token.service'; +import { MicrosoftAPIRefreshAccessTokenService } from 'src/modules/connected-account/refresh-tokens-manager/drivers/microsoft/services/microsoft-api-refresh-tokens.service'; import { ConnectedAccountRefreshAccessTokenException, ConnectedAccountRefreshAccessTokenExceptionCode, @@ -19,7 +13,10 @@ import { import { ConnectedAccountWorkspaceEntity } from 'src/modules/connected-account/standard-objects/connected-account.workspace-entity'; import { isAxiosTemporaryError } from 'src/modules/messaging/message-import-manager/drivers/gmail/utils/is-axios-gaxios-error.util'; -export type ConnectedAccountTokens = GoogleTokens | MicrosoftTokens; +export type ConnectedAccountTokens = { + accessToken: string; + refreshToken: string; +}; @Injectable() export class ConnectedAccountRefreshTokensService { @@ -36,7 +33,7 @@ export class ConnectedAccountRefreshTokensService { async refreshAndSaveTokens( connectedAccount: ConnectedAccountWorkspaceEntity, workspaceId: string, - ): Promise { + ): Promise { const refreshToken = connectedAccount.refreshToken; if (!refreshToken) { @@ -52,23 +49,17 @@ export class ConnectedAccountRefreshTokensService { workspaceId, ); - try { - const connectedAccountRepository = - await this.twentyORMManager.getRepository( - 'connectedAccount', - ); - - await connectedAccountRepository.update( - { id: connectedAccount.id }, - connectedAccountTokens, + const connectedAccountRepository = + await this.twentyORMManager.getRepository( + 'connectedAccount', ); - } catch (error) { - throw new Error( - `Error saving the new tokens for connected account ${connectedAccount.id} in workspace ${workspaceId}: ${error.message} `, - ); - } - return connectedAccountTokens.accessToken; + await connectedAccountRepository.update( + { id: connectedAccount.id }, + connectedAccountTokens, + ); + + return connectedAccountTokens; } async refreshTokens( diff --git a/packages/twenty-server/src/modules/messaging/message-import-manager/jobs/messaging-message-list-fetch.job.ts b/packages/twenty-server/src/modules/messaging/message-import-manager/jobs/messaging-message-list-fetch.job.ts index 155c462f85..cf214276cc 100644 --- a/packages/twenty-server/src/modules/messaging/message-import-manager/jobs/messaging-message-list-fetch.job.ts +++ b/packages/twenty-server/src/modules/messaging/message-import-manager/jobs/messaging-message-list-fetch.job.ts @@ -79,11 +79,6 @@ export class MessagingMessageListFetchJob { return; } - await this.messagingAccountAuthenticationService.validateAndPrepareAuthentication( - messageChannel, - workspaceId, - ); - switch (messageChannel.syncStage) { case MessageChannelSyncStage.MESSAGE_LIST_FETCH_PENDING: case MessageChannelSyncStage.PARTIAL_MESSAGE_LIST_FETCH_PENDING: // DEPRECATED diff --git a/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-account-authentication.service.ts b/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-account-authentication.service.ts index 6d01defd48..ae146062fb 100644 --- a/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-account-authentication.service.ts +++ b/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-account-authentication.service.ts @@ -4,15 +4,23 @@ import { isDefined } from 'class-validator'; import { ConnectedAccountProvider } from 'twenty-shared/types'; import { ConnectedAccountRefreshAccessTokenExceptionCode } from 'src/modules/connected-account/refresh-tokens-manager/exceptions/connected-account-refresh-tokens.exception'; -import { ConnectedAccountRefreshTokensService } from 'src/modules/connected-account/refresh-tokens-manager/services/connected-account-refresh-tokens.service'; +import { + ConnectedAccountRefreshTokensService, + ConnectedAccountTokens, +} from 'src/modules/connected-account/refresh-tokens-manager/services/connected-account-refresh-tokens.service'; import { ConnectedAccountWorkspaceEntity } from 'src/modules/connected-account/standard-objects/connected-account.workspace-entity'; -import { MessageChannelWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message-channel.workspace-entity'; import { MessageImportDriverException, MessageImportDriverExceptionCode, } from 'src/modules/messaging/message-import-manager/drivers/exceptions/message-import-driver.exception'; import { MessagingMonitoringService } from 'src/modules/messaging/monitoring/services/messaging-monitoring.service'; +interface ValidateAndRefreshConnectedAccountAuthenticationParams { + connectedAccount: ConnectedAccountWorkspaceEntity; + workspaceId: string; + messageChannelId: string; +} + @Injectable() export class MessagingAccountAuthenticationService { constructor( @@ -20,59 +28,43 @@ export class MessagingAccountAuthenticationService { private readonly messagingMonitoringService: MessagingMonitoringService, ) {} - async validateAndPrepareAuthentication( - messageChannel: MessageChannelWorkspaceEntity, - workspaceId: string, - ): Promise { - if ( - messageChannel.connectedAccount.provider === - ConnectedAccountProvider.IMAP_SMTP_CALDAV - ) { - await this.validateImapCredentials(messageChannel, workspaceId); - - return; - } - - await this.refreshAccessTokenForNonImapProvider( - messageChannel.connectedAccount, - workspaceId, - messageChannel.id, - messageChannel.connectedAccountId, - ); - } - - async validateConnectedAccountAuthentication( - connectedAccount: ConnectedAccountWorkspaceEntity, - workspaceId: string, - messageChannelId: string, - ): Promise { + async validateAndRefreshConnectedAccountAuthentication({ + connectedAccount, + workspaceId, + messageChannelId, + }: ValidateAndRefreshConnectedAccountAuthenticationParams): Promise { if ( connectedAccount.provider === ConnectedAccountProvider.IMAP_SMTP_CALDAV && isDefined(connectedAccount.connectionParameters?.IMAP) ) { - await this.validateImapCredentialsForConnectedAccount( + await this.validateImapCredentialsForConnectedAccount({ connectedAccount, workspaceId, messageChannelId, - ); + }); - return; + return { + accessToken: '', + refreshToken: '', + }; } - await this.refreshAccessTokenForNonImapProvider( + return await this.refreshAccessTokenForNonImapProvider({ connectedAccount, workspaceId, messageChannelId, - connectedAccount.id, - ); + }); } - private async validateImapCredentialsForConnectedAccount( - connectedAccount: ConnectedAccountWorkspaceEntity, - workspaceId: string, - messageChannelId: string, - ): Promise { - if (!connectedAccount.connectionParameters) { + private async validateImapCredentialsForConnectedAccount({ + connectedAccount, + workspaceId, + messageChannelId, + }: ValidateAndRefreshConnectedAccountAuthenticationParams): Promise { + if ( + !isDefined(connectedAccount.connectionParameters) || + !isDefined(connectedAccount.connectionParameters?.IMAP) + ) { await this.messagingMonitoringService.track({ eventName: 'messages_import.error.missing_imap_credentials', workspaceId, @@ -87,41 +79,16 @@ export class MessagingAccountAuthenticationService { } } - private async validateImapCredentials( - messageChannel: MessageChannelWorkspaceEntity, - workspaceId: string, - ): Promise { - if ( - !isDefined(messageChannel.connectedAccount.connectionParameters?.IMAP) - ) { - await this.messagingMonitoringService.track({ - eventName: 'message_list_fetch_job.error.missing_imap_credentials', - workspaceId, - connectedAccountId: messageChannel.connectedAccount.id, - messageChannelId: messageChannel.id, - }); - - throw { - code: MessageImportDriverExceptionCode.INSUFFICIENT_PERMISSIONS, - message: 'Missing IMAP credentials in connectionParameters', - }; - } - } - - private async refreshAccessTokenForNonImapProvider( - connectedAccount: ConnectedAccountWorkspaceEntity, - workspaceId: string, - messageChannelId: string, - connectedAccountId: string, - ): Promise { + private async refreshAccessTokenForNonImapProvider({ + connectedAccount, + workspaceId, + messageChannelId, + }: ValidateAndRefreshConnectedAccountAuthenticationParams): Promise { try { - const accessToken = - await this.connectedAccountRefreshTokensService.refreshAndSaveTokens( - connectedAccount, - workspaceId, - ); - - return accessToken; + return await this.connectedAccountRefreshTokensService.refreshAndSaveTokens( + connectedAccount, + workspaceId, + ); } catch (error) { switch (error.code) { case ConnectedAccountRefreshAccessTokenExceptionCode.TEMPORARY_NETWORK_ERROR: @@ -134,7 +101,7 @@ export class MessagingAccountAuthenticationService { await this.messagingMonitoringService.track({ eventName: `refresh_token.error.insufficient_permissions`, workspaceId, - connectedAccountId, + connectedAccountId: connectedAccount.id, messageChannelId, message: `${error.code}: ${error.reason ?? ''}`, }); diff --git a/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-message-list-fetch.service.spec.ts b/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-message-list-fetch.service.spec.ts index 3d78cc640a..d52e87c671 100644 --- a/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-message-list-fetch.service.spec.ts +++ b/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-message-list-fetch.service.spec.ts @@ -9,6 +9,7 @@ import { MessageChannelSyncStatusService } from 'src/modules/messaging/common/se 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'; import { MessagingMessageCleanerService } from 'src/modules/messaging/message-cleaner/services/messaging-message-cleaner.service'; +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'; import { MessagingGetMessageListService } from 'src/modules/messaging/message-import-manager/services/messaging-get-message-list.service'; import { MessageImportExceptionHandlerService } from 'src/modules/messaging/message-import-manager/services/messaging-import-exception-handler.service'; @@ -18,6 +19,7 @@ import { MessagingMessagesImportService } from 'src/modules/messaging/message-im describe('MessagingMessageListFetchService', () => { let messagingMessageListFetchService: MessagingMessageListFetchService; let messagingGetMessageListService: MessagingGetMessageListService; + let messagingAccountAuthenticationService: MessagingAccountAuthenticationService; let messageChannelSyncStatusService: MessageChannelSyncStatusService; let twentyORMManager: TwentyORMManager; let messagingCursorService: MessagingCursorService; @@ -34,7 +36,8 @@ describe('MessagingMessageListFetchService', () => { id: 'microsoft-connected-account-id', provider: ConnectedAccountProvider.MICROSOFT, handle: 'test@microsoft.com', - refreshToken: 'refresh-token', + accessToken: 'old-microsoft-access-token', + refreshToken: 'microsoft-refresh-token', handleAliases: '', }, messageFolders: [ @@ -53,6 +56,7 @@ describe('MessagingMessageListFetchService', () => { id: 'google-connected-account-id', provider: ConnectedAccountProvider.GOOGLE, handle: 'test@gmail.com', + accessToken: 'old-google-access-token', refreshToken: 'google-refresh-token', handleAliases: '', }, @@ -127,6 +131,38 @@ describe('MessagingMessageListFetchService', () => { processMessageBatchImport: jest.fn().mockResolvedValue(undefined), }, }, + { + provide: MessagingAccountAuthenticationService, + useValue: { + validateAndRefreshConnectedAccountAuthentication: jest + .fn() + .mockImplementation(({ connectedAccount }) => { + if ( + connectedAccount.provider === ConnectedAccountProvider.GOOGLE + ) { + return { + accessToken: 'new-google-access-token', + refreshToken: 'new-google-refresh-token', + }; + } + + if ( + connectedAccount.provider === + ConnectedAccountProvider.MICROSOFT + ) { + return { + accessToken: 'new-microsoft-access-token', + refreshToken: 'new-microsoft-refresh-token', + }; + } + + return { + accessToken: '', + refreshToken: '', + }; + }), + }, + }, { provide: MessageChannelSyncStatusService, useValue: { @@ -178,6 +214,10 @@ describe('MessagingMessageListFetchService', () => { module.get( MessagingMessageListFetchService, ); + messagingAccountAuthenticationService = + module.get( + MessagingAccountAuthenticationService, + ); messagingGetMessageListService = module.get( MessagingGetMessageListService, @@ -198,12 +238,26 @@ describe('MessagingMessageListFetchService', () => { workspaceId, ); + expect( + messagingAccountAuthenticationService.validateAndRefreshConnectedAccountAuthentication, + ).toHaveBeenCalledWith({ + connectedAccount: mockMicrosoftMessageChannel.connectedAccount, + workspaceId, + messageChannelId: mockMicrosoftMessageChannel.id, + }); expect( messageChannelSyncStatusService.markAsMessagesListFetchOngoing, ).toHaveBeenCalledWith([mockMicrosoftMessageChannel.id]); expect(messagingGetMessageListService.getMessageLists).toHaveBeenCalledWith( - mockMicrosoftMessageChannel, + { + ...mockMicrosoftMessageChannel, + connectedAccount: { + ...mockMicrosoftMessageChannel.connectedAccount, + accessToken: 'new-microsoft-access-token', + refreshToken: 'new-microsoft-refresh-token', + }, + }, ); expect(twentyORMManager.getRepository).toHaveBeenCalledWith( @@ -211,7 +265,14 @@ describe('MessagingMessageListFetchService', () => { ); expect(messagingCursorService.updateCursor).toHaveBeenCalledWith( - mockMicrosoftMessageChannel, + { + ...mockMicrosoftMessageChannel, + connectedAccount: { + ...mockMicrosoftMessageChannel.connectedAccount, + accessToken: 'new-microsoft-access-token', + refreshToken: 'new-microsoft-refresh-token', + }, + }, 'new-sync-cursor', 'inbox-folder-id', ); @@ -227,12 +288,26 @@ describe('MessagingMessageListFetchService', () => { workspaceId, ); + expect( + messagingAccountAuthenticationService.validateAndRefreshConnectedAccountAuthentication, + ).toHaveBeenCalledWith({ + connectedAccount: mockGoogleMessageChannel.connectedAccount, + workspaceId, + messageChannelId: mockGoogleMessageChannel.id, + }); expect( messageChannelSyncStatusService.markAsMessagesListFetchOngoing, ).toHaveBeenCalledWith([mockGoogleMessageChannel.id]); expect(messagingGetMessageListService.getMessageLists).toHaveBeenCalledWith( - mockGoogleMessageChannel, + { + ...mockGoogleMessageChannel, + connectedAccount: { + ...mockGoogleMessageChannel.connectedAccount, + accessToken: 'new-google-access-token', + refreshToken: 'new-google-refresh-token', + }, + }, ); expect(twentyORMManager.getRepository).toHaveBeenCalledWith( @@ -240,7 +315,14 @@ describe('MessagingMessageListFetchService', () => { ); expect(messagingCursorService.updateCursor).toHaveBeenCalledWith( - mockGoogleMessageChannel, + { + ...mockGoogleMessageChannel, + connectedAccount: { + ...mockGoogleMessageChannel.connectedAccount, + accessToken: 'new-google-access-token', + refreshToken: 'new-google-refresh-token', + }, + }, 'new-google-history-id', undefined, ); diff --git a/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-message-list-fetch.service.ts b/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-message-list-fetch.service.ts index 68a4223411..90ea00e602 100644 --- a/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-message-list-fetch.service.ts +++ b/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-message-list-fetch.service.ts @@ -13,6 +13,7 @@ import { 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'; +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'; import { MessagingGetMessageListService } from 'src/modules/messaging/message-import-manager/services/messaging-get-message-list.service'; import { @@ -21,7 +22,6 @@ import { } from 'src/modules/messaging/message-import-manager/services/messaging-import-exception-handler.service'; import { MessagingMessagesImportService } from 'src/modules/messaging/message-import-manager/services/messaging-messages-import.service'; -const MAX_MESSAGE_COUNT_FOR_QUICK_IMPORT = 100; const ONE_WEEK_IN_MILLISECONDS = 7 * 24 * 60 * 60 * 1000; @Injectable() @@ -36,6 +36,7 @@ export class MessagingMessageListFetchService { private readonly messagingMessageCleanerService: MessagingMessageCleanerService, private readonly messagingCursorService: MessagingCursorService, private readonly messagingMessagesImportService: MessagingMessagesImportService, + private readonly messagingAccountAuthenticationService: MessagingAccountAuthenticationService, ) {} public async processMessageListFetch( @@ -47,9 +48,27 @@ export class MessagingMessageListFetchService { [messageChannel.id], ); + const { accessToken, refreshToken } = + await this.messagingAccountAuthenticationService.validateAndRefreshConnectedAccountAuthentication( + { + connectedAccount: messageChannel.connectedAccount, + workspaceId, + messageChannelId: messageChannel.id, + }, + ); + + const messageChannelWithFreshTokens = { + ...messageChannel, + connectedAccount: { + ...messageChannel.connectedAccount, + accessToken, + refreshToken, + }, + }; + const messageLists = await this.messagingGetMessageListService.getMessageLists( - messageChannel, + messageChannelWithFreshTokens, ); await this.cacheStorage.del( @@ -116,7 +135,7 @@ export class MessagingMessageListFetchService { if (allMessageExternalIdsToDelete.length) { await messageChannelMessageAssociationRepository.delete({ - messageChannelId: messageChannel.id, + messageChannelId: messageChannelWithFreshTokens.id, messageExternalId: In(allMessageExternalIdsToDelete), }); @@ -127,14 +146,14 @@ export class MessagingMessageListFetchService { if (messageExternalIdsToImport.length) { await this.cacheStorage.setAdd( - `messages-to-import:${workspaceId}:${messageChannel.id}`, + `messages-to-import:${workspaceId}:${messageChannelWithFreshTokens.id}`, messageExternalIdsToImport, ONE_WEEK_IN_MILLISECONDS, ); } await this.messagingCursorService.updateCursor( - messageChannel, + messageChannelWithFreshTokens, nextSyncCursor, folderId, ); @@ -142,26 +161,22 @@ export class MessagingMessageListFetchService { if (totalMessageCount === 0) { await this.messageChannelSyncStatusService.markAsCompletedAndScheduleMessageListFetch( - [messageChannel.id], + [messageChannelWithFreshTokens.id], ); - - return; } await this.messageChannelSyncStatusService.scheduleMessagesImport([ - messageChannel.id, + messageChannelWithFreshTokens.id, ]); - if (totalMessageCount < MAX_MESSAGE_COUNT_FOR_QUICK_IMPORT) { - await this.messagingMessagesImportService.processMessageBatchImport( - { - ...messageChannel, - syncStage: MessageChannelSyncStage.MESSAGES_IMPORT_PENDING, - }, - messageChannel.connectedAccount, - workspaceId, - ); - } + await this.messagingMessagesImportService.processMessageBatchImport( + { + ...messageChannelWithFreshTokens, + syncStage: MessageChannelSyncStage.MESSAGES_IMPORT_PENDING, + }, + messageChannelWithFreshTokens.connectedAccount, + workspaceId, + ); } catch (error) { await this.messageImportErrorHandlerService.handleDriverException( error, diff --git a/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-message.service.ts b/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-message.service.ts index 077127c8e7..44982f4d64 100644 --- a/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-message.service.ts +++ b/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-message.service.ts @@ -44,6 +44,7 @@ export class MessagingMessageService { messages: MessageWithParticipants[], messageChannelId: string, transactionManager: WorkspaceEntityManager, + workspaceId: string, ): Promise<{ createdMessages: Partial[]; messageExternalIdsAndIdsMap: Map; @@ -103,6 +104,7 @@ export class MessagingMessageService { messages, messageAccumulatorMap, messageChannelMessageAssociationsReferencingMessageThread, + workspaceId, ); await this.enrichMessageAccumulatorWithExistingMessageChannelMessageAssociations( @@ -252,6 +254,7 @@ export class MessagingMessageService { MessageChannelMessageAssociationWorkspaceEntity, 'messageThreadExternalId' | 'message' >[], + workspaceId: string, ) { for (const message of messages) { const messageAccumulator = messageAccumulatorMap.get(message.externalId); @@ -305,7 +308,16 @@ export class MessagingMessageService { // this means that we have to channels that have imported messages separately and that this message is the first connection between the two channels // we should merge messageThreads this.logger.warn( - `Message thread id is different for the same message header id and message thread external id, this means that we have to channels that have imported messages separately and that this message is the first connection between the two channels, we should merge messageThreads`, + ` + WorkspaceId: ${workspaceId} / + Message ExternalId: ${message.externalId} / + Message HeaderId: ${message.headerMessageId} / + Message Thread ExternalId: ${message.messageThreadExternalId} / + Message Thread Id in DB: ${existingThreadIdInDBIfMessageIsExistingInDB} / + Message Thread Id in Message Channel Message Association: ${existingThreadIdInDBIfMessageIsReferencedInMessageChannelMessageAssociation} / + Message Subject: ${message.subject} / + Message Received At: ${message.receivedAt} / + Thread inter channel detected`, ); } diff --git a/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-messages-import.service.spec.ts b/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-messages-import.service.spec.ts index 51fbca8c0a..ccdf8c7a6a 100644 --- a/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-messages-import.service.spec.ts +++ b/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-messages-import.service.spec.ts @@ -75,7 +75,10 @@ describe('MessagingMessagesImportService', () => { { provide: ConnectedAccountRefreshTokensService, useValue: { - refreshAndSaveTokens: jest.fn().mockResolvedValue('new-access-token'), + refreshAndSaveTokens: jest.fn().mockResolvedValue({ + accessToken: 'new-access-token', + refreshToken: 'new-refresh-token', + }), }, }, { @@ -211,15 +214,23 @@ describe('MessagingMessagesImportService', () => { expect( messageChannelSyncStatusService.markAsMessagesImportOngoing, ).toHaveBeenCalledWith([mockMessageChannel.id]); + expect( connectedAccountRefreshTokensService.refreshAndSaveTokens, ).toHaveBeenCalledWith(mockConnectedAccount, workspaceId); - expect(emailAliasManagerService.refreshHandleAliases).toHaveBeenCalledWith( - mockConnectedAccount, - ); + + expect(emailAliasManagerService.refreshHandleAliases).toHaveBeenCalledWith({ + ...mockConnectedAccount, + accessToken: 'new-access-token', + refreshToken: 'new-refresh-token', + }); expect(messagingGetMessagesService.getMessages).toHaveBeenCalledWith( ['message-id-1', 'message-id-2'], - mockConnectedAccount, + { + ...mockConnectedAccount, + accessToken: 'new-access-token', + refreshToken: 'new-refresh-token', + }, ); expect( saveMessagesService.saveMessagesAndEnqueueContactCreation, diff --git a/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-messages-import.service.ts b/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-messages-import.service.ts index 0f48279f90..c0267e073c 100644 --- a/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-messages-import.service.ts +++ b/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-messages-import.service.ts @@ -71,14 +71,23 @@ export class MessagingMessagesImportService { messageChannel.id, ]); - await this.messagingAccountAuthenticationService.validateConnectedAccountAuthentication( - connectedAccount, - workspaceId, - messageChannel.id, - ); + const { accessToken, refreshToken } = + await this.messagingAccountAuthenticationService.validateAndRefreshConnectedAccountAuthentication( + { + connectedAccount, + workspaceId, + messageChannelId: messageChannel.id, + }, + ); + + const connectedAccountWithFreshTokens = { + ...connectedAccount, + accessToken, + refreshToken, + }; await this.emailAliasManagerService.refreshHandleAliases( - connectedAccount, + connectedAccountWithFreshTokens, ); messageIdsToFetch = await this.cacheStorage.setPop( @@ -99,17 +108,17 @@ export class MessagingMessagesImportService { const allMessages = await this.messagingGetMessagesService.getMessages( messageIdsToFetch, - connectedAccount, + connectedAccountWithFreshTokens, ); const blocklist = await this.blocklistRepository.getByWorkspaceMemberId( - connectedAccount.accountOwnerId, + connectedAccountWithFreshTokens.accountOwnerId, workspaceId, ); const messagesToSave = filterEmails( messageChannel.handle, - [...connectedAccount.handleAliases.split(',')], + [...connectedAccountWithFreshTokens.handleAliases.split(',')], allMessages, blocklist.map((blocklistItem) => blocklistItem.handle), ); @@ -117,7 +126,7 @@ export class MessagingMessagesImportService { await this.saveMessagesAndEnqueueContactCreationService.saveMessagesAndEnqueueContactCreation( messagesToSave, messageChannel, - connectedAccount, + connectedAccountWithFreshTokens, workspaceId, ); diff --git a/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-save-messages-and-enqueue-contact-creation.service.spec.ts b/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-save-messages-and-enqueue-contact-creation.service.spec.ts index 4e009db238..1de6cfccec 100644 --- a/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-save-messages-and-enqueue-contact-creation.service.spec.ts +++ b/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-save-messages-and-enqueue-contact-creation.service.spec.ts @@ -165,6 +165,7 @@ describe('MessagingSaveMessagesAndEnqueueContactCreationService', () => { mockMessages, mockMessageChannel.id, expect.any(Object), + workspaceId, ); expect( diff --git a/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-save-messages-and-enqueue-contact-creation.service.ts b/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-save-messages-and-enqueue-contact-creation.service.ts index 68acc3c7de..7b55a0ae78 100644 --- a/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-save-messages-and-enqueue-contact-creation.service.ts +++ b/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-save-messages-and-enqueue-contact-creation.service.ts @@ -53,6 +53,7 @@ export class MessagingSaveMessagesAndEnqueueContactCreationService { messagesToSave, messageChannel.id, transactionManager, + workspaceId, ); const participantsWithMessageId: (ParticipantWithMessageId & {