diff --git a/packages/twenty-server/src/database/commands/cron-register-all.command.ts b/packages/twenty-server/src/database/commands/cron-register-all.command.ts index 8dd5554c8c..1623078046 100644 --- a/packages/twenty-server/src/database/commands/cron-register-all.command.ts +++ b/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(); diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/command/workflow-handle-staled-runs.command.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/command/workflow-handle-staled-runs.command.ts new file mode 100644 index 0000000000..1c6db06667 --- /dev/null +++ b/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 { + const { workspaceIds } = options; + + await this.workflowHandleStaledRunsWorkspaceService.handleStaledRuns({ + workspaceIds, + }); + } +} diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/command/workflow-run-enqueue.command.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/command/workflow-run-enqueue.command.ts new file mode 100644 index 0000000000..b0400af45f --- /dev/null +++ b/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 { + const { workspaceIds } = options; + + await this.workflowRunEnqueueWorkspaceService.enqueueRuns({ + workspaceIds, + }); + } +} diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/command/workflow-clean-workflow-runs.cron.command.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/command/workflow-clean-workflow-runs.cron.command.ts index b203a923fd..63f955611f 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/command/workflow-clean-workflow-runs.cron.command.ts +++ b/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, diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/command/workflow-handle-staled-runs.cron.command.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/command/workflow-handle-staled-runs.cron.command.ts index e950e66d6c..cb55bca6b6 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/command/workflow-handle-staled-runs.cron.command.ts +++ b/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', diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/command/workflow-run-enqueue.cron.command.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/command/workflow-run-enqueue.cron.command.ts index 7542fc76da..eb0726ff8b 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/command/workflow-run-enqueue.cron.command.ts +++ b/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', diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-handle-staled-runs-per-workspace.job.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-handle-staled-runs-per-workspace.job.ts deleted file mode 100644 index 32b7507a94..0000000000 --- a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-handle-staled-runs-per-workspace.job.ts +++ /dev/null @@ -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, - ); - } -} diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-handle-staled-runs.job.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-handle-staled-runs.cron.job.ts similarity index 59% rename from packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-handle-staled-runs.job.ts rename to packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-handle-staled-runs.cron.job.ts index f8c47e38df..48a4f50643 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-handle-staled-runs.job.ts +++ b/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, - @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( - WorkflowHandleStaledRunsPerWorkspaceJob.name, - { - workspaceId: activeWorkspace.id, - }, - ); - } + await this.workflowHandleStaledRunsWorkspaceService.handleStaledRuns({ + workspaceIds: activeWorkspaces.map((workspace) => workspace.id), + }); } } diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-run-enqueue-per-workspace.job.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-run-enqueue-per-workspace.job.ts deleted file mode 100644 index e10f1848f4..0000000000 --- a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-run-enqueue-per-workspace.job.ts +++ /dev/null @@ -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( - RunWorkflowJob.name, - { - workflowRunId, - workspaceId, - }, - ); - } - - await this.workflowRunQueueWorkspaceService.recomputeWorkflowRunQueuedCount( - workspaceId, - ); - } catch (error) { - this.metricsService.incrementCounter({ - key: MetricsKeys.WorkflowRunFailedToEnqueue, - eventId: workspaceId, - }); - - throw error; - } - } -} diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-run-enqueue.job.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-run-enqueue.cron.job.ts similarity index 56% rename from packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-run-enqueue.job.ts rename to packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-run-enqueue.cron.job.ts index e30c4114af..b95d5caaf0 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-run-enqueue.job.ts +++ b/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, - @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( - WorkflowRunEnqueuePerWorkspaceJob.name, - { - workspaceId: activeWorkspace.id, - }, - ); - } + await this.workflowRunEnqueueWorkspaceService.enqueueRuns({ + workspaceIds: activeWorkspaces.map((workspace) => workspace.id), + }); } } diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workflow-run-queue.module.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workflow-run-queue.module.ts index ca925d6ec0..44303fd211 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workflow-run-queue.module.ts +++ b/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 {} diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-handle-staled-runs.workspace-service.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-handle-staled-runs.workspace-service.ts new file mode 100644 index 0000000000..7ee7b7292d --- /dev/null +++ b/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, + ); + } + } + } +} diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-run-enqueue.workspace-service.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-run-enqueue.workspace-service.ts new file mode 100644 index 0000000000..550acb6fb5 --- /dev/null +++ b/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( + 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, + ); + } + } + } +}