3491 launch cleaning cron (#3872)

* Add command to delete incomplete workspaces

* Inject command dependencies

* Fix command

* Do not delete core.workspace

* Reorganize files

* Delete src/workspace/cron

* Fix

---------

Co-authored-by: Charles Bochet <charles@twenty.com>

packages/twenty-server/src/workspace/messaging/commands/fetch-workspace-messages-commands.module.ts

@@ -7,6 +7,8 @@ import { DataSourceModule } from 'src/metadata/data-source/data-source.module';
 import { GmailFullSyncCommand } from 'src/workspace/messaging/commands/gmail-full-sync.command';
 import { GmailPartialSyncCommand } from 'src/workspace/messaging/commands/gmail-partial-sync.command';
 import { ConnectedAccountModule } from 'src/workspace/messaging/connected-account/connected-account.module';
+import { StartFetchAllWorkspacesMessagesCronCommand } from 'src/workspace/messaging/commands/start-fetch-all-workspaces-messages.cron.command';
+import { StopFetchAllWorkspacesMessagesCronCommand } from 'src/workspace/messaging/commands/stop-fetch-all-workspaces-messages.cron.command';
 
 @Module({
   imports: [
@@ -15,6 +17,11 @@ import { ConnectedAccountModule } from 'src/workspace/messaging/connected-accoun
     TypeOrmModule.forFeature([FeatureFlagEntity], 'core'),
     ConnectedAccountModule,
   ],
-  providers: [GmailFullSyncCommand, GmailPartialSyncCommand],
+  providers: [
+    GmailFullSyncCommand,
+    GmailPartialSyncCommand,
+    StartFetchAllWorkspacesMessagesCronCommand,
+    StopFetchAllWorkspacesMessagesCronCommand,
+  ],
 })
 export class FetchWorkspaceMessagesCommandsModule {}

packages/twenty-server/src/workspace/messaging/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/messaging/crons/fetch-all-workspaces-messages.cron.pattern';
+import { FetchAllWorkspacesMessagesJob } from 'src/workspace/messaging/crons/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/messaging/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/messaging/crons/fetch-all-workspaces-messages.cron.pattern';
+import { FetchAllWorkspacesMessagesJob } from 'src/workspace/messaging/crons/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/messaging/crons/fetch-all-workspaces-messages.cron.pattern.ts

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

packages/twenty-server/src/workspace/messaging/crons/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,
+        },
+      );
+    }
+  }
+}