[messaging] add cronjob for workspaces messages partial sync (#3800)

* [messaging] add cronjob for workspaces messages partial sync

* run cron every 10 minutes

* use logger

packages/twenty-server/src/command.module.ts

@@ -6,6 +6,8 @@ import { StartCleanInactiveWorkspacesCronCommand } from 'src/workspace/cron/clea
 import { StopCleanInactiveWorkspacesCronCommand } from 'src/workspace/cron/clean-inactive-workspaces/commands/stop-clean-inactive-workspaces.cron.command';
 import { CleanInactiveWorkspacesCommand } from 'src/workspace/cron/clean-inactive-workspaces/commands/clean-inactive-workspaces.command';
 import { WorkspaceHealthCommandModule } from 'src/workspace/workspace-health/commands/workspace-health-command.module';
+import { StartFetchAllWorkspacesMessagesCronCommand } from 'src/workspace/cron/fetch-all-workspaces-messages/commands/start-fetch-all-workspaces-messages.cron.command';
+import { StopFetchAllWorkspacesMessagesCronCommand } from 'src/workspace/cron/fetch-all-workspaces-messages/commands/stop-fetch-all-workspaces-messages.cron.command';
 
 import { AppModule } from './app.module';
 
@@ -23,6 +25,8 @@ import { WorkspaceMigrationRunnerCommandsModule } from './workspace/workspace-mi
     CleanInactiveWorkspacesCommand,
     WorkspaceHealthCommandModule,
     WorkspaceMigrationRunnerCommandsModule,
+    StartFetchAllWorkspacesMessagesCronCommand,
+    StopFetchAllWorkspacesMessagesCronCommand,
   ],
 })
 export class CommandModule {}

packages/twenty-server/src/integrations/message-queue/jobs.module.ts

@@ -17,6 +17,8 @@ import { EmailSenderJob } from 'src/integrations/email/email-sender.job';
 import { UserModule } from 'src/core/user/user.module';
 import { EnvironmentModule } from 'src/integrations/environment/environment.module';
 import { FeatureFlagEntity } from 'src/core/feature-flag/feature-flag.entity';
+import { FetchAllWorkspacesMessagesJob } from 'src/workspace/cron/fetch-all-workspaces-messages/fetch-all-workspaces-messages.job';
+import { ConnectedAccountModule } from 'src/workspace/messaging/connected-account/connected-account.module';
 
 @Module({
   imports: [
@@ -30,6 +32,7 @@ import { FeatureFlagEntity } from 'src/core/feature-flag/feature-flag.entity';
     EnvironmentModule,
     TypeORMModule,
     TypeOrmModule.forFeature([FeatureFlagEntity], 'core'),
+    ConnectedAccountModule,
   ],
   providers: [
     {
@@ -53,6 +56,10 @@ import { FeatureFlagEntity } from 'src/core/feature-flag/feature-flag.entity';
       useClass: CleanInactiveWorkspaceJob,
     },
     { provide: EmailSenderJob.name, useClass: EmailSenderJob },
+    {
+      provide: FetchAllWorkspacesMessagesJob.name,
+      useClass: FetchAllWorkspacesMessagesJob,
+    },
   ],
 })
 export class JobsModule {

packages/twenty-server/src/queue-worker.module.ts

@@ -1,27 +1,9 @@
 import { Module } from '@nestjs/common';
 
-import { EnvironmentModule } from 'src/integrations/environment/environment.module';
-import { EnvironmentService } from 'src/integrations/environment/environment.service';
-import { LoggerModule } from 'src/integrations/logger/logger.module';
-import { loggerModuleFactory } from 'src/integrations/logger/logger.module-factory';
 import { JobsModule } from 'src/integrations/message-queue/jobs.module';
-import { MessageQueueModule } from 'src/integrations/message-queue/message-queue.module';
-import { messageQueueModuleFactory } from 'src/integrations/message-queue/message-queue.module-factory';
 import { IntegrationsModule } from 'src/integrations/integrations.module';
 
 @Module({
-  imports: [
+  imports: [IntegrationsModule, JobsModule],
-    EnvironmentModule.forRoot({}),
-    LoggerModule.forRootAsync({
-      useFactory: loggerModuleFactory,
-      inject: [EnvironmentService],
-    }),
-    MessageQueueModule.forRoot({
-      useFactory: messageQueueModuleFactory,
-      inject: [EnvironmentService],
-    }),
-    JobsModule,
-    IntegrationsModule,
-  ],
 })
 export class QueueWorkerModule {}

packages/twenty-server/src/workspace/cron/fetch-all-workspaces-messages/commands/start-fetch-all-workspaces-messages.cron.command.ts

@@ -0,0 +1,29 @@
+import { Inject } from '@nestjs/common';
+
+import { Command, CommandRunner } from 'nest-commander';
+
+import { MessageQueue } from 'src/integrations/message-queue/message-queue.constants';
+import { MessageQueueService } from 'src/integrations/message-queue/services/message-queue.service';
+import { fetchAllWorkspacesMessagesCronPattern } from 'src/workspace/cron/fetch-all-workspaces-messages/fetch-all-workspaces-messages.cron.pattern';
+import { FetchAllWorkspacesMessagesJob } from 'src/workspace/cron/fetch-all-workspaces-messages/fetch-all-workspaces-messages.job';
+
+@Command({
+  name: 'fetch-all-workspaces-messages:cron:start',
+  description: 'Starts a cron job to fetch all workspaces messages',
+})
+export class StartFetchAllWorkspacesMessagesCronCommand extends CommandRunner {
+  constructor(
+    @Inject(MessageQueue.cronQueue)
+    private readonly messageQueueService: MessageQueueService,
+  ) {
+    super();
+  }
+
+  async run(): Promise<void> {
+    await this.messageQueueService.addCron<undefined>(
+      FetchAllWorkspacesMessagesJob.name,
+      undefined,
+      fetchAllWorkspacesMessagesCronPattern,
+    );
+  }
+}

packages/twenty-server/src/workspace/cron/fetch-all-workspaces-messages/commands/stop-fetch-all-workspaces-messages.cron.command.ts

@@ -0,0 +1,28 @@
+import { Inject } from '@nestjs/common';
+
+import { Command, CommandRunner } from 'nest-commander';
+
+import { MessageQueue } from 'src/integrations/message-queue/message-queue.constants';
+import { MessageQueueService } from 'src/integrations/message-queue/services/message-queue.service';
+import { fetchAllWorkspacesMessagesCronPattern } from 'src/workspace/cron/fetch-all-workspaces-messages/fetch-all-workspaces-messages.cron.pattern';
+import { FetchAllWorkspacesMessagesJob } from 'src/workspace/cron/fetch-all-workspaces-messages/fetch-all-workspaces-messages.job';
+
+@Command({
+  name: 'fetch-all-workspaces-messages:cron:stop',
+  description: 'Stops the fetch all workspaces messages cron job',
+})
+export class StopFetchAllWorkspacesMessagesCronCommand extends CommandRunner {
+  constructor(
+    @Inject(MessageQueue.cronQueue)
+    private readonly messageQueueService: MessageQueueService,
+  ) {
+    super();
+  }
+
+  async run(): Promise<void> {
+    await this.messageQueueService.removeCron(
+      FetchAllWorkspacesMessagesJob.name,
+      fetchAllWorkspacesMessagesCronPattern,
+    );
+  }
+}

packages/twenty-server/src/workspace/cron/fetch-all-workspaces-messages/fetch-all-workspaces-messages.cron.pattern.ts

@@ -0,0 +1 @@
+export const fetchAllWorkspacesMessagesCronPattern = '*/10 * * * *';

packages/twenty-server/src/workspace/cron/fetch-all-workspaces-messages/fetch-all-workspaces-messages.job.ts

@@ -0,0 +1,66 @@
+import { Inject, Injectable } from '@nestjs/common';
+import { InjectRepository } from '@nestjs/typeorm';
+
+import { Repository } from 'typeorm';
+
+import { MessageQueueJob } from 'src/integrations/message-queue/interfaces/message-queue-job.interface';
+
+import {
+  FeatureFlagEntity,
+  FeatureFlagKeys,
+} from 'src/core/feature-flag/feature-flag.entity';
+import { MessageQueue } from 'src/integrations/message-queue/message-queue.constants';
+import { MessageQueueService } from 'src/integrations/message-queue/services/message-queue.service';
+import { ConnectedAccountService } from 'src/workspace/messaging/connected-account/connected-account.service';
+import {
+  GmailPartialSyncJobData,
+  GmailPartialSyncJob,
+} from 'src/workspace/messaging/jobs/gmail-partial-sync.job';
+
+@Injectable()
+export class FetchAllWorkspacesMessagesJob
+  implements MessageQueueJob<undefined>
+{
+  constructor(
+    @InjectRepository(FeatureFlagEntity, 'core')
+    private readonly featureFlagRepository: Repository<FeatureFlagEntity>,
+    @Inject(MessageQueue.messagingQueue)
+    private readonly messageQueueService: MessageQueueService,
+    private readonly connectedAccountService: ConnectedAccountService,
+  ) {}
+
+  async handle(): Promise<void> {
+    const featureFlagsWithMessagingEnabled =
+      await this.featureFlagRepository.findBy({
+        key: FeatureFlagKeys.IsMessagingEnabled,
+        value: true,
+      });
+
+    const workspaceIds = featureFlagsWithMessagingEnabled.map(
+      (featureFlag) => featureFlag.workspaceId,
+    );
+
+    for (const workspaceId of workspaceIds) {
+      await this.fetchWorkspaceMessages(workspaceId);
+    }
+  }
+
+  private async fetchWorkspaceMessages(workspaceId: string): Promise<void> {
+    const connectedAccounts =
+      await this.connectedAccountService.getAll(workspaceId);
+
+    for (const connectedAccount of connectedAccounts) {
+      await this.messageQueueService.add<GmailPartialSyncJobData>(
+        GmailPartialSyncJob.name,
+        {
+          workspaceId,
+          connectedAccountId: connectedAccount.id,
+        },
+        {
+          id: `${workspaceId}-${connectedAccount.id}`,
+          retryLimit: 2,
+        },
+      );
+    }
+  }
+}

packages/twenty-server/src/workspace/messaging/jobs/gmail-full-sync.job.ts

@@ -1,8 +1,7 @@
-import { Injectable } from '@nestjs/common';
+import { Injectable, Logger } from '@nestjs/common';
 
 import { MessageQueueJob } from 'src/integrations/message-queue/interfaces/message-queue-job.interface';
 
-import { EnvironmentService } from 'src/integrations/environment/environment.service';
 import { GmailRefreshAccessTokenService } from 'src/workspace/messaging/services/gmail-refresh-access-token.service';
 import { GmailFullSyncService } from 'src/workspace/messaging/services/gmail-full-sync.service';
 
@@ -14,26 +13,25 @@ export type GmailFullSyncJobData = {
 
 @Injectable()
 export class GmailFullSyncJob implements MessageQueueJob<GmailFullSyncJobData> {
+  private readonly logger = new Logger(GmailFullSyncJob.name);
+
   constructor(
-    private readonly environmentService: EnvironmentService,
     private readonly gmailRefreshAccessTokenService: GmailRefreshAccessTokenService,
-    private readonly fetchWorkspaceMessagesService: GmailFullSyncService,
+    private readonly gmailFullSyncService: GmailFullSyncService,
   ) {}
 
   async handle(data: GmailFullSyncJobData): Promise<void> {
-    console.log(
+    this.logger.log(
       `gmail full-sync for workspace ${data.workspaceId} and account ${
         data.connectedAccountId
-      } ${
+      } ${data.nextPageToken ? `and ${data.nextPageToken} pageToken` : ''}`,
-        data.nextPageToken ? `and ${data.nextPageToken} pageToken` : ''
-      } with ${this.environmentService.getMessageQueueDriverType()}`,
     );
     await this.gmailRefreshAccessTokenService.refreshAndSaveAccessToken(
       data.workspaceId,
       data.connectedAccountId,
     );
 
-    await this.fetchWorkspaceMessagesService.fetchConnectedAccountThreads(
+    await this.gmailFullSyncService.fetchConnectedAccountThreads(
       data.workspaceId,
       data.connectedAccountId,
       data.nextPageToken,

packages/twenty-server/src/workspace/messaging/jobs/gmail-partial-sync.job.ts

@@ -1,8 +1,7 @@
-import { Injectable } from '@nestjs/common';
+import { Injectable, Logger } from '@nestjs/common';
 
 import { MessageQueueJob } from 'src/integrations/message-queue/interfaces/message-queue-job.interface';
 
-import { EnvironmentService } from 'src/integrations/environment/environment.service';
 import { GmailRefreshAccessTokenService } from 'src/workspace/messaging/services/gmail-refresh-access-token.service';
 import { GmailPartialSyncService } from 'src/workspace/messaging/services/gmail-partial-sync.service';
 
@@ -15,17 +14,16 @@ export type GmailPartialSyncJobData = {
 export class GmailPartialSyncJob
   implements MessageQueueJob<GmailPartialSyncJobData>
 {
+  private readonly logger = new Logger(GmailPartialSyncJob.name);
+
   constructor(
-    private readonly environmentService: EnvironmentService,
     private readonly gmailRefreshAccessTokenService: GmailRefreshAccessTokenService,
     private readonly gmailPartialSyncService: GmailPartialSyncService,
   ) {}
 
   async handle(data: GmailPartialSyncJobData): Promise<void> {
-    console.log(
+    this.logger.log(
-      `gmail partial-sync for workspace ${data.workspaceId} and account ${
+      `gmail partial-sync for workspace ${data.workspaceId} and account ${data.connectedAccountId}`,
-        data.connectedAccountId
-      } with ${this.environmentService.getMessageQueueDriverType()}`,
     );
     await this.gmailRefreshAccessTokenService.refreshAndSaveAccessToken(
       data.workspaceId,

packages/twenty-server/src/workspace/messaging/services/gmail-full-sync.service.ts

@@ -1,4 +1,4 @@
-import { Inject, Injectable } from '@nestjs/common';
+import { Inject, Injectable, Logger } from '@nestjs/common';
 
 import { FetchMessagesByBatchesService } from 'src/workspace/messaging/services/fetch-messages-by-batches.service';
 import { GmailClientProvider } from 'src/workspace/messaging/providers/gmail/gmail-client.provider';
@@ -16,6 +16,8 @@ import { WorkspaceDataSourceService } from 'src/workspace/workspace-datasource/w
 
 @Injectable()
 export class GmailFullSyncService {
+  private readonly logger = new Logger(GmailFullSyncService.name);
+
   constructor(
     private readonly gmailClientProvider: GmailClientProvider,
     private readonly fetchMessagesByBatchesService: FetchMessagesByBatchesService,
@@ -135,7 +137,7 @@ export class GmailFullSyncService {
       workspaceId,
     );
 
-    console.log(
+    this.logger.log(
       `gmail full-sync for workspace ${workspaceId} and account ${connectedAccountId} ${
         nextPageToken ? `and ${nextPageToken} pageToken` : ''
       }done.`,

packages/twenty-server/src/workspace/messaging/services/gmail-partial-sync.service.ts

@@ -1,4 +1,4 @@
-import { Inject, Injectable } from '@nestjs/common';
+import { Inject, Injectable, Logger } from '@nestjs/common';
 
 import { gmail_v1 } from 'googleapis';
 
@@ -17,6 +17,8 @@ import { WorkspaceDataSourceService } from 'src/workspace/workspace-datasource/w
 
 @Injectable()
 export class GmailPartialSyncService {
+  private readonly logger = new Logger(GmailPartialSyncService.name);
+
   constructor(
     private readonly gmailClientProvider: GmailClientProvider,
     private readonly fetchMessagesByBatchesService: FetchMessagesByBatchesService,
@@ -77,6 +79,10 @@ export class GmailPartialSyncService {
     }
 
     if (newHistoryId === lastSyncHistoryId) {
+      this.logger.log(
+        `gmail partial-sync for workspace ${workspaceId} and account ${connectedAccountId} done with nothing to update.`,
+      );
+
       return;
     }
 
@@ -127,6 +133,10 @@ export class GmailPartialSyncService {
       connectedAccount.id,
       workspaceId,
     );
+
+    this.logger.log(
+      `gmail partial-sync for workspace ${workspaceId} and account ${connectedAccountId} done.`,
+    );
   }
 
   private async getMessageIdsFromHistory(