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 09ce49886dc..ba3d6086833 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 @@ -10,6 +10,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 { 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'; @@ -33,6 +34,7 @@ export class CronRegisterAllCommand extends CommandRunner { private readonly checkCustomDomainValidRecordsCronCommand: CheckCustomDomainValidRecordsCronCommand, private readonly workflowRunEnqueueCronCommand: WorkflowRunEnqueueCronCommand, private readonly workflowHandleStaledRunsCronCommand: WorkflowHandleStaledRunsCronCommand, + private readonly workflowCleanWorkflowRunsCronCommand: WorkflowCleanWorkflowRunsCommand, ) { super(); } @@ -85,6 +87,10 @@ export class CronRegisterAllCommand extends CommandRunner { name: 'WorkflowHandleStaledRuns', command: this.workflowHandleStaledRunsCronCommand, }, + { + name: 'WorkflowCleanWorkflowRuns', + command: this.workflowCleanWorkflowRunsCronCommand, + }, ]; let successCount = 0; diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/command/cron-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 similarity index 82% rename from packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/command/cron-clean-workflow-runs.cron.command.ts rename to packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/command/workflow-clean-workflow-runs.cron.command.ts index 6aee33d4560..b203a923fd2 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/command/cron-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 @@ -5,14 +5,14 @@ import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queu import { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service'; import { CLEAN_WORKFLOW_RUN_CRON_PATTERN, - CleanWorkflowRunsJob, -} from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/clean-workflow-runs.cron.job'; + WorkflowCleanWorkflowRunsJob, +} from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-clean-workflow-runs.cron.job'; @Command({ name: 'cron:workflow:clean-workflow-runs', description: 'Clean workflow runs', }) -export class CronCleanWorkflowRunsCommand extends CommandRunner { +export class WorkflowCleanWorkflowRunsCommand extends CommandRunner { constructor( @InjectMessageQueue(MessageQueue.cronQueue) private readonly messageQueueService: MessageQueueService, @@ -22,7 +22,7 @@ export class CronCleanWorkflowRunsCommand extends CommandRunner { async run(): Promise { await this.messageQueueService.addCron({ - jobName: CleanWorkflowRunsJob.name, + jobName: WorkflowCleanWorkflowRunsJob.name, data: undefined, options: { repeat: { diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/clean-workflow-runs.cron.job.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-clean-workflow-runs.cron.job.ts similarity index 91% rename from packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/clean-workflow-runs.cron.job.ts rename to packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-clean-workflow-runs.cron.job.ts index cc3160ddfc5..dea32cc745c 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/clean-workflow-runs.cron.job.ts +++ b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-clean-workflow-runs.cron.job.ts @@ -22,8 +22,8 @@ export const CLEAN_WORKFLOW_RUN_CRON_PATTERN = '0 0 * * *'; const NUMBER_OF_WORKFLOW_RUNS_TO_KEEP = 1000; @Processor(MessageQueue.cronQueue) -export class CleanWorkflowRunsJob { - private readonly logger = new Logger(CleanWorkflowRunsJob.name); +export class WorkflowCleanWorkflowRunsJob { + private readonly logger = new Logger(WorkflowCleanWorkflowRunsJob.name); constructor( @InjectRepository(Workspace, 'core') @@ -32,8 +32,11 @@ export class CleanWorkflowRunsJob { private readonly twentyORMGlobalManager: TwentyORMGlobalManager, ) {} - @Process(CleanWorkflowRunsJob.name) - @SentryCronMonitor(CleanWorkflowRunsJob.name, CLEAN_WORKFLOW_RUN_CRON_PATTERN) + @Process(WorkflowCleanWorkflowRunsJob.name) + @SentryCronMonitor( + WorkflowCleanWorkflowRunsJob.name, + CLEAN_WORKFLOW_RUN_CRON_PATTERN, + ) async handle() { const activeWorkspaces = await this.workspaceRepository.find({ where: { 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 index 983cfe313b6..32b7507a94c 100644 --- 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 @@ -1,4 +1,4 @@ -import { LessThan } from 'typeorm'; +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'; @@ -37,7 +37,7 @@ export class WorkflowHandleStaledRunsPerWorkspaceJob { const staledWorkflowRuns = await workflowRunRepository.find({ where: { status: WorkflowRunStatus.ENQUEUED, - enqueuedAt: LessThan(oneHourAgo), + enqueuedAt: Or(LessThan(oneHourAgo), IsNull()), }, }); 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 78e76494e6c..7ca8e714a4e 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,10 +6,10 @@ 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 { CronCleanWorkflowRunsCommand } from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/command/cron-clean-workflow-runs.cron.command'; +import { WorkflowCleanWorkflowRunsCommand } 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 { CleanWorkflowRunsJob } from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/clean-workflow-runs.cron.job'; +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'; @@ -32,13 +32,14 @@ import { WorkflowRunQueueWorkspaceService } from 'src/modules/workflow/workflow- WorkflowHandleStaledRunsCronCommand, WorkflowHandleStaledRunsJob, WorkflowHandleStaledRunsPerWorkspaceJob, - CleanWorkflowRunsJob, - CronCleanWorkflowRunsCommand, + WorkflowCleanWorkflowRunsJob, + WorkflowCleanWorkflowRunsCommand, ], exports: [ WorkflowRunQueueWorkspaceService, WorkflowRunEnqueueCronCommand, WorkflowHandleStaledRunsCronCommand, + WorkflowCleanWorkflowRunsCommand, ], }) export class WorkflowRunQueueModule {}