From c521406bf89619e02a0e0274e7f6584457859881 Mon Sep 17 00:00:00 2001 From: Thomas Trompette Date: Thu, 5 Feb 2026 14:36:32 +0100 Subject: [PATCH] Improve workflow crons (#17720) Issue 1: no info to debug cron trigger. Stop catching exception + using logs instead of throwing for now Issue 2: sentry often send timeouts errors for workflow crons. Probably not real ones, it sends it if the job takes more than 5 minutes to run. To fix, on each workflow cron we do: - loop over active workspaces - perform a query check that workspace is relevant, using count for performances - send a job if relevant --- .../workflow-handle-staled-runs.command.ts | 29 +++- .../not-started-runs-find-options.ts | 11 ++ .../number-of-workflow-runs-to-keep.ts | 1 + .../constants/runs-to-clean-threshold.ts | 1 + .../constants/staled-runs-threshold.ts | 1 + ...rkflow-clean-workflow-runs.cron.command.ts | 4 +- ...orkflow-handle-staled-runs.cron.command.ts | 4 +- .../workflow-clean-workflow-runs.cron.job.ts | 132 +++++++++--------- .../workflow-handle-staled-runs.cron.job.ts | 69 +++++++-- .../jobs/workflow-run-enqueue.cron.job.ts | 86 ++++++++---- .../jobs/workflow-clean-workflow-runs.job.ts | 75 ++++++++++ .../jobs/workflow-handle-staled-runs.job.ts | 26 ++++ .../get-runs-to-clean-find-options.util.ts | 19 +++ .../get-staled-runs-find-options.util.ts | 17 +++ .../workflow-run-queue.module.ts | 10 +- ...ow-handle-staled-runs.workspace-service.ts | 40 +----- .../workflow-run-enqueue.workspace-service.ts | 5 +- .../workflow-throttling.workspace-service.ts | 16 +-- .../jobs/workflow-cron-trigger-cron.job.ts | 35 ++++- .../jobs/workflow-trigger.job.ts | 129 +++++++---------- 20 files changed, 465 insertions(+), 245 deletions(-) create mode 100644 packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/constants/not-started-runs-find-options.ts create mode 100644 packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/constants/number-of-workflow-runs-to-keep.ts create mode 100644 packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/constants/runs-to-clean-threshold.ts create mode 100644 packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/constants/staled-runs-threshold.ts create mode 100644 packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/jobs/workflow-clean-workflow-runs.job.ts create mode 100644 packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/jobs/workflow-handle-staled-runs.job.ts create mode 100644 packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/utils/get-runs-to-clean-find-options.util.ts create mode 100644 packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/utils/get-staled-runs-find-options.util.ts 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 index 1c6db06667..f55b7215c7 100644 --- 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 @@ -1,3 +1,5 @@ +import { Logger } from '@nestjs/common'; + 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'; @@ -11,6 +13,8 @@ type WorkflowHandleStaledRunsCommandOptions = { description: 'Handles staled workflow runs', }) export class WorkflowHandleStaledRunsCommand extends CommandRunner { + private readonly logger = new Logger(WorkflowHandleStaledRunsCommand.name); + constructor( private readonly workflowHandleStaledRunsWorkspaceService: WorkflowHandleStaledRunsWorkspaceService, ) { @@ -32,8 +36,27 @@ export class WorkflowHandleStaledRunsCommand extends CommandRunner { ): Promise { const { workspaceIds } = options; - await this.workflowHandleStaledRunsWorkspaceService.handleStaledRuns({ - workspaceIds, - }); + this.logger.log('Starting WorkflowHandleStaledRunsCommand command'); + + for (let i = 0; i < workspaceIds.length; i++) { + const workspaceId = workspaceIds[i]; + + this.logger.log( + `Processing workspace ${workspaceId} (${i + 1}/${workspaceIds.length})`, + ); + + try { + await this.workflowHandleStaledRunsWorkspaceService.handleStaledRunsForWorkspace( + workspaceId, + ); + } catch (error) { + this.logger.error( + `Failed to handle staled runs for workspace ${workspaceId}`, + error instanceof Error ? error.stack : String(error), + ); + } + } + + this.logger.log('Completed WorkflowHandleStaledRunsCommand command'); } } diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/constants/not-started-runs-find-options.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/constants/not-started-runs-find-options.ts new file mode 100644 index 0000000000..73517a5811 --- /dev/null +++ b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/constants/not-started-runs-find-options.ts @@ -0,0 +1,11 @@ +import { type FindOptionsWhere } from 'typeorm'; + +import { + WorkflowRunStatus, + type WorkflowRunWorkspaceEntity, +} from 'src/modules/workflow/common/standard-objects/workflow-run.workspace-entity'; + +export const NOT_STARTED_RUNS_FIND_OPTIONS: FindOptionsWhere = + { + status: WorkflowRunStatus.NOT_STARTED, + }; diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/constants/number-of-workflow-runs-to-keep.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/constants/number-of-workflow-runs-to-keep.ts new file mode 100644 index 0000000000..0bb4904f1c --- /dev/null +++ b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/constants/number-of-workflow-runs-to-keep.ts @@ -0,0 +1 @@ +export const NUMBER_OF_WORKFLOW_RUNS_TO_KEEP = 1000; diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/constants/runs-to-clean-threshold.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/constants/runs-to-clean-threshold.ts new file mode 100644 index 0000000000..c1d63fe742 --- /dev/null +++ b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/constants/runs-to-clean-threshold.ts @@ -0,0 +1 @@ +export const RUNS_TO_CLEAN_THRESHOLD_DAYS = 14; diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/constants/staled-runs-threshold.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/constants/staled-runs-threshold.ts new file mode 100644 index 0000000000..a0f7aa3f49 --- /dev/null +++ b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/constants/staled-runs-threshold.ts @@ -0,0 +1 @@ +export const STALED_RUNS_THRESHOLD_MS = 60 * 60 * 1000; // 1 hour 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 63f955611f..728de686a2 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 @@ -5,7 +5,7 @@ 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, - WorkflowCleanWorkflowRunsJob, + WorkflowCleanWorkflowRunsCronJob, } from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-clean-workflow-runs.cron.job'; @Command({ @@ -22,7 +22,7 @@ export class WorkflowCleanWorkflowRunsCronCommand extends CommandRunner { async run(): Promise { await this.messageQueueService.addCron({ - jobName: WorkflowCleanWorkflowRunsJob.name, + jobName: WorkflowCleanWorkflowRunsCronJob.name, data: undefined, options: { repeat: { 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 cb55bca6b6..c3a45f4a90 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 @@ -5,7 +5,7 @@ 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 { WORKFLOW_HANDLE_STALED_RUNS_CRON_PATTERN, - WorkflowHandleStaledRunsJob, + WorkflowHandleStaledRunsCronJob, } from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-handle-staled-runs.cron.job'; @Command({ @@ -22,7 +22,7 @@ export class WorkflowHandleStaledRunsCronCommand extends CommandRunner { async run(): Promise { await this.messageQueueService.addCron({ - jobName: WorkflowHandleStaledRunsJob.name, + jobName: WorkflowHandleStaledRunsCronJob.name, data: undefined, options: { repeat: { diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-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 index 4f75ef8f03..b8977c4914 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-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 @@ -1,114 +1,108 @@ import { Logger } from '@nestjs/common'; -import { InjectDataSource, InjectRepository } from '@nestjs/typeorm'; +import { InjectRepository } from '@nestjs/typeorm'; import { WorkspaceActivationStatus } from 'twenty-shared/workspace'; -import { DataSource, Repository } from 'typeorm'; +import { In, 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 { WorkspaceEntity } from 'src/engine/core-modules/workspace/workspace.entity'; import { GlobalWorkspaceOrmManager } from 'src/engine/twenty-orm/global-workspace-datasource/global-workspace-orm.manager'; import { buildSystemAuthContext } from 'src/engine/twenty-orm/utils/build-system-auth-context.util'; -import { getWorkspaceSchemaName } from 'src/engine/workspace-datasource/utils/get-workspace-schema-name.util'; import { WorkflowRunStatus, WorkflowRunWorkspaceEntity, } from 'src/modules/workflow/common/standard-objects/workflow-run.workspace-entity'; +import { NUMBER_OF_WORKFLOW_RUNS_TO_KEEP } from 'src/modules/workflow/workflow-runner/workflow-run-queue/constants/number-of-workflow-runs-to-keep'; +import { + WorkflowCleanWorkflowRunsJob, + WorkflowCleanWorkflowRunsJobData, +} from 'src/modules/workflow/workflow-runner/workflow-run-queue/jobs/workflow-clean-workflow-runs.job'; +import { getRunsToCleanFindOptions } from 'src/modules/workflow/workflow-runner/workflow-run-queue/utils/get-runs-to-clean-find-options.util'; export const CLEAN_WORKFLOW_RUN_CRON_PATTERN = '0 0 * * *'; -const NUMBER_OF_WORKFLOW_RUNS_TO_KEEP = 1000; - @Processor(MessageQueue.cronQueue) -export class WorkflowCleanWorkflowRunsJob { - private readonly logger = new Logger(WorkflowCleanWorkflowRunsJob.name); +export class WorkflowCleanWorkflowRunsCronJob { + private readonly logger = new Logger(WorkflowCleanWorkflowRunsCronJob.name); constructor( @InjectRepository(WorkspaceEntity) private readonly workspaceRepository: Repository, + @InjectMessageQueue(MessageQueue.workflowQueue) + private readonly messageQueueService: MessageQueueService, private readonly globalWorkspaceOrmManager: GlobalWorkspaceOrmManager, - @InjectDataSource() - private readonly coreDataSource: DataSource, ) {} - @Process(WorkflowCleanWorkflowRunsJob.name) + @Process(WorkflowCleanWorkflowRunsCronJob.name) @SentryCronMonitor( - WorkflowCleanWorkflowRunsJob.name, + WorkflowCleanWorkflowRunsCronJob.name, CLEAN_WORKFLOW_RUN_CRON_PATTERN, ) async handle() { - this.logger.log('Starting WorkflowCleanWorkflowRunsJob cron'); + this.logger.log('Starting WorkflowCleanWorkflowRunsCronJob cron'); - try { - const activeWorkspaces = await this.workspaceRepository.find({ - where: { - activationStatus: WorkspaceActivationStatus.ACTIVE, - }, - }); + const activeWorkspaces = await this.workspaceRepository.find({ + where: { + activationStatus: WorkspaceActivationStatus.ACTIVE, + }, + select: ['id'], + }); - for (let i = 0; i < activeWorkspaces.length; i++) { - const activeWorkspace = activeWorkspaces[i]; + let enqueuedCount = 0; - this.logger.log( - `Processing workspace ${activeWorkspace.id} (${i + 1}/${activeWorkspaces.length})`, + for (const workspace of activeWorkspaces) { + const hasRunsToClean = await this.hasRunsToClean(workspace.id); + + if (hasRunsToClean) { + await this.messageQueueService.add( + WorkflowCleanWorkflowRunsJob.name, + { + workspaceId: workspace.id, + }, ); - - try { - await this.cleanWorkflowRunsForWorkspace(activeWorkspace.id); - } catch (error) { - this.logger.error( - `Failed to clean workflow runs for workspace ${activeWorkspace.id}`, - error, - ); - } + enqueuedCount++; } - - this.logger.log('Completed WorkflowCleanWorkflowRunsJob cron'); - } catch (error) { - this.logger.error('WorkflowCleanWorkflowRunsJob cron failed', error); - throw error; } + + this.logger.log( + `Completed WorkflowCleanWorkflowRunsCronJob cron, enqueued ${enqueuedCount} jobs`, + ); } - private async cleanWorkflowRunsForWorkspace(workspaceId: string) { - const schemaName = getWorkspaceSchemaName(workspaceId); + private async hasRunsToClean(workspaceId: string): Promise { const authContext = buildSystemAuthContext(workspaceId); - await this.globalWorkspaceOrmManager.executeInWorkspaceContext(async () => { - const workflowRunsToDelete = await this.coreDataSource.query( - ` - WITH ranked_runs AS ( - SELECT id, - ROW_NUMBER() OVER ( - PARTITION BY "workflowId" - ORDER BY "createdAt" DESC - ) AS rn, - "createdAt" - FROM ${schemaName}."workflowRun" - WHERE status IN ('${WorkflowRunStatus.COMPLETED}', '${WorkflowRunStatus.FAILED}') - ) - SELECT id, rn FROM ranked_runs - WHERE rn > ${NUMBER_OF_WORKFLOW_RUNS_TO_KEEP} - OR "createdAt" < NOW() - INTERVAL '14 days'; - `, - ); + return this.globalWorkspaceOrmManager.executeInWorkspaceContext( + async () => { + const workflowRunRepository = + await this.globalWorkspaceOrmManager.getRepository( + workspaceId, + WorkflowRunWorkspaceEntity, + { shouldBypassPermissionChecks: true }, + ); - const workflowRunRepository = - await this.globalWorkspaceOrmManager.getRepository( - workspaceId, - WorkflowRunWorkspaceEntity, - { shouldBypassPermissionChecks: true }, - ); + const hasOldRuns = await workflowRunRepository.exists({ + where: getRunsToCleanFindOptions(), + }); - for (const workflowRunToDelete of workflowRunsToDelete) { - await workflowRunRepository.delete(workflowRunToDelete.id); - } + if (hasOldRuns) { + return true; + } - this.logger.log( - `Deleted ${workflowRunsToDelete.length} workflow runs for workspace ${workspaceId}`, - ); - }, authContext); + const totalCompletedRunsCount = await workflowRunRepository.count({ + where: { + status: In([WorkflowRunStatus.COMPLETED, WorkflowRunStatus.FAILED]), + }, + }); + + return totalCompletedRunsCount > NUMBER_OF_WORKFLOW_RUNS_TO_KEEP; + }, + authContext, + ); } } diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-handle-staled-runs.cron.job.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-handle-staled-runs.cron.job.ts index 3599bb0737..ef604ea8bd 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-handle-staled-runs.cron.job.ts +++ b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-handle-staled-runs.cron.job.ts @@ -1,39 +1,92 @@ +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 { WorkspaceEntity } from 'src/engine/core-modules/workspace/workspace.entity'; -import { WorkflowHandleStaledRunsWorkspaceService } from 'src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-handle-staled-runs.workspace-service'; +import { GlobalWorkspaceOrmManager } from 'src/engine/twenty-orm/global-workspace-datasource/global-workspace-orm.manager'; +import { buildSystemAuthContext } from 'src/engine/twenty-orm/utils/build-system-auth-context.util'; +import { WorkflowRunWorkspaceEntity } from 'src/modules/workflow/common/standard-objects/workflow-run.workspace-entity'; +import { + WorkflowHandleStaledRunsJob, + WorkflowHandleStaledRunsJobData, +} from 'src/modules/workflow/workflow-runner/workflow-run-queue/jobs/workflow-handle-staled-runs.job'; +import { getStaledRunsFindOptions } from 'src/modules/workflow/workflow-runner/workflow-run-queue/utils/get-staled-runs-find-options.util'; export const WORKFLOW_HANDLE_STALED_RUNS_CRON_PATTERN = '0 * * * *'; @Processor(MessageQueue.cronQueue) -export class WorkflowHandleStaledRunsJob { +export class WorkflowHandleStaledRunsCronJob { + private readonly logger = new Logger(WorkflowHandleStaledRunsCronJob.name); + constructor( @InjectRepository(WorkspaceEntity) private readonly workspaceRepository: Repository, - private readonly workflowHandleStaledRunsWorkspaceService: WorkflowHandleStaledRunsWorkspaceService, + @InjectMessageQueue(MessageQueue.workflowQueue) + private readonly messageQueueService: MessageQueueService, + private readonly globalWorkspaceOrmManager: GlobalWorkspaceOrmManager, ) {} - @Process(WorkflowHandleStaledRunsJob.name) + @Process(WorkflowHandleStaledRunsCronJob.name) @SentryCronMonitor( - WorkflowHandleStaledRunsJob.name, + WorkflowHandleStaledRunsCronJob.name, WORKFLOW_HANDLE_STALED_RUNS_CRON_PATTERN, ) async handle() { + this.logger.log('Starting WorkflowHandleStaledRunsCronJob cron'); + const activeWorkspaces = await this.workspaceRepository.find({ where: { activationStatus: WorkspaceActivationStatus.ACTIVE, }, + select: ['id'], }); - await this.workflowHandleStaledRunsWorkspaceService.handleStaledRuns({ - workspaceIds: activeWorkspaces.map((workspace) => workspace.id), - }); + let enqueuedCount = 0; + + for (const workspace of activeWorkspaces) { + const hasStaledRuns = await this.hasStaledRuns(workspace.id); + + if (hasStaledRuns) { + await this.messageQueueService.add( + WorkflowHandleStaledRunsJob.name, + { + workspaceId: workspace.id, + }, + ); + enqueuedCount++; + } + } + + this.logger.log( + `Completed WorkflowHandleStaledRunsCronJob cron, enqueued ${enqueuedCount} jobs`, + ); + } + + private async hasStaledRuns(workspaceId: string): Promise { + const authContext = buildSystemAuthContext(workspaceId); + + return this.globalWorkspaceOrmManager.executeInWorkspaceContext( + async () => { + const workflowRunRepository = + await this.globalWorkspaceOrmManager.getRepository( + workspaceId, + WorkflowRunWorkspaceEntity, + { shouldBypassPermissionChecks: true }, + ); + + return workflowRunRepository.exists({ + where: getStaledRunsFindOptions(), + }); + }, + authContext, + ); } } 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 index cd65d29433..e0521b3e33 100644 --- 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 @@ -5,11 +5,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 { WorkspaceEntity } from 'src/engine/core-modules/workspace/workspace.entity'; -import { WorkflowRunEnqueueWorkspaceService } from 'src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-run-enqueue.workspace-service'; +import { GlobalWorkspaceOrmManager } from 'src/engine/twenty-orm/global-workspace-datasource/global-workspace-orm.manager'; +import { buildSystemAuthContext } from 'src/engine/twenty-orm/utils/build-system-auth-context.util'; +import { WorkflowRunWorkspaceEntity } from 'src/modules/workflow/common/standard-objects/workflow-run.workspace-entity'; +import { NOT_STARTED_RUNS_FIND_OPTIONS } from 'src/modules/workflow/workflow-runner/workflow-run-queue/constants/not-started-runs-find-options'; +import { + WorkflowRunEnqueueJob, + WorkflowRunEnqueueJobData, +} from 'src/modules/workflow/workflow-runner/workflow-run-queue/jobs/workflow-run-enqueue.job'; export const WORKFLOW_RUN_ENQUEUE_CRON_PATTERN = '*/5 * * * *'; @@ -20,7 +29,9 @@ export class WorkflowRunEnqueueCronJob { constructor( @InjectRepository(WorkspaceEntity) private readonly workspaceRepository: Repository, - private readonly workflowRunEnqueueWorkspaceService: WorkflowRunEnqueueWorkspaceService, + @InjectMessageQueue(MessageQueue.workflowQueue) + private readonly messageQueueService: MessageQueueService, + private readonly globalWorkspaceOrmManager: GlobalWorkspaceOrmManager, ) {} @Process(WorkflowRunEnqueueCronJob.name) @@ -31,39 +42,52 @@ export class WorkflowRunEnqueueCronJob { async handle() { this.logger.log('Starting WorkflowRunEnqueueCronJob cron'); - try { - const activeWorkspaces = await this.workspaceRepository.find({ - where: { - activationStatus: WorkspaceActivationStatus.ACTIVE, - }, - }); + const activeWorkspaces = await this.workspaceRepository.find({ + where: { + activationStatus: WorkspaceActivationStatus.ACTIVE, + }, + select: ['id'], + }); - for (let i = 0; i < activeWorkspaces.length; i++) { - const workspace = activeWorkspaces[i]; + let enqueuedCount = 0; - this.logger.log( - `Processing workspace ${workspace.id} (${i + 1}/${activeWorkspaces.length})`, + for (const workspace of activeWorkspaces) { + const hasNotStartedRuns = await this.hasNotStartedRuns(workspace.id); + + if (hasNotStartedRuns) { + await this.messageQueueService.add( + WorkflowRunEnqueueJob.name, + { + workspaceId: workspace.id, + isCacheMode: false, + }, ); - - try { - await this.workflowRunEnqueueWorkspaceService.enqueueRunsForWorkspace( - { - workspaceId: workspace.id, - isCacheMode: false, - }, - ); - } catch (error) { - this.logger.error( - `Failed to enqueue runs for workspace ${workspace.id}`, - error, - ); - } + enqueuedCount++; } - - this.logger.log('Completed WorkflowRunEnqueueCronJob cron'); - } catch (error) { - this.logger.error('WorkflowRunEnqueueCronJob cron failed', error); - throw error; } + + this.logger.log( + `Completed WorkflowRunEnqueueCronJob cron, enqueued ${enqueuedCount} jobs`, + ); + } + + private async hasNotStartedRuns(workspaceId: string): Promise { + const authContext = buildSystemAuthContext(workspaceId); + + return this.globalWorkspaceOrmManager.executeInWorkspaceContext( + async () => { + const workflowRunRepository = + await this.globalWorkspaceOrmManager.getRepository( + workspaceId, + WorkflowRunWorkspaceEntity, + { shouldBypassPermissionChecks: true }, + ); + + return workflowRunRepository.exists({ + where: NOT_STARTED_RUNS_FIND_OPTIONS, + }); + }, + authContext, + ); } } diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/jobs/workflow-clean-workflow-runs.job.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/jobs/workflow-clean-workflow-runs.job.ts new file mode 100644 index 0000000000..5c05bc7f9b --- /dev/null +++ b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/jobs/workflow-clean-workflow-runs.job.ts @@ -0,0 +1,75 @@ +import { Logger, Scope } from '@nestjs/common'; +import { InjectDataSource } from '@nestjs/typeorm'; + +import { DataSource } 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 { GlobalWorkspaceOrmManager } from 'src/engine/twenty-orm/global-workspace-datasource/global-workspace-orm.manager'; +import { buildSystemAuthContext } from 'src/engine/twenty-orm/utils/build-system-auth-context.util'; +import { getWorkspaceSchemaName } from 'src/engine/workspace-datasource/utils/get-workspace-schema-name.util'; +import { + WorkflowRunStatus, + WorkflowRunWorkspaceEntity, +} from 'src/modules/workflow/common/standard-objects/workflow-run.workspace-entity'; +import { NUMBER_OF_WORKFLOW_RUNS_TO_KEEP } from 'src/modules/workflow/workflow-runner/workflow-run-queue/constants/number-of-workflow-runs-to-keep'; +import { RUNS_TO_CLEAN_THRESHOLD_DAYS } from 'src/modules/workflow/workflow-runner/workflow-run-queue/constants/runs-to-clean-threshold'; + +export type WorkflowCleanWorkflowRunsJobData = { + workspaceId: string; +}; + +@Processor({ queueName: MessageQueue.workflowQueue, scope: Scope.REQUEST }) +export class WorkflowCleanWorkflowRunsJob { + private readonly logger = new Logger(WorkflowCleanWorkflowRunsJob.name); + + constructor( + private readonly globalWorkspaceOrmManager: GlobalWorkspaceOrmManager, + @InjectDataSource() + private readonly dataSource: DataSource, + ) {} + + @Process(WorkflowCleanWorkflowRunsJob.name) + async handle({ + workspaceId, + }: WorkflowCleanWorkflowRunsJobData): Promise { + const schemaName = getWorkspaceSchemaName(workspaceId); + const authContext = buildSystemAuthContext(workspaceId); + + await this.globalWorkspaceOrmManager.executeInWorkspaceContext(async () => { + const workflowRunsToDelete = await this.dataSource.query( + ` + WITH ranked_runs AS ( + SELECT id, + ROW_NUMBER() OVER ( + PARTITION BY "workflowId" + ORDER BY "createdAt" DESC + ) AS rn, + "createdAt" + FROM ${schemaName}."workflowRun" + WHERE status IN ('${WorkflowRunStatus.COMPLETED}', '${WorkflowRunStatus.FAILED}') + ) + SELECT id, rn FROM ranked_runs + WHERE rn > ${NUMBER_OF_WORKFLOW_RUNS_TO_KEEP} + OR "createdAt" < NOW() - INTERVAL '${RUNS_TO_CLEAN_THRESHOLD_DAYS} days'; + `, + ); + + const workflowRunRepository = + await this.globalWorkspaceOrmManager.getRepository( + workspaceId, + WorkflowRunWorkspaceEntity, + { shouldBypassPermissionChecks: true }, + ); + + for (const workflowRunToDelete of workflowRunsToDelete) { + await workflowRunRepository.delete(workflowRunToDelete.id); + } + + this.logger.log( + `Deleted ${workflowRunsToDelete.length} workflow runs for workspace ${workspaceId}`, + ); + }, authContext); + } +} diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/jobs/workflow-handle-staled-runs.job.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/jobs/workflow-handle-staled-runs.job.ts new file mode 100644 index 0000000000..ee8f7c3ab8 --- /dev/null +++ b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/jobs/workflow-handle-staled-runs.job.ts @@ -0,0 +1,26 @@ +import { Scope } from '@nestjs/common'; + +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 { WorkflowHandleStaledRunsWorkspaceService } from 'src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-handle-staled-runs.workspace-service'; + +export type WorkflowHandleStaledRunsJobData = { + workspaceId: string; +}; + +@Processor({ queueName: MessageQueue.workflowQueue, scope: Scope.REQUEST }) +export class WorkflowHandleStaledRunsJob { + constructor( + private readonly workflowHandleStaledRunsWorkspaceService: WorkflowHandleStaledRunsWorkspaceService, + ) {} + + @Process(WorkflowHandleStaledRunsJob.name) + async handle({ + workspaceId, + }: WorkflowHandleStaledRunsJobData): Promise { + await this.workflowHandleStaledRunsWorkspaceService.handleStaledRunsForWorkspace( + workspaceId, + ); + } +} diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/utils/get-runs-to-clean-find-options.util.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/utils/get-runs-to-clean-find-options.util.ts new file mode 100644 index 0000000000..e86bcc3b13 --- /dev/null +++ b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/utils/get-runs-to-clean-find-options.util.ts @@ -0,0 +1,19 @@ +import { type FindOptionsWhere, In, LessThan } from 'typeorm'; + +import { + WorkflowRunStatus, + type WorkflowRunWorkspaceEntity, +} from 'src/modules/workflow/common/standard-objects/workflow-run.workspace-entity'; +import { RUNS_TO_CLEAN_THRESHOLD_DAYS } from 'src/modules/workflow/workflow-runner/workflow-run-queue/constants/runs-to-clean-threshold'; + +export const getRunsToCleanFindOptions = + (): FindOptionsWhere => { + const thresholdDate = new Date( + Date.now() - RUNS_TO_CLEAN_THRESHOLD_DAYS * 24 * 60 * 60 * 1000, + ).toISOString(); + + return { + status: In([WorkflowRunStatus.COMPLETED, WorkflowRunStatus.FAILED]), + createdAt: LessThan(thresholdDate), + }; + }; diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/utils/get-staled-runs-find-options.util.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/utils/get-staled-runs-find-options.util.ts new file mode 100644 index 0000000000..62c12fd8c1 --- /dev/null +++ b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/utils/get-staled-runs-find-options.util.ts @@ -0,0 +1,17 @@ +import { type FindOptionsWhere, IsNull, LessThan, Or } from 'typeorm'; + +import { + WorkflowRunStatus, + type WorkflowRunWorkspaceEntity, +} from 'src/modules/workflow/common/standard-objects/workflow-run.workspace-entity'; +import { STALED_RUNS_THRESHOLD_MS } from 'src/modules/workflow/workflow-runner/workflow-run-queue/constants/staled-runs-threshold'; + +export const getStaledRunsFindOptions = + (): FindOptionsWhere => { + const thresholdDate = new Date(Date.now() - STALED_RUNS_THRESHOLD_MS); + + return { + status: WorkflowRunStatus.ENQUEUED, + enqueuedAt: Or(LessThan(thresholdDate), 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 25c68baf68..ce214144ce 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 @@ -11,9 +11,11 @@ import { WorkflowHandleStaledRunsCommand } from 'src/modules/workflow/workflow-r 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 { WorkflowHandleStaledRunsJob } from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-handle-staled-runs.cron.job'; +import { WorkflowCleanWorkflowRunsCronJob } from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-clean-workflow-runs.cron.job'; +import { WorkflowHandleStaledRunsCronJob } from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-handle-staled-runs.cron.job'; import { WorkflowRunEnqueueCronJob } from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-run-enqueue.cron.job'; +import { WorkflowCleanWorkflowRunsJob } from 'src/modules/workflow/workflow-runner/workflow-run-queue/jobs/workflow-clean-workflow-runs.job'; +import { WorkflowHandleStaledRunsJob } from 'src/modules/workflow/workflow-runner/workflow-run-queue/jobs/workflow-handle-staled-runs.job'; import { WorkflowRunEnqueueJob } from 'src/modules/workflow/workflow-runner/workflow-run-queue/jobs/workflow-run-enqueue.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'; @@ -37,7 +39,9 @@ import { WorkflowThrottlingWorkspaceService } from 'src/modules/workflow/workflo WorkflowHandleStaledRunsWorkspaceService, WorkflowHandleStaledRunsCronCommand, WorkflowHandleStaledRunsCommand, + WorkflowHandleStaledRunsCronJob, WorkflowHandleStaledRunsJob, + WorkflowCleanWorkflowRunsCronJob, WorkflowCleanWorkflowRunsJob, WorkflowCleanWorkflowRunsCronCommand, ], @@ -46,8 +50,10 @@ import { WorkflowThrottlingWorkspaceService } from 'src/modules/workflow/workflo WorkflowRunEnqueueJob, WorkflowRunEnqueueCronJob, WorkflowRunEnqueueCronCommand, + WorkflowHandleStaledRunsCronJob, WorkflowHandleStaledRunsCronCommand, WorkflowHandleStaledRunsCommand, + WorkflowCleanWorkflowRunsCronJob, WorkflowCleanWorkflowRunsCronCommand, ], }) 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 index ed03548ede..35192a88f1 100644 --- 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 @@ -1,13 +1,12 @@ import { Injectable, Logger } from '@nestjs/common'; -import { IsNull, LessThan, Or } from 'typeorm'; - import { GlobalWorkspaceOrmManager } from 'src/engine/twenty-orm/global-workspace-datasource/global-workspace-orm.manager'; import { buildSystemAuthContext } from 'src/engine/twenty-orm/utils/build-system-auth-context.util'; import { WorkflowRunStatus, WorkflowRunWorkspaceEntity, } from 'src/modules/workflow/common/standard-objects/workflow-run.workspace-entity'; +import { getStaledRunsFindOptions } from 'src/modules/workflow/workflow-runner/workflow-run-queue/utils/get-staled-runs-find-options.util'; import { WorkflowThrottlingWorkspaceService } from 'src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-throttling.workspace-service'; @Injectable() @@ -20,35 +19,7 @@ export class WorkflowHandleStaledRunsWorkspaceService { private readonly workflowThrottlingWorkspaceService: WorkflowThrottlingWorkspaceService, ) {} - async handleStaledRuns({ workspaceIds }: { workspaceIds: string[] }) { - this.logger.log('Starting handleStaledRuns'); - - try { - for (let i = 0; i < workspaceIds.length; i++) { - const workspaceId = workspaceIds[i]; - - this.logger.log( - `Processing workspace ${workspaceId} (${i + 1}/${workspaceIds.length})`, - ); - - try { - await this.handleStaledRunsForWorkspace(workspaceId); - } catch (error) { - this.logger.error( - `Failed to handle staled runs for workspace ${workspaceId}`, - error, - ); - } - } - - this.logger.log('Completed handleStaledRuns'); - } catch (error) { - this.logger.error('handleStaledRuns failed', error); - throw error; - } - } - - private async handleStaledRunsForWorkspace(workspaceId: string) { + async handleStaledRunsForWorkspace(workspaceId: string) { const authContext = buildSystemAuthContext(workspaceId); await this.globalWorkspaceOrmManager.executeInWorkspaceContext(async () => { @@ -59,13 +30,8 @@ export class WorkflowHandleStaledRunsWorkspaceService { { 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()), - }, + where: getStaledRunsFindOptions(), }); if (staledWorkflowRuns.length <= 0) { 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 index d19d2e4996..04d906f8e6 100644 --- 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 @@ -15,6 +15,7 @@ import { } from 'src/modules/workflow/common/standard-objects/workflow-run.workspace-entity'; import { RunWorkflowJob } from 'src/modules/workflow/workflow-runner/jobs/run-workflow.job'; import { type RunWorkflowJobData } from 'src/modules/workflow/workflow-runner/types/run-workflow-job-data.type'; +import { NOT_STARTED_RUNS_FIND_OPTIONS } from 'src/modules/workflow/workflow-runner/workflow-run-queue/constants/not-started-runs-find-options'; import { WorkflowThrottlingWorkspaceService } from 'src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-throttling.workspace-service'; @Injectable() @@ -86,9 +87,7 @@ export class WorkflowRunEnqueueWorkspaceService { ); const batchRuns = await workflowRunRepository.find({ - where: { - status: WorkflowRunStatus.NOT_STARTED, - }, + where: NOT_STARTED_RUNS_FIND_OPTIONS, select: { id: true, }, diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-throttling.workspace-service.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-throttling.workspace-service.ts index 43c1cb2078..1fb29539aa 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-throttling.workspace-service.ts +++ b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-throttling.workspace-service.ts @@ -1,7 +1,5 @@ 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'; @@ -9,10 +7,8 @@ import { ThrottlerService } from 'src/engine/core-modules/throttler/throttler.se import { TwentyConfigService } from 'src/engine/core-modules/twenty-config/twenty-config.service'; import { GlobalWorkspaceOrmManager } from 'src/engine/twenty-orm/global-workspace-datasource/global-workspace-orm.manager'; import { buildSystemAuthContext } from 'src/engine/twenty-orm/utils/build-system-auth-context.util'; -import { - WorkflowRunStatus, - WorkflowRunWorkspaceEntity, -} from 'src/modules/workflow/common/standard-objects/workflow-run.workspace-entity'; +import { WorkflowRunWorkspaceEntity } from 'src/modules/workflow/common/standard-objects/workflow-run.workspace-entity'; +import { NOT_STARTED_RUNS_FIND_OPTIONS } from 'src/modules/workflow/workflow-runner/workflow-run-queue/constants/not-started-runs-find-options'; @Injectable() export class WorkflowThrottlingWorkspaceService { @@ -89,9 +85,7 @@ export class WorkflowThrottlingWorkspaceService { ); return workflowRunRepository.count({ - where: { - status: In([WorkflowRunStatus.NOT_STARTED]), - }, + where: NOT_STARTED_RUNS_FIND_OPTIONS, }); }, authContext, @@ -122,9 +116,7 @@ export class WorkflowThrottlingWorkspaceService { ); return workflowRunRepository.count({ - where: { - status: In([WorkflowRunStatus.NOT_STARTED]), - }, + where: NOT_STARTED_RUNS_FIND_OPTIONS, }); }, authContext, diff --git a/packages/twenty-server/src/modules/workflow/workflow-trigger/automated-trigger/crons/jobs/workflow-cron-trigger-cron.job.ts b/packages/twenty-server/src/modules/workflow/workflow-trigger/automated-trigger/crons/jobs/workflow-cron-trigger-cron.job.ts index 1e920afc15..23e073b023 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-trigger/automated-trigger/crons/jobs/workflow-cron-trigger-cron.job.ts +++ b/packages/twenty-server/src/modules/workflow/workflow-trigger/automated-trigger/crons/jobs/workflow-cron-trigger-cron.job.ts @@ -1,3 +1,4 @@ +import { Logger } from '@nestjs/common'; import { InjectDataSource, InjectRepository } from '@nestjs/typeorm'; import { isDefined } from 'twenty-shared/utils'; @@ -25,6 +26,8 @@ export const WORKFLOW_CRON_TRIGGER_CRON_PATTERN = '* * * * *'; @Processor(MessageQueue.cronQueue) export class WorkflowCronTriggerCronJob { + private readonly logger = new Logger(WorkflowCronTriggerCronJob.name); + constructor( @InjectDataSource() private readonly coreDataSource: DataSource, @@ -41,12 +44,16 @@ export class WorkflowCronTriggerCronJob { WORKFLOW_CRON_TRIGGER_CRON_PATTERN, ) async handle() { + this.logger.log('WorkflowCronTriggerCronJob started'); + const activeWorkspaces = await this.workspaceRepository.find({ where: { activationStatus: WorkspaceActivationStatus.ACTIVE, }, }); + this.logger.log(`Found ${activeWorkspaces.length} active workspaces`); + const now = new Date(); for (const activeWorkspace of activeWorkspaces) { @@ -57,18 +64,39 @@ export class WorkflowCronTriggerCronJob { `SELECT * FROM ${schemaName}."workflowAutomatedTrigger" WHERE type = '${AutomatedTriggerType.CRON}'`, ); + this.logger.log( + `Workspace ${activeWorkspace.id}: found ${workflowAutomatedCronTriggers.length} cron triggers`, + ); + for (const workflowAutomatedCronTrigger of workflowAutomatedCronTriggers) { const settings = workflowAutomatedCronTrigger.settings as CronTriggerSettings; + this.logger.log( + `Trigger ${workflowAutomatedCronTrigger.id} for workflow ${workflowAutomatedCronTrigger.workflowId}: pattern=${settings.pattern}`, + ); + if (!isDefined(settings.pattern)) { + this.logger.warn( + `Trigger ${workflowAutomatedCronTrigger.id}: skipping - pattern not defined`, + ); continue; } - if (!shouldRunNow(settings.pattern, now)) { + const shouldRun = shouldRunNow(settings.pattern, now); + + this.logger.log( + `Trigger ${workflowAutomatedCronTrigger.id}: shouldRunNow(${settings.pattern}, ${now.toISOString()}) = ${shouldRun}`, + ); + + if (!shouldRun) { continue; } + this.logger.log( + `Trigger ${workflowAutomatedCronTrigger.id}: enqueuing WorkflowTriggerJob for workflow ${workflowAutomatedCronTrigger.workflowId}`, + ); + await this.messageQueueService.add( WorkflowTriggerJob.name, { @@ -80,6 +108,9 @@ export class WorkflowCronTriggerCronJob { ); } } catch (error) { + this.logger.error( + `Error processing workspace ${activeWorkspace.id}: ${error}`, + ); this.exceptionHandlerService.captureExceptions([error], { workspace: { id: activeWorkspace.id, @@ -87,5 +118,7 @@ export class WorkflowCronTriggerCronJob { }); } } + + this.logger.log('WorkflowCronTriggerCronJob completed'); } } diff --git a/packages/twenty-server/src/modules/workflow/workflow-trigger/jobs/workflow-trigger.job.ts b/packages/twenty-server/src/modules/workflow/workflow-trigger/jobs/workflow-trigger.job.ts index 2ec97cebfd..a30f704402 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-trigger/jobs/workflow-trigger.job.ts +++ b/packages/twenty-server/src/modules/workflow/workflow-trigger/jobs/workflow-trigger.job.ts @@ -1,27 +1,19 @@ -import { Scope } from '@nestjs/common'; +import { Logger, Scope } from '@nestjs/common'; import isEmpty from 'lodash.isempty'; import { FieldActorSource } from 'twenty-shared/types'; import { isDefined } from 'twenty-shared/utils'; -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 { handleWorkflowTriggerException } from 'src/engine/core-modules/workflow/filters/workflow-trigger-graphql-api-exception.filter'; import { GlobalWorkspaceOrmManager } from 'src/engine/twenty-orm/global-workspace-datasource/global-workspace-orm.manager'; import { buildSystemAuthContext } from 'src/engine/twenty-orm/utils/build-system-auth-context.util'; -import { - WorkflowVersionStatus, - type WorkflowVersionWorkspaceEntity, -} from 'src/modules/workflow/common/standard-objects/workflow-version.workspace-entity'; +import { WorkflowVersionStatus } from 'src/modules/workflow/common/standard-objects/workflow-version.workspace-entity'; import { type WorkflowWorkspaceEntity } from 'src/modules/workflow/common/standard-objects/workflow.workspace-entity'; +import { WorkflowCommonWorkspaceService } from 'src/modules/workflow/common/workspace-services/workflow-common.workspace-service'; import { WorkflowRunnerWorkspaceService } from 'src/modules/workflow/workflow-runner/workspace-services/workflow-runner.workspace-service'; -import { - WorkflowTriggerException, - WorkflowTriggerExceptionCode, -} from 'src/modules/workflow/workflow-trigger/exceptions/workflow-trigger.exception'; +import { WorkflowTriggerExceptionCode } from 'src/modules/workflow/workflow-trigger/exceptions/workflow-trigger.exception'; export type WorkflowTriggerJobData = { workspaceId: string; @@ -33,11 +25,11 @@ const DEFAULT_WORKFLOW_NAME = 'Workflow'; @Processor({ queueName: MessageQueue.workflowQueue, scope: Scope.REQUEST }) export class WorkflowTriggerJob { + private readonly logger = new Logger(WorkflowTriggerJob.name); constructor( private readonly globalWorkspaceOrmManager: GlobalWorkspaceOrmManager, + private readonly workflowCommonWorkspaceService: WorkflowCommonWorkspaceService, private readonly workflowRunnerWorkspaceService: WorkflowRunnerWorkspaceService, - @InjectMessageQueue(MessageQueue.workflowQueue) - private readonly messageQueueService: MessageQueueService, ) {} @Process(WorkflowTriggerJob.name) @@ -45,77 +37,64 @@ export class WorkflowTriggerJob { const authContext = buildSystemAuthContext(data.workspaceId); await this.globalWorkspaceOrmManager.executeInWorkspaceContext(async () => { - try { - const workflowRepository = - await this.globalWorkspaceOrmManager.getRepository( - data.workspaceId, - 'workflow', - { shouldBypassPermissionChecks: true }, - ); + const workflowRepository = + await this.globalWorkspaceOrmManager.getRepository( + data.workspaceId, + 'workflow', + { shouldBypassPermissionChecks: true }, + ); - const workflow = await workflowRepository.findOneBy({ - id: data.workflowId, - }); + const workflow = await workflowRepository.findOneBy({ + id: data.workflowId, + }); - if (!workflow) { - throw new WorkflowTriggerException( - `Workflow ${data.workflowId} not found in workspace ${data.workspaceId}`, - WorkflowTriggerExceptionCode.NOT_FOUND, - ); - } + if (!workflow) { + this.logger.error( + `Workflow ${data.workflowId} not found in workspace ${data.workspaceId}`, + WorkflowTriggerExceptionCode.NOT_FOUND, + ); - if (!workflow.lastPublishedVersionId) { - throw new WorkflowTriggerException( - `Workflow ${data.workflowId} has no published version in workspace ${data.workspaceId}`, - WorkflowTriggerExceptionCode.INTERNAL_ERROR, - ); - } + return; + } - const workflowVersionRepository = - await this.globalWorkspaceOrmManager.getRepository( - data.workspaceId, - 'workflowVersion', - { shouldBypassPermissionChecks: true }, - ); + if (!workflow.lastPublishedVersionId) { + this.logger.error( + `Workflow ${data.workflowId} has no published version in workspace ${data.workspaceId}`, + WorkflowTriggerExceptionCode.INTERNAL_ERROR, + ); - const workflowVersion = await workflowVersionRepository.findOneBy({ - id: workflow.lastPublishedVersionId, - }); + return; + } - if (!workflowVersion) { - throw new WorkflowTriggerException( - `Workflow version ${workflow.lastPublishedVersionId} not found in workspace ${data.workspaceId}`, - WorkflowTriggerExceptionCode.NOT_FOUND, - ); - } - if (workflowVersion.status !== WorkflowVersionStatus.ACTIVE) { - throw new WorkflowTriggerException( - `Workflow version ${workflowVersion.id} is not active in workspace ${data.workspaceId}`, - WorkflowTriggerExceptionCode.INTERNAL_ERROR, - ); - } - - await this.workflowRunnerWorkspaceService.run({ + const workflowVersion = + await this.workflowCommonWorkspaceService.getWorkflowVersionOrFail({ workspaceId: data.workspaceId, workflowVersionId: workflow.lastPublishedVersionId, - payload: data.payload, - source: { - source: FieldActorSource.WORKFLOW, - name: - isDefined(workflow.name) && !isEmpty(workflow.name) - ? workflow.name - : DEFAULT_WORKFLOW_NAME, - context: {}, - workspaceMemberId: null, - }, }); - } catch (e) { - await this.messageQueueService.removeCron({ - jobName: WorkflowTriggerJob.name, - jobId: data.workflowId, - }); - handleWorkflowTriggerException(e); + + if (workflowVersion.status !== WorkflowVersionStatus.ACTIVE) { + this.logger.error( + `Workflow version ${workflowVersion?.id} is not active in workspace ${data.workspaceId}`, + WorkflowTriggerExceptionCode.INTERNAL_ERROR, + ); + + return; } + + await this.workflowRunnerWorkspaceService.run({ + workspaceId: data.workspaceId, + workflowVersionId: workflow.lastPublishedVersionId, + payload: data.payload, + source: { + source: FieldActorSource.WORKFLOW, + name: + isDefined(workflow.name) && !isEmpty(workflow.name) + ? workflow.name + : DEFAULT_WORKFLOW_NAME, + context: {}, + workspaceMemberId: null, + }, + }); }, authContext); } }