diff --git a/packages/twenty-server/src/engine/core-modules/workflow/controllers/workflow-trigger.controller.ts b/packages/twenty-server/src/engine/core-modules/workflow/controllers/workflow-trigger.controller.ts index 1e9a862cd1..e9fdd9922c 100644 --- a/packages/twenty-server/src/engine/core-modules/workflow/controllers/workflow-trigger.controller.ts +++ b/packages/twenty-server/src/engine/core-modules/workflow/controllers/workflow-trigger.controller.ts @@ -7,12 +7,15 @@ import { UseFilters, UseGuards, } from '@nestjs/common'; +import { InjectRepository } from '@nestjs/typeorm'; import { Request } from 'express'; import { FieldActorSource } from 'twenty-shared/types'; import { isDefined } from 'twenty-shared/utils'; +import { Repository } from 'typeorm'; import { WorkflowTriggerRestApiExceptionFilter } from 'src/engine/core-modules/workflow/filters/workflow-trigger-rest-api-exception.filter'; +import { WorkspaceEntity } from 'src/engine/core-modules/workspace/workspace.entity'; import { NoPermissionGuard } from 'src/engine/guards/no-permission.guard'; import { PublicEndpointGuard } from 'src/engine/guards/public-endpoint.guard'; import { PermissionsGraphqlApiExceptionFilter } from 'src/engine/metadata-modules/permissions/utils/permissions-graphql-api-exception.filter'; @@ -39,6 +42,8 @@ export class WorkflowTriggerController { constructor( private readonly globalWorkspaceOrmManager: GlobalWorkspaceOrmManager, private readonly workflowTriggerWorkspaceService: WorkflowTriggerWorkspaceService, + @InjectRepository(WorkspaceEntity) + protected readonly workspaceRepository: Repository, ) {} @Post('workflows/:workspaceId/:workflowId') @@ -73,6 +78,17 @@ export class WorkflowTriggerController { payload?: object; workspaceId: string; }) { + const workspaceExists = await this.workspaceRepository.existsBy({ + id: workspaceId, + }); + + if (!workspaceExists) { + throw new WorkflowTriggerException( + `[Webhook trigger] Workspace ${workspaceId} not found`, + WorkflowTriggerExceptionCode.NOT_FOUND, + ); + } + const authContext = buildSystemAuthContext(workspaceId); const { workflow } = diff --git a/packages/twenty-server/src/engine/core-modules/workflow/workflow-api.module.ts b/packages/twenty-server/src/engine/core-modules/workflow/workflow-api.module.ts index c449843562..cfdc0c332f 100644 --- a/packages/twenty-server/src/engine/core-modules/workflow/workflow-api.module.ts +++ b/packages/twenty-server/src/engine/core-modules/workflow/workflow-api.module.ts @@ -1,4 +1,5 @@ import { Module } from '@nestjs/common'; +import { TypeOrmModule } from '@nestjs/typeorm'; import { FeatureFlagModule } from 'src/engine/core-modules/feature-flag/feature-flag.module'; import { ToolModule } from 'src/engine/core-modules/tool/tool.module'; @@ -8,6 +9,7 @@ import { WorkflowTriggerResolver } from 'src/engine/core-modules/workflow/resolv import { WorkflowVersionEdgeResolver } from 'src/engine/core-modules/workflow/resolvers/workflow-version-edge.resolver'; import { WorkflowVersionStepResolver } from 'src/engine/core-modules/workflow/resolvers/workflow-version-step.resolver'; import { WorkflowVersionResolver } from 'src/engine/core-modules/workflow/resolvers/workflow-version.resolver'; +import { WorkspaceEntity } from 'src/engine/core-modules/workspace/workspace.entity'; import { PermissionsModule } from 'src/engine/metadata-modules/permissions/permissions.module'; import { WorkflowCommonModule } from 'src/modules/workflow/common/workflow-common.module'; import { WorkflowBuilderModule } from 'src/modules/workflow/workflow-builder/workflow-builder.module'; @@ -18,6 +20,7 @@ import { WorkflowTriggerModule } from 'src/modules/workflow/workflow-trigger/wor @Module({ imports: [ + TypeOrmModule.forFeature([WorkspaceEntity]), FeatureFlagModule, WorkflowTriggerModule, WorkflowBuilderModule, 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 9af9eb9e37..873fcc9f53 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 @@ -1,6 +1,6 @@ import { Injectable, Logger } from '@nestjs/common'; -import { Not } from 'typeorm'; +import { QUERY_MAX_RECORDS } from 'twenty-shared/constants'; 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'; @@ -90,15 +90,17 @@ export class WorkflowRunEnqueueWorkspaceService { workspaceId, ); - const workflowRunIdsToEnqueue: string[] = []; + let totalEnqueuedCount = 0; - if (remainingWorkflowRunToEnqueueCount > 0) { - const additionalRunsToEnqueue = await workflowRunRepository.find({ + while (remainingWorkflowRunToEnqueueCount > 0) { + const batchSize = Math.min( + remainingWorkflowRunToEnqueueCount, + QUERY_MAX_RECORDS, + ); + + const batchRuns = await workflowRunRepository.find({ where: { status: WorkflowRunStatus.NOT_STARTED, - ...(workflowRunIdsToEnqueue.length > 0 - ? { id: Not(workflowRunIdsToEnqueue[0]) } - : {}), }, select: { id: true, @@ -106,17 +108,37 @@ export class WorkflowRunEnqueueWorkspaceService { order: { createdAt: 'ASC', }, - take: remainingWorkflowRunToEnqueueCount, + take: batchSize, }); - workflowRunIdsToEnqueue.push( - ...additionalRunsToEnqueue.map( - (workflowRun: WorkflowRunWorkspaceEntity) => workflowRun.id, - ), + if (batchRuns.length === 0) { + break; + } + + const batchIds = batchRuns.map( + (workflowRun: WorkflowRunWorkspaceEntity) => workflowRun.id, ); + + await workflowRunRepository.update(batchIds, { + enqueuedAt: new Date().toISOString(), + status: WorkflowRunStatus.ENQUEUED, + }); + + for (const workflowRunId of batchIds) { + await this.messageQueueService.add( + RunWorkflowJob.name, + { + workflowRunId, + workspaceId, + }, + ); + } + + totalEnqueuedCount += batchRuns.length; + remainingWorkflowRunToEnqueueCount -= batchRuns.length; } - if (workflowRunIdsToEnqueue.length <= 0) { + if (totalEnqueuedCount === 0) { if (!isCacheMode) { await this.workflowThrottlingWorkspaceService.recomputeWorkflowRunNotStartedCount( workspaceId, @@ -126,30 +148,15 @@ export class WorkflowRunEnqueueWorkspaceService { return; } - await workflowRunRepository.update(workflowRunIdsToEnqueue, { - enqueuedAt: new Date().toISOString(), - status: WorkflowRunStatus.ENQUEUED, - }); - await this.workflowThrottlingWorkspaceService.consumeRemainingRunsToEnqueueCount( workspaceId, - workflowRunIdsToEnqueue.length, + totalEnqueuedCount, ); - for (const workflowRunId of workflowRunIdsToEnqueue) { - await this.messageQueueService.add( - RunWorkflowJob.name, - { - workflowRunId, - workspaceId, - }, - ); - } - if (isCacheMode) { await this.workflowThrottlingWorkspaceService.decreaseWorkflowRunNotStartedCount( workspaceId, - workflowRunIdsToEnqueue.length, + totalEnqueuedCount, ); } else { await this.workflowThrottlingWorkspaceService.recomputeWorkflowRunNotStartedCount(