Throw 400 for webhook triggering deleted workspaces + enqueue workflows by batches (#17154)

Fixes https://github.com/twentyhq/twenty/issues/17027

Fixes https://github.com/twentyhq/twenty/issues/16824

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<WorkspaceEntity>,
   ) {}
 
   @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 } =

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,

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<RunWorkflowJobData>(
+                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<RunWorkflowJobData>(
-              RunWorkflowJob.name,
-              {
-                workflowRunId,
-                workspaceId,
-              },
-            );
-          }
-
           if (isCacheMode) {
             await this.workflowThrottlingWorkspaceService.decreaseWorkflowRunNotStartedCount(
               workspaceId,
-              workflowRunIdsToEnqueue.length,
+              totalEnqueuedCount,
             );
           } else {
             await this.workflowThrottlingWorkspaceService.recomputeWorkflowRunNotStartedCount(