Improve messaging behavior (#13746)

This PR contains multiple small enhancements
2025-08-07 20:21:18 +02:00
parent 6c2e11f830
commit 4baabd80ef
12 changed files with 236 additions and 147 deletions
@@ -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<GoogleTokens> {
+  async refreshAccessToken(
+    refreshToken: string,
+  ): Promise<ConnectedAccountTokens> {
    const response = await axios.post<GoogleRefreshTokenResponse>(
      'https://oauth2.googleapis.com/token',
      {
@@ -42,6 +45,7 @@ export class GoogleAPIRefreshAccessTokenService {

    return {
      accessToken: response.data.access_token,
+      refreshToken,
    };
  }
 }
@@ -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<MicrosoftTokens> {
+  async refreshTokens(refreshToken: string): Promise<ConnectedAccountTokens> {
    const response = await axios.post<MicrosoftRefreshTokenResponse>(
      'https://login.microsoftonline.com/common/oauth2/v2.0/token',
      new URLSearchParams({
@@ -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<string> {
+  ): Promise<ConnectedAccountTokens> {
    const refreshToken = connectedAccount.refreshToken;

    if (!refreshToken) {
@@ -52,23 +49,17 @@ export class ConnectedAccountRefreshTokensService {
      workspaceId,
    );

-    try {
-      const connectedAccountRepository =
-        await this.twentyORMManager.getRepository<ConnectedAccountWorkspaceEntity>(
-          'connectedAccount',
-        );
-
-      await connectedAccountRepository.update(
-        { id: connectedAccount.id },
-        connectedAccountTokens,
+    const connectedAccountRepository =
+      await this.twentyORMManager.getRepository<ConnectedAccountWorkspaceEntity>(
+        '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(
@@ -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
@@ -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<void> {
-    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<void> {
+  async validateAndRefreshConnectedAccountAuthentication({
+    connectedAccount,
+    workspaceId,
+    messageChannelId,
+  }: ValidateAndRefreshConnectedAccountAuthenticationParams): Promise<ConnectedAccountTokens> {
    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<void> {
-    if (!connectedAccount.connectionParameters) {
+  private async validateImapCredentialsForConnectedAccount({
+    connectedAccount,
+    workspaceId,
+    messageChannelId,
+  }: ValidateAndRefreshConnectedAccountAuthenticationParams): Promise<void> {
+    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<void> {
-    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<string> {
+  private async refreshAccessTokenForNonImapProvider({
+    connectedAccount,
+    workspaceId,
+    messageChannelId,
+  }: ValidateAndRefreshConnectedAccountAuthenticationParams): Promise<ConnectedAccountTokens> {
    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 ?? ''}`,
          });
@@ -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>(
        MessagingMessageListFetchService,
      );
+    messagingAccountAuthenticationService =
+      module.get<MessagingAccountAuthenticationService>(
+        MessagingAccountAuthenticationService,
+      );

    messagingGetMessageListService = module.get<MessagingGetMessageListService>(
      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,
    );
@@ -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,
@@ -44,6 +44,7 @@ export class MessagingMessageService {
    messages: MessageWithParticipants[],
    messageChannelId: string,
    transactionManager: WorkspaceEntityManager,
+    workspaceId: string,
  ): Promise<{
    createdMessages: Partial<MessageWorkspaceEntity>[];
    messageExternalIdsAndIdsMap: Map<string, string>;
@@ -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`,
        );
      }

@@ -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,
@@ -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,
      );

@@ -165,6 +165,7 @@ describe('MessagingSaveMessagesAndEnqueueContactCreationService', () => {
      mockMessages,
      mockMessageChannel.id,
      expect.any(Object),
+      workspaceId,
    );

    expect(
@@ -53,6 +53,7 @@ export class MessagingSaveMessagesAndEnqueueContactCreationService {
              messagesToSave,
              messageChannel.id,
              transactionManager,
+              workspaceId,
            );

          const participantsWithMessageId: (ParticipantWithMessageId & {