5531 update gmail full sync to v2

packages/twenty-server/src/modules/connected-account/repositories/connected-account.repository.ts

@@ -270,14 +270,6 @@ export class ConnectedAccountRepository {
       );
     }
 
-    const refreshToken = connectedAccount.refreshToken;
-
-    if (!refreshToken) {
-      throw new Error(
-        `No refresh token found for connected account ${connectedAccountId} in workspace ${workspaceId}`,
-      );
-    }
-
     return connectedAccount;
   }
 }

packages/twenty-server/src/modules/connected-account/services/google-api-refresh-access-token/google-api-refresh-access-token.module.ts

@@ -3,12 +3,14 @@ import { Module } from '@nestjs/common';
 import { ObjectMetadataRepositoryModule } from 'src/engine/object-metadata-repository/object-metadata-repository.module';
 import { GoogleAPIRefreshAccessTokenService } from 'src/modules/connected-account/services/google-api-refresh-access-token/google-api-refresh-access-token.service';
 import { ConnectedAccountWorkspaceEntity } from 'src/modules/connected-account/standard-objects/connected-account.workspace-entity';
+import { SetMessageChannelSyncStatusModule } from 'src/modules/messaging/services/set-message-channel-sync-status/set-message-channel-sync-status.module';
 
 @Module({
   imports: [
     ObjectMetadataRepositoryModule.forFeature([
       ConnectedAccountWorkspaceEntity,
     ]),
+    SetMessageChannelSyncStatusModule,
   ],
   providers: [GoogleAPIRefreshAccessTokenService],
   exports: [GoogleAPIRefreshAccessTokenService],

packages/twenty-server/src/modules/connected-account/services/google-api-refresh-access-token/google-api-refresh-access-token.service.ts

@@ -6,13 +6,19 @@ import { EnvironmentService } from 'src/engine/integrations/environment/environm
 import { InjectObjectMetadataRepository } from 'src/engine/object-metadata-repository/object-metadata-repository.decorator';
 import { ConnectedAccountRepository } from 'src/modules/connected-account/repositories/connected-account.repository';
 import { ConnectedAccountWorkspaceEntity } from 'src/modules/connected-account/standard-objects/connected-account.workspace-entity';
+import { MessageChannelRepository } from 'src/modules/messaging/repositories/message-channel.repository';
+import { SetMessageChannelSyncStatusService } from 'src/modules/messaging/services/set-message-channel-sync-status/set-message-channel-sync-status.service';
+import { MessageChannelWorkspaceEntity } from 'src/modules/messaging/standard-objects/message-channel.workspace-entity';
 
 @Injectable()
 export class GoogleAPIRefreshAccessTokenService {
   constructor(
     private readonly environmentService: EnvironmentService,
     @InjectObjectMetadataRepository(ConnectedAccountWorkspaceEntity)
     private readonly connectedAccountRepository: ConnectedAccountRepository,
+    private readonly setMessageChannelSyncStatusService: SetMessageChannelSyncStatusService,
+    @InjectObjectMetadataRepository(MessageChannelWorkspaceEntity)
+    private readonly messageChannelRepository: MessageChannelRepository,
   ) {}
 
   async refreshAndSaveAccessToken(
@@ -87,7 +93,26 @@ export class GoogleAPIRefreshAccessTokenService {
         workspaceId,
       );
 
-      throw new Error(`Error refreshing access token: ${error.message}`);
+      const messageChannel =
+        await this.messageChannelRepository.getFirstByConnectedAccountId(
+          connectedAccountId,
+          workspaceId,
+        );
+
+      if (!messageChannel) {
+        throw new Error(
+          `No message channel found for connected account ${connectedAccountId} in workspace ${workspaceId}`,
+        );
+      }
+
+      await this.setMessageChannelSyncStatusService.setFailedInsufficientPermissionsStatus(
+        messageChannel.id,
+        workspaceId,
+      );
+
+      throw new Error(
+        `Error refreshing access token: ${error.code}: ${error.message}`,
+      );
     }
   }
 }

packages/twenty-server/src/modules/messaging/crons/commands/gmail-message-list-fetch.cron.command.ts

@@ -6,7 +6,7 @@ import { MessageQueue } from 'src/engine/integrations/message-queue/message-queu
 import { MessageQueueService } from 'src/engine/integrations/message-queue/services/message-queue.service';
 import { GmailMessageListFetchCronJob } from 'src/modules/messaging/crons/jobs/gmail-message-list-fetch.cron.job';
 
-const GMAIL_PARTIAL_SYNC_CRON_PATTERN = '*/5 * * * *';
+const GMAIL_MESSAGE_LIST_FETCH_CRON_PATTERN = '*/5 * * * *';
 
 @Command({
   name: 'cron:messaging:gmail-message-list-fetch',
@@ -26,7 +26,7 @@ export class GmailMessageListFetchCronCommand extends CommandRunner {
       GmailMessageListFetchCronJob.name,
       undefined,
       {
-        repeat: { pattern: GMAIL_PARTIAL_SYNC_CRON_PATTERN },
+        repeat: { pattern: GMAIL_MESSAGE_LIST_FETCH_CRON_PATTERN },
       },
     );
   }

packages/twenty-server/src/modules/messaging/jobs/blocklist-reimport-messages.job.ts

@@ -22,7 +22,7 @@ export class BlocklistReimportMessagesJob
   constructor(
     @InjectObjectMetadataRepository(ConnectedAccountWorkspaceEntity)
     private readonly connectedAccountRepository: ConnectedAccountRepository,
-    private readonly gmailFullSyncService: GmailFullMessageListFetchService,
+    private readonly gmailFullMessageListFetchService: GmailFullMessageListFetchService,
   ) {}
 
   async handle(data: BlocklistReimportMessagesJobData): Promise<void> {
@@ -46,7 +46,7 @@ export class BlocklistReimportMessagesJob
       return;
     }
 
-    await this.gmailFullSyncService.fetchConnectedAccountThreads(
+    await this.gmailFullMessageListFetchService.fetchConnectedAccountThreads(
       workspaceId,
       connectedAccount[0].id,
       [handle],

packages/twenty-server/src/modules/messaging/jobs/gmail-full-message-list-fetch.job.ts

@@ -1,9 +1,23 @@
 import { Injectable, Logger } from '@nestjs/common';
+import { InjectRepository } from '@nestjs/typeorm';
+
+import { Repository } from 'typeorm';
 
 import { MessageQueueJob } from 'src/engine/integrations/message-queue/interfaces/message-queue-job.interface';
 
+import {
+  FeatureFlagEntity,
+  FeatureFlagKeys,
+} from 'src/engine/core-modules/feature-flag/feature-flag.entity';
 import { GoogleAPIRefreshAccessTokenService } from 'src/modules/connected-account/services/google-api-refresh-access-token/google-api-refresh-access-token.service';
 import { GmailFullMessageListFetchService } from 'src/modules/messaging/services/gmail-full-message-list-fetch/gmail-full-message-list-fetch.service';
+import { GmailFullMessageListFetchV2Service } from 'src/modules/messaging/services/gmail-full-message-list-fetch/gmail-full-message-list-fetch-v2.service';
+import { InjectObjectMetadataRepository } from 'src/engine/object-metadata-repository/object-metadata-repository.decorator';
+import { ConnectedAccountRepository } from 'src/modules/connected-account/repositories/connected-account.repository';
+import { ConnectedAccountWorkspaceEntity } from 'src/modules/connected-account/standard-objects/connected-account.workspace-entity';
+import { MessageChannelRepository } from 'src/modules/messaging/repositories/message-channel.repository';
+import { SetMessageChannelSyncStatusService } from 'src/modules/messaging/services/set-message-channel-sync-status/set-message-channel-sync-status.service';
+import { MessageChannelWorkspaceEntity } from 'src/modules/messaging/standard-objects/message-channel.workspace-entity';
 
 export type GmailFullMessageListFetchJobData = {
   workspaceId: string;
@@ -18,31 +32,101 @@ export class GmailFullMessageListFetchJob
 
   constructor(
     private readonly googleAPIsRefreshAccessTokenService: GoogleAPIRefreshAccessTokenService,
-    private readonly gmailFullSyncService: GmailFullMessageListFetchService,
+    private readonly gmailFullMessageListFetchService: GmailFullMessageListFetchService,
+    @InjectRepository(FeatureFlagEntity, 'core')
+    private readonly featureFlagRepository: Repository<FeatureFlagEntity>,
+    private readonly gmailFullMessageListFetchV2Service: GmailFullMessageListFetchV2Service,
+    @InjectObjectMetadataRepository(ConnectedAccountWorkspaceEntity)
+    private readonly connectedAccountRepository: ConnectedAccountRepository,
+    @InjectObjectMetadataRepository(MessageChannelWorkspaceEntity)
+    private readonly messageChannelRepository: MessageChannelRepository,
+    private readonly setMessageChannelSyncStatusService: SetMessageChannelSyncStatusService,
   ) {}
 
   async handle(data: GmailFullMessageListFetchJobData): Promise<void> {
+    const { workspaceId, connectedAccountId } = data;
+
     this.logger.log(
-      `gmail full-sync for workspace ${data.workspaceId} and account ${data.connectedAccountId}`,
+      `gmail full-sync for workspace ${workspaceId} and account ${connectedAccountId}`,
     );
 
     try {
       await this.googleAPIsRefreshAccessTokenService.refreshAndSaveAccessToken(
-        data.workspaceId,
-        data.connectedAccountId,
+        workspaceId,
+        connectedAccountId,
       );
     } catch (e) {
       this.logger.error(
-        `Error refreshing access token for connected account ${data.connectedAccountId} in workspace ${data.workspaceId}`,
+        `Error refreshing access token for connected account ${connectedAccountId} in workspace ${workspaceId}`,
         e,
       );
 
       return;
     }
 
-    await this.gmailFullSyncService.fetchConnectedAccountThreads(
-      data.workspaceId,
-      data.connectedAccountId,
-    );
+    const isGmailSyncV2EnabledFeatureFlag =
+      await this.featureFlagRepository.findOneBy({
+        workspaceId: workspaceId,
+        key: FeatureFlagKeys.IsGmailSyncV2Enabled,
+        value: true,
+      });
+
+    const isGmailSyncV2Enabled = isGmailSyncV2EnabledFeatureFlag?.value;
+
+    if (isGmailSyncV2Enabled) {
+      const connectedAccount = await this.connectedAccountRepository.getById(
+        connectedAccountId,
+        workspaceId,
+      );
+
+      if (!connectedAccount) {
+        throw new Error(
+          `Connected account ${connectedAccountId} not found in workspace ${workspaceId}`,
+        );
+      }
+
+      const messageChannel =
+        await this.messageChannelRepository.getFirstByConnectedAccountId(
+          connectedAccountId,
+          workspaceId,
+        );
+
+      if (!messageChannel) {
+        throw new Error(
+          `No message channel found for connected account ${connectedAccountId} in workspace ${workspaceId}`,
+        );
+      }
+
+      const refreshToken = connectedAccount.refreshToken;
+
+      if (!refreshToken) {
+        if (!connectedAccount.authFailedAt) {
+          await this.connectedAccountRepository.updateAuthFailedAt(
+            connectedAccountId,
+            workspaceId,
+          );
+        }
+
+        await this.setMessageChannelSyncStatusService.setFailedInsufficientPermissionsStatus(
+          messageChannel.id,
+          workspaceId,
+        );
+
+        throw new Error(
+          `No refresh token found for connected account ${connectedAccountId} in workspace ${workspaceId}`,
+        );
+      }
+
+      await this.gmailFullMessageListFetchV2Service.processMessageListFetch(
+        messageChannel,
+        connectedAccount,
+        workspaceId,
+      );
+    } else {
+      await this.gmailFullMessageListFetchService.fetchConnectedAccountThreads(
+        data.workspaceId,
+        data.connectedAccountId,
+      );
+    }
   }
 }

packages/twenty-server/src/modules/messaging/jobs/gmail-message-list-fetch.job.ts

@@ -4,9 +4,15 @@ import { MessageQueueJob } from 'src/engine/integrations/message-queue/interface
 
 import { GoogleAPIRefreshAccessTokenService } from 'src/modules/connected-account/services/google-api-refresh-access-token/google-api-refresh-access-token.service';
 import { GmailPartialMessageListFetchV2Service } from 'src/modules/messaging/services/gmail-partial-message-list-fetch/gmail-partial-message-list-fetch-v2.service';
-import { GetConnectedAccountAndMessageChannelService } from 'src/modules/messaging/services/get-connected-account-and-message-channel/get-connected-account-and-message-channel.service';
-import { MessageChannelSyncSubStatus } from 'src/modules/messaging/standard-objects/message-channel.workspace-entity';
-import { GmailFullMessageListFetchService } from 'src/modules/messaging/services/gmail-full-message-list-fetch/gmail-full-message-list-fetch.service';
+import {
+  MessageChannelSyncSubStatus,
+  MessageChannelWorkspaceEntity,
+} from 'src/modules/messaging/standard-objects/message-channel.workspace-entity';
+import { GmailFullMessageListFetchV2Service } from 'src/modules/messaging/services/gmail-full-message-list-fetch/gmail-full-message-list-fetch-v2.service';
+import { InjectObjectMetadataRepository } from 'src/engine/object-metadata-repository/object-metadata-repository.decorator';
+import { ConnectedAccountRepository } from 'src/modules/connected-account/repositories/connected-account.repository';
+import { ConnectedAccountWorkspaceEntity } from 'src/modules/connected-account/standard-objects/connected-account.workspace-entity';
+import { MessageChannelRepository } from 'src/modules/messaging/repositories/message-channel.repository';
 
 export type GmailMessageListFetchJobData = {
   workspaceId: string;
@@ -21,9 +27,12 @@ export class GmailMessageListFetchJob
 
   constructor(
     private readonly googleAPIsRefreshAccessTokenService: GoogleAPIRefreshAccessTokenService,
-    private readonly gmailFullSyncService: GmailFullMessageListFetchService,
+    private readonly gmailFullMessageListFetchV2Service: GmailFullMessageListFetchV2Service,
     private readonly gmailPartialMessageListFetchV2Service: GmailPartialMessageListFetchV2Service,
-    private readonly getConnectedAccountAndMessageChannelService: GetConnectedAccountAndMessageChannelService,
+    @InjectObjectMetadataRepository(ConnectedAccountWorkspaceEntity)
+    private readonly connectedAccountRepository: ConnectedAccountRepository,
+    @InjectObjectMetadataRepository(MessageChannelWorkspaceEntity)
+    private readonly messageChannelRepository: MessageChannelRepository,
   ) {}
 
   async handle(data: GmailMessageListFetchJobData): Promise<void> {
@@ -47,12 +56,29 @@ export class GmailMessageListFetchJob
       return;
     }
 
-    const { messageChannel, connectedAccount } =
-      await this.getConnectedAccountAndMessageChannelService.getConnectedAccountAndMessageChannelOrThrow(
-        workspaceId,
+    const connectedAccount = await this.connectedAccountRepository.getById(
+      connectedAccountId,
+      workspaceId,
+    );
+
+    if (!connectedAccount) {
+      throw new Error(
+        `Connected account ${connectedAccountId} not found in workspace ${workspaceId}`,
+      );
+    }
+
+    const messageChannel =
+      await this.messageChannelRepository.getFirstByConnectedAccountId(
         connectedAccountId,
+        workspaceId,
       );
 
+    if (!messageChannel) {
+      throw new Error(
+        `No message channel found for connected account ${connectedAccountId} in workspace ${workspaceId}`,
+      );
+    }
+
     switch (messageChannel.syncSubStatus) {
       case MessageChannelSyncSubStatus.PARTIAL_MESSAGES_LIST_FETCH_PENDING:
         try {
@@ -69,9 +95,10 @@ export class GmailMessageListFetchJob
 
       case MessageChannelSyncSubStatus.FULL_MESSAGES_LIST_FETCH_PENDING:
         try {
-          await this.gmailFullSyncService.fetchConnectedAccountThreads(
+          await this.gmailFullMessageListFetchV2Service.processMessageListFetch(
+            messageChannel,
+            connectedAccount,
             workspaceId,
-            connectedAccountId,
           );
         } catch (e) {
           this.logger.error(e);
@@ -87,7 +114,7 @@ export class GmailMessageListFetchJob
         return;
 
       default:
-        this.logger.error(
+        this.logger.log(
           `Messaging import for workspace ${workspaceId} and account ${connectedAccountId} is locked, import will be retried later.`,
         );
 

packages/twenty-server/src/modules/messaging/jobs/messaging-job.module.ts

@@ -14,9 +14,9 @@ import { GmailFullMessageListFetchJob } from 'src/modules/messaging/jobs/gmail-f
 import { GmailMessageListFetchJob } from 'src/modules/messaging/jobs/gmail-message-list-fetch.job';
 import { GmailPartialMessageListFetchJob } from 'src/modules/messaging/jobs/gmail-partial-message-list-fetch.job';
 import { MessagingCreateCompanyAndContactAfterSyncJob } from 'src/modules/messaging/jobs/messaging-create-company-and-contact-after-sync.job';
-import { GetConnectedAccountAndMessageChannelModule } from 'src/modules/messaging/services/get-connected-account-and-message-channel/get-connected-account-and-message-channel.module';
 import { GmailFullMessageListFetchModule } from 'src/modules/messaging/services/gmail-full-message-list-fetch/gmail-full-message-list-fetch.module';
 import { GmailPartialMessageListFetchModule } from 'src/modules/messaging/services/gmail-partial-message-list-fetch/gmail-partial-message-list-fetch.module';
+import { SetMessageChannelSyncStatusModule } from 'src/modules/messaging/services/set-message-channel-sync-status/set-message-channel-sync-status.module';
 import { ThreadCleanerModule } from 'src/modules/messaging/services/thread-cleaner/thread-cleaner.module';
 import { MessageChannelMessageAssociationWorkspaceEntity } from 'src/modules/messaging/standard-objects/message-channel-message-association.workspace-entity';
 import { MessageChannelWorkspaceEntity } from 'src/modules/messaging/standard-objects/message-channel.workspace-entity';
@@ -37,7 +37,7 @@ import { MessageParticipantWorkspaceEntity } from 'src/modules/messaging/standar
     GoogleAPIRefreshAccessTokenModule,
     AutoCompaniesAndContactsCreationModule,
     TypeOrmModule.forFeature([FeatureFlagEntity], 'core'),
-    GetConnectedAccountAndMessageChannelModule,
+    SetMessageChannelSyncStatusModule,
   ],
   providers: [
     {