[messaging] add cronjob for workspaces messages partial sync

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/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 = '* * * * *';

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/services/gmail-partial-sync.service.ts

@@ -77,6 +77,10 @@ export class GmailPartialSyncService {
     }
 
     if (newHistoryId === lastSyncHistoryId) {
+      console.log(
+        `gmail partial-sync for workspace ${workspaceId} and account ${connectedAccountId} done with nothing to update.`,
+      );
+
       return;
     }
 
@@ -127,6 +131,10 @@ export class GmailPartialSyncService {
       connectedAccount.id,
       workspaceId,
     );
+
+    console.log(
+      `gmail partial-sync for workspace ${workspaceId} and account ${connectedAccountId} done.`,
+    );
   }
 
   private async getMessageIdsFromHistory(