From cee647435cd407a47323bb8439969faec4e60bd0 Mon Sep 17 00:00:00 2001 From: Thomas Trompette Date: Tue, 12 Aug 2025 15:54:58 +0200 Subject: [PATCH] Improve workflow queue cron reliability (#13818) Improve workflow enqueue cron : instead of relying on the cache to know how many workflows we can enqueue, query the DB. Then set the cache and process the not started workflows. Also adding a second cron that will look for workflows enqueued one hour ago or more and put these back in the not started status. This will allow the first cron to start these again. --- .../commands/cron-register-all.command.ts | 23 ++- .../commands/database-command.module.ts | 7 +- .../constants/standard-field-ids.ts | 1 + .../workflow-run.workspace-entity.ts | 10 ++ .../workflow-runner/jobs/run-workflow.job.ts | 4 +- .../workflow-run-queue-throttle-limit.ts | 1 + ...orkflow-handle-staled-runs.cron.command.ts | 34 +++++ ...s => workflow-run-enqueue.cron.command.ts} | 4 +- ...ow-handle-staled-runs-per-workspace.job.ts | 60 ++++++++ .../jobs/workflow-handle-staled-runs.job.ts | 50 +++++++ .../workflow-run-enqueue-per-workspace.job.ts | 107 ++++++++++++++ .../jobs/workflow-run-enqueue.cron.job.ts | 138 ------------------ .../cron/jobs/workflow-run-enqueue.job.ts | 50 +++++++ .../workflow-run-queue.module.ts | 24 ++- .../workflow-run-queue.workspace-service.ts | 68 ++++++++- .../workflow-run.workspace-service.ts | 21 ++- .../workflow-runner.workspace-service.ts | 4 +- .../automated-trigger.module.ts | 8 +- ... => workflow-cron-trigger.cron.command.ts} | 2 +- 19 files changed, 443 insertions(+), 173 deletions(-) create mode 100644 packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/constants/workflow-run-queue-throttle-limit.ts create mode 100644 packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/command/workflow-handle-staled-runs.cron.command.ts rename packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/command/{cron-workflow-run-enqueue.cron.command.ts => workflow-run-enqueue.cron.command.ts} (90%) create mode 100644 packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-handle-staled-runs-per-workspace.job.ts create mode 100644 packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-handle-staled-runs.job.ts create mode 100644 packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-run-enqueue-per-workspace.job.ts delete mode 100644 packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-run-enqueue.cron.job.ts create mode 100644 packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-run-enqueue.job.ts rename packages/twenty-server/src/modules/workflow/workflow-trigger/automated-trigger/crons/commands/{cron-trigger.cron.command.ts => workflow-cron-trigger.cron.command.ts} (94%) 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 4f13794235..09ce49886d 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 @@ -2,6 +2,7 @@ import { Logger } from '@nestjs/common'; import { Command, CommandRunner } from 'nest-commander'; +import { CheckCustomDomainValidRecordsCronCommand } from 'src/engine/core-modules/domain-manager/crons/commands/check-custom-domain-valid-records.cron.command'; import { CleanupOrphanedFilesCronCommand } from 'src/engine/core-modules/file/crons/commands/cleanup-orphaned-files.cron.command'; import { CalendarEventListFetchCronCommand } from 'src/modules/calendar/calendar-event-import-manager/crons/commands/calendar-event-list-fetch.cron.command'; import { CalendarEventsImportCronCommand } from 'src/modules/calendar/calendar-event-import-manager/crons/commands/calendar-import.cron.command'; @@ -9,8 +10,9 @@ 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 { CronTriggerCronCommand } from 'src/modules/workflow/workflow-trigger/automated-trigger/crons/commands/cron-trigger.cron.command'; -import { CheckCustomDomainValidRecordsCronCommand } from 'src/engine/core-modules/domain-manager/crons/commands/check-custom-domain-valid-records.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'; @Command({ name: 'cron:register:all', @@ -26,9 +28,11 @@ export class CronRegisterAllCommand extends CommandRunner { private readonly calendarEventListFetchCronCommand: CalendarEventListFetchCronCommand, private readonly calendarEventsImportCronCommand: CalendarEventsImportCronCommand, private readonly calendarOngoingStaleCronCommand: CalendarOngoingStaleCronCommand, - private readonly cronTriggerCronCommand: CronTriggerCronCommand, + private readonly workflowCronTriggerCronCommand: WorkflowCronTriggerCronCommand, private readonly cleanupOrphanedFilesCronCommand: CleanupOrphanedFilesCronCommand, private readonly checkCustomDomainValidRecordsCronCommand: CheckCustomDomainValidRecordsCronCommand, + private readonly workflowRunEnqueueCronCommand: WorkflowRunEnqueueCronCommand, + private readonly workflowHandleStaledRunsCronCommand: WorkflowHandleStaledRunsCronCommand, ) { super(); } @@ -61,7 +65,6 @@ export class CronRegisterAllCommand extends CommandRunner { name: 'CalendarOngoingStale', command: this.calendarOngoingStaleCronCommand, }, - { name: 'CronTrigger', command: this.cronTriggerCronCommand }, { name: 'CleanupOrphanedFiles', command: this.cleanupOrphanedFilesCronCommand, @@ -70,6 +73,18 @@ export class CronRegisterAllCommand extends CommandRunner { name: 'CheckCustomDomainValidRecords', command: this.checkCustomDomainValidRecordsCronCommand, }, + { + name: 'WorkflowCronTrigger', + command: this.workflowCronTriggerCronCommand, + }, + { + name: 'WorkflowRunEnqueue', + command: this.workflowRunEnqueueCronCommand, + }, + { + name: 'WorkflowHandleStaledRuns', + command: this.workflowHandleStaledRunsCronCommand, + }, ]; let successCount = 0; diff --git a/packages/twenty-server/src/database/commands/database-command.module.ts b/packages/twenty-server/src/database/commands/database-command.module.ts index 16e43f875c..464be1473b 100644 --- a/packages/twenty-server/src/database/commands/database-command.module.ts +++ b/packages/twenty-server/src/database/commands/database-command.module.ts @@ -7,6 +7,7 @@ import { UpgradeVersionCommandModule } from 'src/database/commands/upgrade-versi import { MigrateViewsToCoreCommand } from 'src/database/commands/views-migration/migrate-views-to-core.command'; import { TypeORMModule } from 'src/database/typeorm/typeorm.module'; import { ApiKeyModule } from 'src/engine/core-modules/api-key/api-key.module'; +import { DomainManagerModule } from 'src/engine/core-modules/domain-manager/domain-manager.module'; import { FeatureFlagModule } from 'src/engine/core-modules/feature-flag/feature-flag.module'; import { FileModule } from 'src/engine/core-modules/file/file.module'; import { Workspace } from 'src/engine/core-modules/workspace/workspace.entity'; @@ -18,24 +19,22 @@ import { DevSeederModule } from 'src/engine/workspace-manager/dev-seeder/dev-see import { WorkspaceManagerModule } from 'src/engine/workspace-manager/workspace-manager.module'; import { CalendarEventImportManagerModule } from 'src/modules/calendar/calendar-event-import-manager/calendar-event-import-manager.module'; import { MessagingImportManagerModule } from 'src/modules/messaging/message-import-manager/messaging-import-manager.module'; +import { WorkflowRunQueueModule } from 'src/modules/workflow/workflow-runner/workflow-run-queue/workflow-run-queue.module'; import { AutomatedTriggerModule } from 'src/modules/workflow/workflow-trigger/automated-trigger/automated-trigger.module'; -import { DomainManagerModule } from 'src/engine/core-modules/domain-manager/domain-manager.module'; import { DataSeedWorkspaceCommand } from './data-seed-dev-workspace.command'; @Module({ imports: [ UpgradeVersionCommandModule, - TypeOrmModule.forFeature([Workspace], 'core'), - // Cron command dependencies MessagingImportManagerModule, CalendarEventImportManagerModule, AutomatedTriggerModule, FileModule, DomainManagerModule, - + WorkflowRunQueueModule, // Data seeding dependencies TypeORMModule, FieldMetadataModule, diff --git a/packages/twenty-server/src/engine/workspace-manager/workspace-sync-metadata/constants/standard-field-ids.ts b/packages/twenty-server/src/engine/workspace-manager/workspace-sync-metadata/constants/standard-field-ids.ts index e0b887ec39..f2f37d4d16 100644 --- a/packages/twenty-server/src/engine/workspace-manager/workspace-sync-metadata/constants/standard-field-ids.ts +++ b/packages/twenty-server/src/engine/workspace-manager/workspace-sync-metadata/constants/standard-field-ids.ts @@ -466,6 +466,7 @@ export const WORKFLOW_RUN_STANDARD_FIELD_IDS = { name: '20202020-b840-4253-aef9-4e5013694587', workflowVersion: '20202020-2f52-4ba8-8dc4-d0d6adb9578d', workflow: '20202020-8c57-4e7f-84f5-f373f68e1b82', + enqueuedAt: '20202020-f1e3-4de1-a461-b5c4fdbc861d', startedAt: '20202020-a234-4e2d-bd15-85bcea6bb183', endedAt: '20202020-e1c1-4b6b-bbbd-b2beaf2e159e', status: '20202020-6b3e-4f9c-8c2b-2e5b8e6d6f3b', diff --git a/packages/twenty-server/src/modules/workflow/common/standard-objects/workflow-run.workspace-entity.ts b/packages/twenty-server/src/modules/workflow/common/standard-objects/workflow-run.workspace-entity.ts index 2a73f4555b..50433a3e3d 100644 --- a/packages/twenty-server/src/modules/workflow/common/standard-objects/workflow-run.workspace-entity.ts +++ b/packages/twenty-server/src/modules/workflow/common/standard-objects/workflow-run.workspace-entity.ts @@ -90,6 +90,16 @@ export class WorkflowRunWorkspaceEntity extends BaseWorkspaceEntity { }) name: string; + @WorkspaceField({ + standardId: WORKFLOW_RUN_STANDARD_FIELD_IDS.enqueuedAt, + type: FieldMetadataType.DATE_TIME, + label: msg`Workflow run enqueued at`, + description: msg`Workflow run enqueued at`, + icon: 'IconHistory', + }) + @WorkspaceIsNullable() + enqueuedAt: Date | null; + @WorkspaceField({ standardId: WORKFLOW_RUN_STANDARD_FIELD_IDS.startedAt, type: FieldMetadataType.DATE_TIME, diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/jobs/run-workflow.job.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/jobs/run-workflow.job.ts index f6b4fde97a..782e82ace8 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-runner/jobs/run-workflow.job.ts +++ b/packages/twenty-server/src/modules/workflow/workflow-runner/jobs/run-workflow.job.ts @@ -2,6 +2,8 @@ import { Scope } from '@nestjs/common'; import { isDefined } from 'twenty-shared/utils'; +import { FeatureFlagKey } from 'src/engine/core-modules/feature-flag/enums/feature-flag-key.enum'; +import { FeatureFlagService } from 'src/engine/core-modules/feature-flag/services/feature-flag.service'; 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'; @@ -20,8 +22,6 @@ import { getRootSteps } from 'src/modules/workflow/workflow-runner/utils/get-roo import { WorkflowRunQueueWorkspaceService } from 'src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-run-queue.workspace-service'; import { WorkflowRunWorkspaceService } from 'src/modules/workflow/workflow-runner/workflow-run/workflow-run.workspace-service'; import { WorkflowTriggerType } from 'src/modules/workflow/workflow-trigger/types/workflow-trigger.type'; -import { FeatureFlagKey } from 'src/engine/core-modules/feature-flag/enums/feature-flag-key.enum'; -import { FeatureFlagService } from 'src/engine/core-modules/feature-flag/services/feature-flag.service'; export type RunWorkflowJobData = { workspaceId: string; diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/constants/workflow-run-queue-throttle-limit.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/constants/workflow-run-queue-throttle-limit.ts new file mode 100644 index 0000000000..9b8ac7e683 --- /dev/null +++ b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/constants/workflow-run-queue-throttle-limit.ts @@ -0,0 +1 @@ +export const WORKFLOW_RUN_QUEUE_THROTTLE_LIMIT = 100; 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 new file mode 100644 index 0000000000..e950e66d6c --- /dev/null +++ b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/command/workflow-handle-staled-runs.cron.command.ts @@ -0,0 +1,34 @@ +import { Command, CommandRunner } from 'nest-commander'; + +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 { + WORKFLOW_HANDLE_STALED_RUNS_CRON_PATTERN, + WorkflowHandleStaledRunsJob, +} from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-handle-staled-runs.job'; + +@Command({ + name: 'cron:workflow:handle-staled-runs', + description: 'Handles staled workflow runs', +}) +export class WorkflowHandleStaledRunsCronCommand extends CommandRunner { + constructor( + @InjectMessageQueue(MessageQueue.cronQueue) + private readonly messageQueueService: MessageQueueService, + ) { + super(); + } + + async run(): Promise { + await this.messageQueueService.addCron({ + jobName: WorkflowHandleStaledRunsJob.name, + data: undefined, + options: { + repeat: { + pattern: WORKFLOW_HANDLE_STALED_RUNS_CRON_PATTERN, + }, + }, + }); + } +} diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/command/cron-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 similarity index 90% rename from packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/command/cron-workflow-run-enqueue.cron.command.ts rename to packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/command/workflow-run-enqueue.cron.command.ts index 0a3fcd1448..7542fc76da 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/command/cron-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,13 +6,13 @@ 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.cron.job'; +} from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-run-enqueue.job'; @Command({ name: 'cron:workflow:enqueue-awaiting-workflow-run', description: 'Enqueues awaiting workflow runs', }) -export class CronWorkflowRunEnqueueCommand extends CommandRunner { +export class WorkflowRunEnqueueCronCommand 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/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 new file mode 100644 index 0000000000..983cfe313b --- /dev/null +++ b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-handle-staled-runs-per-workspace.job.ts @@ -0,0 +1,60 @@ +import { LessThan } 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: LessThan(oneHourAgo), + }, + }); + + 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.job.ts new file mode 100644 index 0000000000..6820aa4bea --- /dev/null +++ b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-handle-staled-runs.job.ts @@ -0,0 +1,50 @@ +import { InjectRepository } from '@nestjs/typeorm'; + +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'; + +export const WORKFLOW_HANDLE_STALED_RUNS_CRON_PATTERN = '0 * * * *'; + +@Processor(MessageQueue.cronQueue) +export class WorkflowHandleStaledRunsJob { + constructor( + @InjectRepository(Workspace, 'core') + private readonly workspaceRepository: Repository, + @InjectMessageQueue(MessageQueue.workflowQueue) + private readonly messageQueueService: MessageQueueService, + ) {} + + @Process(WorkflowHandleStaledRunsJob.name) + @SentryCronMonitor( + WorkflowHandleStaledRunsJob.name, + WORKFLOW_HANDLE_STALED_RUNS_CRON_PATTERN, + ) + async handle() { + const activeWorkspaces = await this.workspaceRepository.find({ + where: { + activationStatus: WorkspaceActivationStatus.ACTIVE, + }, + }); + + for (const activeWorkspace of activeWorkspaces) { + await this.messageQueueService.add( + WorkflowHandleStaledRunsPerWorkspaceJob.name, + { + workspaceId: activeWorkspace.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 new file mode 100644 index 0000000000..e10f1848f4 --- /dev/null +++ b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-run-enqueue-per-workspace.job.ts @@ -0,0 +1,107 @@ +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.cron.job.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-run-enqueue.cron.job.ts deleted file mode 100644 index d4607ece53..0000000000 --- a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-run-enqueue.cron.job.ts +++ /dev/null @@ -1,138 +0,0 @@ -import { Logger } from '@nestjs/common'; -import { InjectRepository } from '@nestjs/typeorm'; - -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 { MetricsService } from 'src/engine/core-modules/metrics/metrics.service'; -import { MetricsKeys } from 'src/engine/core-modules/metrics/types/metrics-keys.type'; -import { Workspace } from 'src/engine/core-modules/workspace/workspace.entity'; -import { TwentyORMGlobalManager } from 'src/engine/twenty-orm/twenty-orm-global.manager'; -import { getWorkspaceSchemaName } from 'src/engine/workspace-datasource/utils/get-workspace-schema-name.util'; -import { WorkspaceDataSourceService } from 'src/engine/workspace-datasource/workspace-datasource.service'; -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 const WORKFLOW_RUN_ENQUEUE_CRON_PATTERN = '* * * * *'; - -@Processor(MessageQueue.cronQueue) -export class WorkflowRunEnqueueJob { - private readonly logger = new Logger(WorkflowRunEnqueueJob.name); - - constructor( - @InjectRepository(Workspace, 'core') - private readonly workspaceRepository: Repository, - private readonly workflowRunQueueWorkspaceService: WorkflowRunQueueWorkspaceService, - private readonly workspaceDataSourceService: WorkspaceDataSourceService, - @InjectMessageQueue(MessageQueue.workflowQueue) - private readonly messageQueueService: MessageQueueService, - private readonly metricsService: MetricsService, - private readonly twentyORMGlobalManager: TwentyORMGlobalManager, - ) {} - - @Process(WorkflowRunEnqueueJob.name) - @SentryCronMonitor( - WorkflowRunEnqueueJob.name, - WORKFLOW_RUN_ENQUEUE_CRON_PATTERN, - ) - async handle() { - const activeWorkspaces = await this.workspaceRepository.find({ - where: { - activationStatus: WorkspaceActivationStatus.ACTIVE, - }, - }); - - const mainDataSource = - await this.workspaceDataSourceService.connectToMainDataSource(); - - for (const activeWorkspace of activeWorkspaces) { - let enqueuedWorkflowRunCount = 0; - - try { - const remainingWorkflowRunCount = - await this.workflowRunQueueWorkspaceService.getRemainingRunsToEnqueueCount( - activeWorkspace.id, - ); - - if (remainingWorkflowRunCount <= 0) { - continue; - } - - const schemaName = getWorkspaceSchemaName(activeWorkspace.id); - - // Using raw query to avoid storing repository in cache - const workflowRuns = await mainDataSource.query( - `SELECT id FROM ${schemaName}."workflowRun" WHERE status = '${WorkflowRunStatus.NOT_STARTED}' ORDER BY "createdAt" ASC`, - ); - - const workflowRunsToEnqueueCount = Math.min( - remainingWorkflowRunCount, - workflowRuns.length, - ); - - if (workflowRunsToEnqueueCount <= 0) { - continue; - } - - const workflowRunRepository = - await this.twentyORMGlobalManager.getRepositoryForWorkspace( - activeWorkspace.id, - WorkflowRunWorkspaceEntity, - { shouldBypassPermissionChecks: true }, - ); - - for ( - let runIndex = 0; - runIndex < workflowRunsToEnqueueCount; - runIndex++ - ) { - const workflowRunId = workflowRuns[runIndex].id; - - await this.messageQueueService.add( - RunWorkflowJob.name, - { - workflowRunId, - workspaceId: activeWorkspace.id, - }, - ); - - await workflowRunRepository.update(workflowRunId, { - status: WorkflowRunStatus.ENQUEUED, - }); - - enqueuedWorkflowRunCount++; - } - } catch (error) { - this.logger.error( - `Error enqueuing workflow runs for workspace ${activeWorkspace.id}`, - error, - ); - - this.metricsService.incrementCounter({ - key: MetricsKeys.WorkflowRunFailedToEnqueue, - eventId: activeWorkspace.id, - }); - } finally { - if (enqueuedWorkflowRunCount > 0) { - await this.workflowRunQueueWorkspaceService.increaseWorkflowRunQueuedCount( - activeWorkspace.id, - enqueuedWorkflowRunCount, - ); - } - } - } - } -} 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.job.ts new file mode 100644 index 0000000000..605943d370 --- /dev/null +++ b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-run-enqueue.job.ts @@ -0,0 +1,50 @@ +import { InjectRepository } from '@nestjs/typeorm'; + +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'; + +export const WORKFLOW_RUN_ENQUEUE_CRON_PATTERN = '* * * * *'; + +@Processor(MessageQueue.cronQueue) +export class WorkflowRunEnqueueJob { + constructor( + @InjectRepository(Workspace, 'core') + private readonly workspaceRepository: Repository, + @InjectMessageQueue(MessageQueue.workflowQueue) + private readonly messageQueueService: MessageQueueService, + ) {} + + @Process(WorkflowRunEnqueueJob.name) + @SentryCronMonitor( + WorkflowRunEnqueueJob.name, + WORKFLOW_RUN_ENQUEUE_CRON_PATTERN, + ) + async handle() { + const activeWorkspaces = await this.workspaceRepository.find({ + where: { + activationStatus: WorkspaceActivationStatus.ACTIVE, + }, + }); + + for (const activeWorkspace of activeWorkspaces) { + await this.messageQueueService.add( + WorkflowRunEnqueuePerWorkspaceJob.name, + { + workspaceId: activeWorkspace.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 298f0fa4e6..78e76494e6 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,11 +6,15 @@ 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 { CronWorkflowRunEnqueueCommand } from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/command/cron-workflow-run-enqueue.cron.command'; -import { WorkflowRunEnqueueJob } from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-run-enqueue.cron.job'; -import { WorkflowRunQueueWorkspaceService } from 'src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-run-queue.workspace-service'; -import { CleanWorkflowRunsJob } from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/clean-workflow-runs.cron.job'; import { CronCleanWorkflowRunsCommand } from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/command/cron-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 { CleanWorkflowRunsJob } from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/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 { WorkflowRunQueueWorkspaceService } from 'src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-run-queue.workspace-service'; @Module({ imports: [ @@ -22,11 +26,19 @@ import { CronCleanWorkflowRunsCommand } from 'src/modules/workflow/workflow-runn ], providers: [ WorkflowRunQueueWorkspaceService, + WorkflowRunEnqueueCronCommand, WorkflowRunEnqueueJob, + WorkflowRunEnqueuePerWorkspaceJob, + WorkflowHandleStaledRunsCronCommand, + WorkflowHandleStaledRunsJob, + WorkflowHandleStaledRunsPerWorkspaceJob, CleanWorkflowRunsJob, - CronWorkflowRunEnqueueCommand, CronCleanWorkflowRunsCommand, ], - exports: [WorkflowRunQueueWorkspaceService], + exports: [ + WorkflowRunQueueWorkspaceService, + WorkflowRunEnqueueCronCommand, + WorkflowHandleStaledRunsCronCommand, + ], }) export class WorkflowRunQueueModule {} diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-run-queue.workspace-service.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-run-queue.workspace-service.ts index d3e358bd3f..758721bade 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-run-queue.workspace-service.ts +++ b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-run-queue.workspace-service.ts @@ -1,17 +1,24 @@ import { Injectable } from '@nestjs/common'; +import { In } from 'typeorm'; + import { InjectCacheStorage } from 'src/engine/core-modules/cache-storage/decorators/cache-storage.decorator'; import { CacheStorageService } from 'src/engine/core-modules/cache-storage/services/cache-storage.service'; import { CacheStorageNamespace } from 'src/engine/core-modules/cache-storage/types/cache-storage-namespace.enum'; +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 { WORKFLOW_RUN_QUEUE_THROTTLE_LIMIT } from 'src/modules/workflow/workflow-runner/workflow-run-queue/constants/workflow-run-queue-throttle-limit'; import { getWorkflowRunQueuedCountCacheKey } from 'src/modules/workflow/workflow-runner/workflow-run-queue/utils/get-cache-workflow-run-count-key.util'; @Injectable() export class WorkflowRunQueueWorkspaceService { - private readonly WORKFLOW_RUN_QUEUE_THROTTLE_LIMIT = 100; - constructor( @InjectCacheStorage(CacheStorageNamespace.ModuleWorkflow) private readonly cacheStorage: CacheStorageService, + private readonly twentyORMGlobalManager: TwentyORMGlobalManager, ) {} async increaseWorkflowRunQueuedCount( @@ -40,11 +47,64 @@ export class WorkflowRunQueueWorkspaceService { ); } - async getRemainingRunsToEnqueueCount(workspaceId: string): Promise { + async recomputeWorkflowRunQueuedCount(workspaceId: string): Promise { + const workflowRunRepository = + await this.twentyORMGlobalManager.getRepositoryForWorkspace( + workspaceId, + WorkflowRunWorkspaceEntity, + { shouldBypassPermissionChecks: true }, + ); + + const currentlyEnqueuedWorkflowRunCount = await workflowRunRepository.count( + { + where: { + status: In([WorkflowRunStatus.ENQUEUED, WorkflowRunStatus.RUNNING]), + }, + }, + ); + + await this.setWorkflowRunQueuedCount( + workspaceId, + currentlyEnqueuedWorkflowRunCount, + ); + } + + async getRemainingRunsToEnqueueCountFromCache( + workspaceId: string, + ): Promise { const currentCount = await this.getCurrentWorkflowRunQueuedCount(workspaceId); - return this.WORKFLOW_RUN_QUEUE_THROTTLE_LIMIT - currentCount; + return WORKFLOW_RUN_QUEUE_THROTTLE_LIMIT - currentCount; + } + + async getRemainingRunsToEnqueueCountFromDatabase( + workspaceId: string, + ): Promise { + const workflowRunRepository = + await this.twentyORMGlobalManager.getRepositoryForWorkspace( + workspaceId, + WorkflowRunWorkspaceEntity, + { shouldBypassPermissionChecks: true }, + ); + + const currentCount = await workflowRunRepository.count({ + where: { + status: In([WorkflowRunStatus.ENQUEUED, WorkflowRunStatus.RUNNING]), + }, + }); + + return WORKFLOW_RUN_QUEUE_THROTTLE_LIMIT - currentCount; + } + + private async setWorkflowRunQueuedCount( + workspaceId: string, + count: number, + ): Promise { + await this.cacheStorage.set( + getWorkflowRunQueuedCountCacheKey(workspaceId), + count, + ); } private async getCurrentWorkflowRunQueuedCount( diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run/workflow-run.workspace-service.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run/workflow-run.workspace-service.ts index 4abf2ad030..9a41143cd3 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run/workflow-run.workspace-service.ts +++ b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run/workflow-run.workspace-service.ts @@ -1,7 +1,7 @@ import { Injectable } from '@nestjs/common'; import { isDefined } from 'twenty-shared/utils'; -import { StepStatus, WorkflowRunStepInfo } from 'twenty-shared/workflow'; +import { StepStatus, type WorkflowRunStepInfo } from 'twenty-shared/workflow'; import { type QueryDeepPartialEntity } from 'typeorm/query-builder/QueryPartialEntity'; import { v4 } from 'uuid'; @@ -44,9 +44,9 @@ export class WorkflowRunWorkspaceService { }: { workflowVersionId: string; createdBy: ActorMetadata; - workflowRunId?: string; status: WorkflowRunStatus.NOT_STARTED | WorkflowRunStatus.ENQUEUED; triggerPayload: object; + workflowRunId?: string; }) { const workspaceId = this.scopedWorkspaceContextFactory.create()?.workspaceId; @@ -124,6 +124,8 @@ export class WorkflowRunWorkspaceService { status, position, state: initState, + enqueuedAt: + status === WorkflowRunStatus.ENQUEUED ? new Date().toISOString() : null, }); await workflowRunRepository.insert(workflowRun); @@ -144,14 +146,21 @@ export class WorkflowRunWorkspaceService { workspaceId, }); + if ( + workflowRunToUpdate.status === WorkflowRunStatus.COMPLETED || + workflowRunToUpdate.status === WorkflowRunStatus.FAILED + ) { + throw new WorkflowRunException( + 'Cannot start a workflow run already ended', + WorkflowRunExceptionCode.INVALID_OPERATION, + ); + } + if ( workflowRunToUpdate.status !== WorkflowRunStatus.ENQUEUED && workflowRunToUpdate.status !== WorkflowRunStatus.NOT_STARTED ) { - throw new WorkflowRunException( - 'Workflow run already started', - WorkflowRunExceptionCode.INVALID_OPERATION, - ); + return; } const partialUpdate = { diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workspace-services/workflow-runner.workspace-service.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workspace-services/workflow-runner.workspace-service.ts index c3ffc34c2f..5a372290ca 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-runner/workspace-services/workflow-runner.workspace-service.ts +++ b/packages/twenty-server/src/modules/workflow/workflow-runner/workspace-services/workflow-runner.workspace-service.ts @@ -9,7 +9,7 @@ import { WorkflowRunStatus } from 'src/modules/workflow/common/standard-objects/ import { WorkflowCommonWorkspaceService } from 'src/modules/workflow/common/workspace-services/workflow-common.workspace-service'; import { RunWorkflowJob, - type RunWorkflowJobData, + 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'; import { WorkflowRunWorkspaceService } from 'src/modules/workflow/workflow-runner/workflow-run/workflow-run.workspace-service'; @@ -56,7 +56,7 @@ export class WorkflowRunnerWorkspaceService { }); const remainingRunsToEnqueueCount = - await this.workflowRunQueueWorkspaceService.getRemainingRunsToEnqueueCount( + await this.workflowRunQueueWorkspaceService.getRemainingRunsToEnqueueCountFromCache( workspaceId, ); diff --git a/packages/twenty-server/src/modules/workflow/workflow-trigger/automated-trigger/automated-trigger.module.ts b/packages/twenty-server/src/modules/workflow/workflow-trigger/automated-trigger/automated-trigger.module.ts index dbff9374c0..60b53f0498 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-trigger/automated-trigger/automated-trigger.module.ts +++ b/packages/twenty-server/src/modules/workflow/workflow-trigger/automated-trigger/automated-trigger.module.ts @@ -2,12 +2,12 @@ import { Module } from '@nestjs/common'; import { TypeOrmModule } from '@nestjs/typeorm'; import { Workspace } from 'src/engine/core-modules/workspace/workspace.entity'; +import { WorkspaceDataSourceModule } from 'src/engine/workspace-datasource/workspace-datasource.module'; import { WorkflowCommonModule } from 'src/modules/workflow/common/workflow-common.module'; import { AutomatedTriggerWorkspaceService } from 'src/modules/workflow/workflow-trigger/automated-trigger/automated-trigger.workspace-service'; -import { CronTriggerCronCommand } from 'src/modules/workflow/workflow-trigger/automated-trigger/crons/commands/cron-trigger.cron.command'; +import { WorkflowCronTriggerCronCommand } from 'src/modules/workflow/workflow-trigger/automated-trigger/crons/commands/workflow-cron-trigger.cron.command'; import { CronTriggerCronJob } from 'src/modules/workflow/workflow-trigger/automated-trigger/crons/jobs/cron-trigger.cron.job'; import { DatabaseEventTriggerListener } from 'src/modules/workflow/workflow-trigger/automated-trigger/listeners/database-event-trigger.listener'; -import { WorkspaceDataSourceModule } from 'src/engine/workspace-datasource/workspace-datasource.module'; @Module({ imports: [ @@ -19,8 +19,8 @@ import { WorkspaceDataSourceModule } from 'src/engine/workspace-datasource/works AutomatedTriggerWorkspaceService, DatabaseEventTriggerListener, CronTriggerCronJob, - CronTriggerCronCommand, + WorkflowCronTriggerCronCommand, ], - exports: [AutomatedTriggerWorkspaceService, CronTriggerCronCommand], + exports: [AutomatedTriggerWorkspaceService, WorkflowCronTriggerCronCommand], }) export class AutomatedTriggerModule {} diff --git a/packages/twenty-server/src/modules/workflow/workflow-trigger/automated-trigger/crons/commands/cron-trigger.cron.command.ts b/packages/twenty-server/src/modules/workflow/workflow-trigger/automated-trigger/crons/commands/workflow-cron-trigger.cron.command.ts similarity index 94% rename from packages/twenty-server/src/modules/workflow/workflow-trigger/automated-trigger/crons/commands/cron-trigger.cron.command.ts rename to packages/twenty-server/src/modules/workflow/workflow-trigger/automated-trigger/crons/commands/workflow-cron-trigger.cron.command.ts index aa00ffd538..de07c9361b 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-trigger/automated-trigger/crons/commands/cron-trigger.cron.command.ts +++ b/packages/twenty-server/src/modules/workflow/workflow-trigger/automated-trigger/crons/commands/workflow-cron-trigger.cron.command.ts @@ -12,7 +12,7 @@ import { name: 'cron:workflow:automated-cron-trigger', description: 'Starts a cron job to trigger cron triggered workflows', }) -export class CronTriggerCronCommand extends CommandRunner { +export class WorkflowCronTriggerCronCommand extends CommandRunner { constructor( @InjectMessageQueue(MessageQueue.cronQueue) private readonly messageQueueService: MessageQueueService,