Repository: twentyhq/twenty
File: packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-run-enqueue.workspace-service.ts (TypeScript)

Code before:

            const batchRuns = await workflowRunRepository.find({
              where: {
                status: WorkflowRunStatus.NOT_STARTED,
                ...(alreadyEnqueuedIds.length > 0
                  ? { id: Not(In(alreadyEnqueuedIds)) }
                  : {}),
              },
              select: {
                id: true,
              },
              order: {
                createdAt: 'ASC',
              },
              take: batchSize,
            });

            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,
                },
              );
            }

            alreadyEnqueuedIds.push(...batchIds);
            totalEnqueuedCount += batchRuns.length;
            remainingWorkflowRunToEnqueueCount -= batchRuns.length;
          }

          if (totalEnqueuedCount === 0) {
            if (!isCacheMode) {
              await this.workflowThrottlingWorkspaceService.recomputeWorkflowRunNotStartedCount(
                workspaceId,
              );
            }

Review comment (bug):
Note: the status update is in batch but the enqueue is not. If at some point it fails after the update, some workflows could stay in "enqueued" status forever, right?

Code after:
              where: {
                status: WorkflowRunStatus.NOT_STARTED,
              },
              select: {
                id: true,
              },
              order: {
                createdAt: 'ASC',
              },
              take: batchSize,
            });

            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 (totalEnqueuedCount === 0) {
            if (!isCacheMode) {
              await this.workflowThrottlingWorkspaceService.recomputeWorkflowRunNotStartedCount(
                workspaceId,
              );
            }

            return;
          }

          await this.workflowThrottlingWorkspaceService.consumeRemainingRunsToEnqueueCount(
            workspaceId,