Improve workflow queue jobs (#14259)

- move cron commands logic to a workspace service
- add commands to enqueue not started and dequeue staled workflows
- stop dispatching one job per workspace. Handle all workspaces in one
job
- run enqueue cron every 5 minutes instead of every minutes

Follow up:
- add a separated redis key to control if there are not started
workflows. It will avoid fetching workflow runs for each workspace

packages/twenty-server/src/database/commands/cron-register-all.command.ts

@@ -11,7 +11,7 @@ import { CalendarOngoingStaleCronCommand } from 'src/modules/calendar/calendar-e
 import { MessagingMessageListFetchCronCommand } from 'src/modules/messaging/message-import-manager/crons/commands/messaging-message-list-fetch.cron.command';
 import { MessagingMessagesImportCronCommand } from 'src/modules/messaging/message-import-manager/crons/commands/messaging-messages-import.cron.command';
 import { MessagingOngoingStaleCronCommand } from 'src/modules/messaging/message-import-manager/crons/commands/messaging-ongoing-stale.cron.command';
-import { WorkflowCleanWorkflowRunsCommand } from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/command/workflow-clean-workflow-runs.cron.command';
+import { WorkflowCleanWorkflowRunsCronCommand } from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/command/workflow-clean-workflow-runs.cron.command';
 import { WorkflowHandleStaledRunsCronCommand } from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/command/workflow-handle-staled-runs.cron.command';
 import { WorkflowRunEnqueueCronCommand } from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/command/workflow-run-enqueue.cron.command';
 import { WorkflowCronTriggerCronCommand } from 'src/modules/workflow/workflow-trigger/automated-trigger/crons/commands/workflow-cron-trigger.cron.command';
@@ -36,7 +36,7 @@ export class CronRegisterAllCommand extends CommandRunner {
     private readonly checkCustomDomainValidRecordsCronCommand: CheckCustomDomainValidRecordsCronCommand,
     private readonly workflowRunEnqueueCronCommand: WorkflowRunEnqueueCronCommand,
     private readonly workflowHandleStaledRunsCronCommand: WorkflowHandleStaledRunsCronCommand,
-    private readonly workflowCleanWorkflowRunsCronCommand: WorkflowCleanWorkflowRunsCommand,
+    private readonly workflowCleanWorkflowRunsCronCommand: WorkflowCleanWorkflowRunsCronCommand,
     private readonly cronTriggerCronCommand: CronTriggerCronCommand,
   ) {
     super();

packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/command/workflow-handle-staled-runs.command.ts

@@ -0,0 +1,39 @@
+import { Command, CommandRunner, Option } from 'nest-commander';
+
+import { WorkflowHandleStaledRunsWorkspaceService } from 'src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-handle-staled-runs.workspace-service';
+
+type WorkflowHandleStaledRunsCommandOptions = {
+  workspaceIds: string[];
+};
+
+@Command({
+  name: 'workflow:handle-staled-runs',
+  description: 'Handles staled workflow runs',
+})
+export class WorkflowHandleStaledRunsCommand extends CommandRunner {
+  constructor(
+    private readonly workflowHandleStaledRunsWorkspaceService: WorkflowHandleStaledRunsWorkspaceService,
+  ) {
+    super();
+  }
+
+  @Option({
+    flags: '-w, --workspace-ids [workspace_ids]',
+    description: 'comma separated workspace ids - mandatory',
+    required: true,
+  })
+  parseWorkspaceIds(val: string): string[] {
+    return val.split(',');
+  }
+
+  async run(
+    _passedParam: string[],
+    options: WorkflowHandleStaledRunsCommandOptions,
+  ): Promise<void> {
+    const { workspaceIds } = options;
+
+    await this.workflowHandleStaledRunsWorkspaceService.handleStaledRuns({
+      workspaceIds,
+    });
+  }
+}

packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/command/workflow-run-enqueue.command.ts

@@ -0,0 +1,39 @@
+import { Command, CommandRunner, Option } from 'nest-commander';
+
+import { WorkflowRunEnqueueWorkspaceService } from 'src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-run-enqueue.workspace-service';
+
+type WorkflowRunEnqueueCommandOptions = {
+  workspaceIds: string[];
+};
+
+@Command({
+  name: 'workflow:run:enqueue',
+  description: 'Enqueues not started workflow runs',
+})
+export class WorkflowRunEnqueueCommand extends CommandRunner {
+  constructor(
+    private readonly workflowRunEnqueueWorkspaceService: WorkflowRunEnqueueWorkspaceService,
+  ) {
+    super();
+  }
+
+  @Option({
+    flags: '-w, --workspace-ids [workspace_ids]',
+    description: 'comma separated workspace ids - mandatory',
+    required: true,
+  })
+  parseWorkspaceIds(val: string): string[] {
+    return val.split(',');
+  }
+
+  async run(
+    _passedParam: string[],
+    options: WorkflowRunEnqueueCommandOptions,
+  ): Promise<void> {
+    const { workspaceIds } = options;
+
+    await this.workflowRunEnqueueWorkspaceService.enqueueRuns({
+      workspaceIds,
+    });
+  }
+}

packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/command/workflow-clean-workflow-runs.cron.command.ts

@@ -12,7 +12,7 @@ import {
   name: 'cron:workflow:clean-workflow-runs',
   description: 'Clean workflow runs',
 })
-export class WorkflowCleanWorkflowRunsCommand extends CommandRunner {
+export class WorkflowCleanWorkflowRunsCronCommand extends CommandRunner {
   constructor(
     @InjectMessageQueue(MessageQueue.cronQueue)
     private readonly messageQueueService: MessageQueueService,

packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/command/workflow-handle-staled-runs.cron.command.ts

@@ -6,7 +6,7 @@ import { MessageQueueService } from 'src/engine/core-modules/message-queue/servi
 import {
   WORKFLOW_HANDLE_STALED_RUNS_CRON_PATTERN,
   WorkflowHandleStaledRunsJob,
-} from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-handle-staled-runs.job';
+} from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-handle-staled-runs.cron.job';
 
 @Command({
   name: 'cron:workflow:handle-staled-runs',

packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/command/workflow-run-enqueue.cron.command.ts

@@ -6,7 +6,7 @@ import { MessageQueueService } from 'src/engine/core-modules/message-queue/servi
 import {
   WORKFLOW_RUN_ENQUEUE_CRON_PATTERN,
   WorkflowRunEnqueueJob,
-} from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-run-enqueue.job';
+} from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-run-enqueue.cron.job';
 
 @Command({
   name: 'cron:workflow:enqueue-awaiting-workflow-run',

packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-handle-staled-runs-per-workspace.job.ts

@@ -1,60 +0,0 @@
-import { IsNull, LessThan, Or } from 'typeorm';
-
-import { Process } from 'src/engine/core-modules/message-queue/decorators/process.decorator';
-import { Processor } from 'src/engine/core-modules/message-queue/decorators/processor.decorator';
-import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants';
-import { TwentyORMGlobalManager } from 'src/engine/twenty-orm/twenty-orm-global.manager';
-import {
-  WorkflowRunStatus,
-  WorkflowRunWorkspaceEntity,
-} from 'src/modules/workflow/common/standard-objects/workflow-run.workspace-entity';
-import { WorkflowRunQueueWorkspaceService } from 'src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-run-queue.workspace-service';
-
-export type WorkflowHandleStaledRunsPerWorkspaceJobData = {
-  workspaceId: string;
-};
-
-@Processor(MessageQueue.workflowQueue)
-export class WorkflowHandleStaledRunsPerWorkspaceJob {
-  constructor(
-    private readonly workflowRunQueueWorkspaceService: WorkflowRunQueueWorkspaceService,
-    private readonly twentyORMGlobalManager: TwentyORMGlobalManager,
-  ) {}
-
-  @Process(WorkflowHandleStaledRunsPerWorkspaceJob.name)
-  async handle(data: WorkflowHandleStaledRunsPerWorkspaceJobData) {
-    const { workspaceId } = data;
-
-    const workflowRunRepository =
-      await this.twentyORMGlobalManager.getRepositoryForWorkspace(
-        workspaceId,
-        WorkflowRunWorkspaceEntity,
-        { shouldBypassPermissionChecks: true },
-      );
-
-    const oneHourAgo = new Date(Date.now() - 60 * 60 * 1000);
-
-    const staledWorkflowRuns = await workflowRunRepository.find({
-      where: {
-        status: WorkflowRunStatus.ENQUEUED,
-        enqueuedAt: Or(LessThan(oneHourAgo), IsNull()),
-      },
-    });
-
-    if (staledWorkflowRuns.length <= 0) {
-      return;
-    }
-
-    await workflowRunRepository.update(
-      staledWorkflowRuns.map((workflowRun) => workflowRun.id),
-      {
-        enqueuedAt: null,
-        status: WorkflowRunStatus.NOT_STARTED,
-      },
-    );
-
-    await this.workflowRunQueueWorkspaceService.recomputeWorkflowRunQueuedCount(
-      workspaceId,
-    );
-  }
-}

packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-handle-staled-runs.cron.job.ts

@@ -4,16 +4,11 @@ import { WorkspaceActivationStatus } from 'twenty-shared/workspace';
 import { Repository } from 'typeorm';
 
 import { SentryCronMonitor } from 'src/engine/core-modules/cron/sentry-cron-monitor.decorator';
-import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator';
 import { Process } from 'src/engine/core-modules/message-queue/decorators/process.decorator';
 import { Processor } from 'src/engine/core-modules/message-queue/decorators/processor.decorator';
 import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants';
-import { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service';
 import { Workspace } from 'src/engine/core-modules/workspace/workspace.entity';
-import {
-  WorkflowHandleStaledRunsPerWorkspaceJob,
-  type WorkflowHandleStaledRunsPerWorkspaceJobData,
-} from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-handle-staled-runs-per-workspace.job';
+import { WorkflowHandleStaledRunsWorkspaceService } from 'src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-handle-staled-runs.workspace-service';
 
 export const WORKFLOW_HANDLE_STALED_RUNS_CRON_PATTERN = '0 * * * *';
 
@@ -22,8 +17,7 @@ export class WorkflowHandleStaledRunsJob {
   constructor(
     @InjectRepository(Workspace)
     private readonly workspaceRepository: Repository<Workspace>,
-    @InjectMessageQueue(MessageQueue.workflowQueue)
-    private readonly messageQueueService: MessageQueueService,
+    private readonly workflowHandleStaledRunsWorkspaceService: WorkflowHandleStaledRunsWorkspaceService,
   ) {}
 
   @Process(WorkflowHandleStaledRunsJob.name)
@@ -38,13 +32,8 @@ export class WorkflowHandleStaledRunsJob {
       },
     });
 
-    for (const activeWorkspace of activeWorkspaces) {
-      await this.messageQueueService.add<WorkflowHandleStaledRunsPerWorkspaceJobData>(
-        WorkflowHandleStaledRunsPerWorkspaceJob.name,
-        {
-          workspaceId: activeWorkspace.id,
-        },
-      );
-    }
+    await this.workflowHandleStaledRunsWorkspaceService.handleStaledRuns({
+      workspaceIds: activeWorkspaces.map((workspace) => workspace.id),
+    });
   }
 }

packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-run-enqueue-per-workspace.job.ts

@@ -1,107 +0,0 @@
-import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator';
-import { Process } from 'src/engine/core-modules/message-queue/decorators/process.decorator';
-import { Processor } from 'src/engine/core-modules/message-queue/decorators/processor.decorator';
-import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants';
-import { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service';
-import { MetricsService } from 'src/engine/core-modules/metrics/metrics.service';
-import { MetricsKeys } from 'src/engine/core-modules/metrics/types/metrics-keys.type';
-import { TwentyORMGlobalManager } from 'src/engine/twenty-orm/twenty-orm-global.manager';
-import {
-  WorkflowRunStatus,
-  WorkflowRunWorkspaceEntity,
-} from 'src/modules/workflow/common/standard-objects/workflow-run.workspace-entity';
-import {
-  RunWorkflowJob,
-  type RunWorkflowJobData,
-} from 'src/modules/workflow/workflow-runner/jobs/run-workflow.job';
-import { WorkflowRunQueueWorkspaceService } from 'src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-run-queue.workspace-service';
-
-export type WorkflowRunEnqueuePerWorkspaceJobData = {
-  workspaceId: string;
-};
-
-@Processor(MessageQueue.workflowQueue)
-export class WorkflowRunEnqueuePerWorkspaceJob {
-  constructor(
-    private readonly workflowRunQueueWorkspaceService: WorkflowRunQueueWorkspaceService,
-    @InjectMessageQueue(MessageQueue.workflowQueue)
-    private readonly messageQueueService: MessageQueueService,
-    private readonly metricsService: MetricsService,
-    private readonly twentyORMGlobalManager: TwentyORMGlobalManager,
-  ) {}
-
-  @Process(WorkflowRunEnqueuePerWorkspaceJob.name)
-  async handle(data: WorkflowRunEnqueuePerWorkspaceJobData) {
-    const { workspaceId } = data;
-
-    try {
-      const workflowRunRepository =
-        await this.twentyORMGlobalManager.getRepositoryForWorkspace(
-          workspaceId,
-          WorkflowRunWorkspaceEntity,
-          { shouldBypassPermissionChecks: true },
-        );
-
-      const remainingWorkflowRunToEnqueueCount =
-        await this.workflowRunQueueWorkspaceService.getRemainingRunsToEnqueueCountFromDatabase(
-          workspaceId,
-        );
-
-      if (remainingWorkflowRunToEnqueueCount <= 0) {
-        await this.workflowRunQueueWorkspaceService.recomputeWorkflowRunQueuedCount(
-          workspaceId,
-        );
-
-        return;
-      }
-
-      const workflowRunsToEnqueue = await workflowRunRepository.find({
-        where: {
-          status: WorkflowRunStatus.NOT_STARTED,
-        },
-        order: {
-          createdAt: 'ASC',
-        },
-        take: remainingWorkflowRunToEnqueueCount,
-      });
-
-      if (workflowRunsToEnqueue.length <= 0) {
-        await this.workflowRunQueueWorkspaceService.recomputeWorkflowRunQueuedCount(
-          workspaceId,
-        );
-
-        return;
-      }
-
-      const workflowRunIds = workflowRunsToEnqueue.map(
-        (workflowRun: WorkflowRunWorkspaceEntity) => workflowRun.id,
-      );
-
-      await workflowRunRepository.update(workflowRunIds, {
-        enqueuedAt: new Date().toISOString(),
-        status: WorkflowRunStatus.ENQUEUED,
-      });
-
-      for (const workflowRunId of workflowRunIds) {
-        await this.messageQueueService.add<RunWorkflowJobData>(
-          RunWorkflowJob.name,
-          {
-            workflowRunId,
-            workspaceId,
-          },
-        );
-      }
-
-      await this.workflowRunQueueWorkspaceService.recomputeWorkflowRunQueuedCount(
-        workspaceId,
-      );
-    } catch (error) {
-      this.metricsService.incrementCounter({
-        key: MetricsKeys.WorkflowRunFailedToEnqueue,
-        eventId: workspaceId,
-      });
-
-      throw error;
-    }
-  }
-}

packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-run-enqueue.cron.job.ts

@@ -4,26 +4,20 @@ import { WorkspaceActivationStatus } from 'twenty-shared/workspace';
 import { Repository } from 'typeorm';
 
 import { SentryCronMonitor } from 'src/engine/core-modules/cron/sentry-cron-monitor.decorator';
-import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator';
 import { Process } from 'src/engine/core-modules/message-queue/decorators/process.decorator';
 import { Processor } from 'src/engine/core-modules/message-queue/decorators/processor.decorator';
 import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants';
-import { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service';
 import { Workspace } from 'src/engine/core-modules/workspace/workspace.entity';
-import {
-  WorkflowRunEnqueuePerWorkspaceJob,
-  type WorkflowRunEnqueuePerWorkspaceJobData,
-} from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-run-enqueue-per-workspace.job';
+import { WorkflowRunEnqueueWorkspaceService } from 'src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-run-enqueue.workspace-service';
 
-export const WORKFLOW_RUN_ENQUEUE_CRON_PATTERN = '* * * * *';
+export const WORKFLOW_RUN_ENQUEUE_CRON_PATTERN = '*/5 * * * *';
 
 @Processor(MessageQueue.cronQueue)
 export class WorkflowRunEnqueueJob {
   constructor(
     @InjectRepository(Workspace)
     private readonly workspaceRepository: Repository<Workspace>,
-    @InjectMessageQueue(MessageQueue.workflowQueue)
-    private readonly messageQueueService: MessageQueueService,
+    private readonly workflowRunEnqueueWorkspaceService: WorkflowRunEnqueueWorkspaceService,
   ) {}
 
   @Process(WorkflowRunEnqueueJob.name)
@@ -38,13 +32,8 @@ export class WorkflowRunEnqueueJob {
       },
     });
 
-    for (const activeWorkspace of activeWorkspaces) {
-      await this.messageQueueService.add<WorkflowRunEnqueuePerWorkspaceJobData>(
-        WorkflowRunEnqueuePerWorkspaceJob.name,
-        {
-          workspaceId: activeWorkspace.id,
-        },
-      );
-    }
+    await this.workflowRunEnqueueWorkspaceService.enqueueRuns({
+      workspaceIds: activeWorkspaces.map((workspace) => workspace.id),
+    });
   }
 }

packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workflow-run-queue.module.ts

@@ -6,14 +6,16 @@ import { MessageQueueModule } from 'src/engine/core-modules/message-queue/messag
 import { MetricsModule } from 'src/engine/core-modules/metrics/metrics.module';
 import { Workspace } from 'src/engine/core-modules/workspace/workspace.entity';
 import { WorkspaceDataSourceModule } from 'src/engine/workspace-datasource/workspace-datasource.module';
-import { WorkflowCleanWorkflowRunsCommand } from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/command/workflow-clean-workflow-runs.cron.command';
+import { WorkflowHandleStaledRunsCommand } from 'src/modules/workflow/workflow-runner/workflow-run-queue/command/workflow-handle-staled-runs.command';
+import { WorkflowRunEnqueueCommand } from 'src/modules/workflow/workflow-runner/workflow-run-queue/command/workflow-run-enqueue.command';
+import { WorkflowCleanWorkflowRunsCronCommand } from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/command/workflow-clean-workflow-runs.cron.command';
 import { WorkflowHandleStaledRunsCronCommand } from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/command/workflow-handle-staled-runs.cron.command';
 import { WorkflowRunEnqueueCronCommand } from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/command/workflow-run-enqueue.cron.command';
 import { WorkflowCleanWorkflowRunsJob } from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-clean-workflow-runs.cron.job';
-import { WorkflowHandleStaledRunsPerWorkspaceJob } from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-handle-staled-runs-per-workspace.job';
-import { WorkflowHandleStaledRunsJob } from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-handle-staled-runs.job';
-import { WorkflowRunEnqueuePerWorkspaceJob } from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-run-enqueue-per-workspace.job';
-import { WorkflowRunEnqueueJob } from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-run-enqueue.job';
+import { WorkflowHandleStaledRunsJob } from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-handle-staled-runs.cron.job';
+import { WorkflowRunEnqueueJob } from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-run-enqueue.cron.job';
+import { WorkflowHandleStaledRunsWorkspaceService } from 'src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-handle-staled-runs.workspace-service';
+import { WorkflowRunEnqueueWorkspaceService } from 'src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-run-enqueue.workspace-service';
 import { WorkflowRunQueueWorkspaceService } from 'src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-run-queue.workspace-service';
 
 @Module({
@@ -26,20 +28,24 @@ import { WorkflowRunQueueWorkspaceService } from 'src/modules/workflow/workflow-
   ],
   providers: [
     WorkflowRunQueueWorkspaceService,
+    WorkflowRunEnqueueWorkspaceService,
     WorkflowRunEnqueueCronCommand,
+    WorkflowRunEnqueueCommand,
     WorkflowRunEnqueueJob,
-    WorkflowRunEnqueuePerWorkspaceJob,
+    WorkflowHandleStaledRunsWorkspaceService,
     WorkflowHandleStaledRunsCronCommand,
+    WorkflowHandleStaledRunsCommand,
     WorkflowHandleStaledRunsJob,
-    WorkflowHandleStaledRunsPerWorkspaceJob,
     WorkflowCleanWorkflowRunsJob,
-    WorkflowCleanWorkflowRunsCommand,
+    WorkflowCleanWorkflowRunsCronCommand,
   ],
   exports: [
     WorkflowRunQueueWorkspaceService,
     WorkflowRunEnqueueCronCommand,
+    WorkflowRunEnqueueCommand,
     WorkflowHandleStaledRunsCronCommand,
-    WorkflowCleanWorkflowRunsCommand,
+    WorkflowHandleStaledRunsCommand,
+    WorkflowCleanWorkflowRunsCronCommand,
   ],
 })
 export class WorkflowRunQueueModule {}

packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-handle-staled-runs.workspace-service.ts

@@ -0,0 +1,64 @@
+import { Injectable, Logger } from '@nestjs/common';
+
+import { IsNull, LessThan, Or } from 'typeorm';
+
+import { TwentyORMGlobalManager } from 'src/engine/twenty-orm/twenty-orm-global.manager';
+import {
+  WorkflowRunStatus,
+  WorkflowRunWorkspaceEntity,
+} from 'src/modules/workflow/common/standard-objects/workflow-run.workspace-entity';
+import { WorkflowRunQueueWorkspaceService } from 'src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-run-queue.workspace-service';
+
+@Injectable()
+export class WorkflowHandleStaledRunsWorkspaceService {
+  private readonly logger = new Logger(
+    WorkflowHandleStaledRunsWorkspaceService.name,
+  );
+  constructor(
+    private readonly twentyORMGlobalManager: TwentyORMGlobalManager,
+    private readonly workflowRunQueueWorkspaceService: WorkflowRunQueueWorkspaceService,
+  ) {}
+
+  async handleStaledRuns({ workspaceIds }: { workspaceIds: string[] }) {
+    for (const workspaceId of workspaceIds) {
+      try {
+        const workflowRunRepository =
+          await this.twentyORMGlobalManager.getRepositoryForWorkspace(
+            workspaceId,
+            WorkflowRunWorkspaceEntity,
+            { shouldBypassPermissionChecks: true },
+          );
+
+        const oneHourAgo = new Date(Date.now() - 60 * 60 * 1000);
+
+        const staledWorkflowRuns = await workflowRunRepository.find({
+          where: {
+            status: WorkflowRunStatus.ENQUEUED,
+            enqueuedAt: Or(LessThan(oneHourAgo), IsNull()),
+          },
+        });
+
+        if (staledWorkflowRuns.length <= 0) {
+          continue;
+        }
+
+        await workflowRunRepository.update(
+          staledWorkflowRuns.map((workflowRun) => workflowRun.id),
+          {
+            enqueuedAt: null,
+            status: WorkflowRunStatus.NOT_STARTED,
+          },
+        );
+
+        await this.workflowRunQueueWorkspaceService.recomputeWorkflowRunQueuedCount(
+          workspaceId,
+        );
+      } catch (error) {
+        this.logger.error(
+          `Failed to handle staled runs for workspace: ${workspaceId}`,
+          error,
+        );
+      }
+    }
+  }
+}

packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-run-enqueue.workspace-service.ts

@@ -0,0 +1,106 @@
+import { Injectable, Logger } from '@nestjs/common';
+
+import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator';
+import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants';
+import { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service';
+import { MetricsService } from 'src/engine/core-modules/metrics/metrics.service';
+import { MetricsKeys } from 'src/engine/core-modules/metrics/types/metrics-keys.type';
+import { TwentyORMGlobalManager } from 'src/engine/twenty-orm/twenty-orm-global.manager';
+import {
+  WorkflowRunStatus,
+  WorkflowRunWorkspaceEntity,
+} from 'src/modules/workflow/common/standard-objects/workflow-run.workspace-entity';
+import {
+  RunWorkflowJob,
+  RunWorkflowJobData,
+} from 'src/modules/workflow/workflow-runner/jobs/run-workflow.job';
+import { WorkflowRunQueueWorkspaceService } from 'src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-run-queue.workspace-service';
+
+@Injectable()
+export class WorkflowRunEnqueueWorkspaceService {
+  private readonly logger = new Logger(WorkflowRunEnqueueWorkspaceService.name);
+  constructor(
+    private readonly workflowRunQueueWorkspaceService: WorkflowRunQueueWorkspaceService,
+    private readonly twentyORMGlobalManager: TwentyORMGlobalManager,
+    @InjectMessageQueue(MessageQueue.workflowQueue)
+    private readonly messageQueueService: MessageQueueService,
+    private readonly metricsService: MetricsService,
+  ) {}
+
+  async enqueueRuns({ workspaceIds }: { workspaceIds: string[] }) {
+    for (const workspaceId of workspaceIds) {
+      try {
+        const workflowRunRepository =
+          await this.twentyORMGlobalManager.getRepositoryForWorkspace(
+            workspaceId,
+            WorkflowRunWorkspaceEntity,
+            { shouldBypassPermissionChecks: true },
+          );
+
+        const remainingWorkflowRunToEnqueueCount =
+          await this.workflowRunQueueWorkspaceService.getRemainingRunsToEnqueueCountFromDatabase(
+            workspaceId,
+          );
+
+        if (remainingWorkflowRunToEnqueueCount <= 0) {
+          await this.workflowRunQueueWorkspaceService.recomputeWorkflowRunQueuedCount(
+            workspaceId,
+          );
+
+          continue;
+        }
+
+        const workflowRunsToEnqueue = await workflowRunRepository.find({
+          where: {
+            status: WorkflowRunStatus.NOT_STARTED,
+          },
+          order: {
+            createdAt: 'ASC',
+          },
+          take: remainingWorkflowRunToEnqueueCount,
+        });
+
+        if (workflowRunsToEnqueue.length <= 0) {
+          await this.workflowRunQueueWorkspaceService.recomputeWorkflowRunQueuedCount(
+            workspaceId,
+          );
+
+          continue;
+        }
+
+        const workflowRunIds = workflowRunsToEnqueue.map(
+          (workflowRun: WorkflowRunWorkspaceEntity) => workflowRun.id,
+        );
+
+        await workflowRunRepository.update(workflowRunIds, {
+          enqueuedAt: new Date().toISOString(),
+          status: WorkflowRunStatus.ENQUEUED,
+        });
+
+        for (const workflowRunId of workflowRunIds) {
+          await this.messageQueueService.add<RunWorkflowJobData>(
+            RunWorkflowJob.name,
+            {
+              workflowRunId,
+              workspaceId,
+            },
+          );
+        }
+
+        await this.workflowRunQueueWorkspaceService.recomputeWorkflowRunQueuedCount(
+          workspaceId,
+        );
+      } catch (error) {
+        this.metricsService.incrementCounter({
+          key: MetricsKeys.WorkflowRunFailedToEnqueue,
+          eventId: workspaceId,
+        });
+
+        this.logger.error(
+          `Failed to enqueue workflow runs for workspace: ${workspaceId}`,
+          error,
+        );
+      }
+    }
+  }
+}