Repository: pimcore/pimcore (language: php)

Issue:
[Improvement]: Execution modes for `Generic Execution Engine`
### Improvement description

Currently, each selected element inside a JobStep triggers the execution of the action separately. For example, selecting 5 elements results in the action being executed 5 times.

However, we need a feature that allows all selected elements to be passed at once to the action, meaning 5 selected elements would result in the action being executed just once.

### Mode 1: All At Once

Advantages:

- Facilitates easier migration from the old Job Execution Engine to the new Generic Execution Engine.
- FilterActions, such as the PQL action, will only be called once, avoiding redundant operations.
- Could simplify implementation for actions like generating CSV files, ZIP archives, uploading data, etc., which are currently possible but require extensive use of JobRunContext or spawning sub-jobs.

Disadvantages:

- Simplified error handling, limited to "job completed successfully" or "job failed".
- Limits parallel execution to a single worker since only one message is generated.

### Mode 2: For Each

Advantages:

- Potentially faster processing as each element generates a message that can be processed in parallel.
- Improved error handling, with each element able to independently result in success or failure.

Disadvantages:

- May result in unnecessary executions for FilterActions like the PQL action.



Discussion / hints:
This even could be considered as a bug - at least in combination with copilot. 

![image](https://github.com/pimcore/pimcore/assets/8792145/2cb24ab0-5bc2-48f5-8285-895e47aeb6ed)

When I start an action with that configuration with four assets, the configured action will be sent 4 times... 
![image](https://github.com/pimcore/pimcore/assets/8792145/0681da50-976d-4ca4-8cb5-bebff8808729)


Fix patch:
diff --git a/bundles/GenericExecutionEngineBundle/src/Agent/JobExecutionAgent.php b/bundles/GenericExecutionEngineBundle/src/Agent/JobExecutionAgent.php
index cb75ed57da6..68beba0d87f 100644
--- a/bundles/GenericExecutionEngineBundle/src/Agent/JobExecutionAgent.php
+++ b/bundles/GenericExecutionEngineBundle/src/Agent/JobExecutionAgent.php
@@ -19,14 +19,15 @@
 use Doctrine\DBAL\Exception;
 use Pimcore\Bundle\GenericExecutionEngineBundle\Configuration\ExecutionContextInterface;
 use Pimcore\Bundle\GenericExecutionEngineBundle\Entity\JobRun;
-use Pimcore\Bundle\GenericExecutionEngineBundle\Exception\InvalidErrorHandlingModeException;
 use Pimcore\Bundle\GenericExecutionEngineBundle\Messenger\Messages\GenericExecutionEngineMessageInterface;
 use Pimcore\Bundle\GenericExecutionEngineBundle\Model\Job;
 use Pimcore\Bundle\GenericExecutionEngineBundle\Model\JobRunStates;
 use Pimcore\Bundle\GenericExecutionEngineBundle\Repository\JobRunErrorLogRepositoryInterface;
 use Pimcore\Bundle\GenericExecutionEngineBundle\Repository\JobRunRepositoryInterface;
 use Pimcore\Bundle\GenericExecutionEngineBundle\Utils\Enums\ErrorHandlingMode;
+use Pimcore\Bundle\GenericExecutionEngineBundle\Utils\Enums\SelectionProcessingMode;
 use Pimcore\Helper\StopMessengerWorkersTrait;
+use Pimcore\Model\Element\ElementDescriptor;
 use Pimcore\Translation\Translator;
 use Psr\Log\LoggerInterface;
 use Symfony\Component\Messenger\MessageBusInterface;
@@ -146,13 +147,28 @@ public function isRunning(int $jobRunId): bool
         return $jobRun->getState() === JobRunStates::RUNNING;
     }
 
+    private function getSelectionProcessingModeFromJobRun(JobRun $jobRun): SelectionProcessingMode
+    {
+        $steps = $jobRun->getJob()?->getSteps();
+        if($steps !== null) {
+            $step = $steps[$jobRun->getCurrentStep()] ?? null;
+            if($step) {
+                return $step->getSelectionProcessingMode();
+            }
+        }
+
+        return SelectionProcessingMode::FOR_EACH;
+    }
+
     /**
      * @throws Exception
      */
     private function handleNextMessage(GenericExecutionEngineMessageInterface $message): void
     {
         $jobRun = $this->jobRunRepository->getJobRunById($message->getJobRunId());
-        if ($jobRun->getProcessedElementsForStep() === $jobRun->getTotalElements()) {
+
+        if ($this->getSelectionProcessingModeFromJobRun($jobRun) === SelectionProcessingMode::ONCE ||
+            $jobRun->getProcessedElementsForStep() === $jobRun->getTotalElements()) {
             $this->continueJobStepExecution($message);
         }
     }
@@ -224,27 +240,43 @@ private function handleJobExecutionError(
             $errorMessage,
         );
 
-        $errorHandling = $this->executionContext->getErrorHandlingFromContext($jobRun->getExecutionContext());
-        if ($errorHandling === null) {
-            $errorHandling = $this->errorHandlingMode;
-        }
-
-        match ($errorHandling) {
-            ErrorHandlingMode::STOP_ON_FIRST_ERROR->value =>
+        match
+        (
+            $this->getErrorHandlingMode(
+                $jobRun,
+                $this->getSelectionProcessingModeFromJobRun($jobRun)
+            )
+        ) {
+            ErrorHandlingMode::STOP_ON_FIRST_ERROR =>
             $this->stopJobExecutionOnError(
                 $jobRun,
                 $errorMessage
             ),
-            ErrorHandlingMode::CONTINUE_ON_ERROR->value =>
+            ErrorHandlingMode::CONTINUE_ON_ERROR =>
             $this->continueJobExecutionOnError(
                 $jobRun,
                 $message,
                 $errorMessage
-            ),
-            default => throw new InvalidErrorHandlingModeException(),
+            )
         };
     }
 
+    private function getErrorHandlingMode(
+        JobRun $jobRun,
+        SelectionProcessingMode $selectionProcessingMode
+    ): ErrorHandlingMode {
+        if($selectionProcessingMode === SelectionProcessingMode::ONCE) {
+            return ErrorHandlingMode::STOP_ON_FIRST_ERROR;
+        }
+
+        $errorHandling = $this->executionContext->getErrorHandlingFromContext($jobRun->getExecutionContext());
+        if ($errorHandling === null) {
+            $errorHandling = $this->errorHandlingMode;
+        }
+
+        return ErrorHandlingMode::from($errorHandling);
+    }
+
     /**
      * @throws Exception
      */
@@ -354,21 +386,13 @@ private function dispatchStepMessage(JobRun $jobRun): void
             return;
         }
 
-        $selectedElements = $job->getSelectedElements();
-        if (empty($selectedElements)) {
-            $this->executionEngineBus->dispatch(new $messageString($jobRun->getId(), $jobRun->getCurrentStep()));
-
-            return;
-        }
-
-        foreach ($selectedElements as $selectedElement) {
-            $this->executionEngineBus->dispatch(new $messageString(
-                $jobRun->getId(),
-                $jobRun->getCurrentStep(),
-                $selectedElement
-            )
-            );
-        }
+        $this->dispatchSelectedElements(
+            $jobRun->getId(),
+            $jobRun->getCurrentStep(),
+            $messageString,
+            $this->getSelectionProcessingModeFromJobRun($jobRun),
+            $job->getSelectedElements()
+        );
     }
 
     private function getLogParams(JobRun $jobRun): array
@@ -411,4 +435,34 @@ private function setCompletionState(JobRun $jobRun): void
             $this->getLogParams($jobRun)
         );
     }
+
+    /**
+     * @param ElementDescriptor[] $selectedElements
+     */
+    private function dispatchSelectedElements(
+        int $jobRunId,
+        int $currentStepId,
+        string $messageString,
+        SelectionProcessingMode $selectionProcessingMode,
+        array $selectedElements = []
+    ): void {
+        if (empty($selectedElements) || $selectionProcessingMode === SelectionProcessingMode::ONCE) {
+            $this->executionEngineBus->dispatch(new $messageString(
+                $jobRunId,
+                $currentStepId
+            )
+            );
+
+            return;
+        }
+
+        foreach ($selectedElements as $selectedElement) {
+            $this->executionEngineBus->dispatch(new $messageString(
+                $jobRunId,
+                $currentStepId,
+                $selectedElement
+            )
+            );
+        }
+    }
 }
diff --git a/bundles/GenericExecutionEngineBundle/src/Entity/JobRun.php b/bundles/GenericExecutionEngineBundle/src/Entity/JobRun.php
index 5ae61094d9f..9d86ba019ca 100644
--- a/bundles/GenericExecutionEngineBundle/src/Entity/JobRun.php
+++ b/bundles/GenericExecutionEngineBundle/src/Entity/JobRun.php
@@ -29,6 +29,7 @@
 use Symfony\Component\PropertyInfo\PropertyInfoExtractor;
 use Symfony\Component\Serializer\Encoder\JsonEncoder;
 use Symfony\Component\Serializer\Normalizer\ArrayDenormalizer;
+use Symfony\Component\Serializer\Normalizer\BackedEnumNormalizer;
 use Symfony\Component\Serializer\Normalizer\ObjectNormalizer;
 use Symfony\Component\Serializer\Serializer;
 use Symfony\Component\Serializer\SerializerInterface;
@@ -254,10 +255,25 @@ public function setProcessedElementsForStep(int $processedElementsForStep): void
     private function getSerializer(): SerializerInterface
     {
         if ($this->serializer === null) {
-            $encoder = [new JsonEncoder()];
-            $extractor = new PropertyInfoExtractor([], [new PhpDocExtractor(), new ReflectionExtractor()]);
-            $normalizer = [new ArrayDenormalizer(), new ObjectNormalizer(null, null, null, $extractor)];
-            $this->serializer = new Serializer($normalizer, $encoder);
+            $encoder = [
+                new JsonEncoder(),
+            ];
+            $extractor = new PropertyInfoExtractor(
+                [],
+                [
+                    new PhpDocExtractor(),
+                    new ReflectionExtractor(),
+                ]
+            );
+            $normalizer = [
+                new ArrayDenormalizer(),
+                new BackedEnumNormalizer(),
+                new ObjectNormalizer(null, null, null, $extractor),
+            ];
+            $this->serializer = new Serializer(
+                $normalizer,
+                $encoder
+            );
         }
 
         return $this->serializer;
diff --git a/bundles/GenericExecutionEngineBundle/src/Extractor/JobRunExtractor.php b/bundles/GenericExecutionEngineBundle/src/Extractor/JobRunExtractor.php
index ef08fdaee57..3aac5d8734f 100644
--- a/bundles/GenericExecutionEngineBundle/src/Extractor/JobRunExtractor.php
+++ b/bundles/GenericExecutionEngineBundle/src/Extractor/JobRunExtractor.php
@@ -22,6 +22,7 @@
 use Pimcore\Bundle\GenericExecutionEngineBundle\Model\JobStepInterface;
 use Pimcore\Bundle\GenericExecutionEngineBundle\Repository\JobRunRepositoryInterface;
 use Pimcore\Helper\SymfonyExpression\ExpressionServiceInterface;
+use Pimcore\Model\Element\ElementDescriptor;
 use Pimcore\Model\Element\ElementInterface;
 use Pimcore\Model\Element\Service;
 use Pimcore\Model\Exception\NotFoundException;
@@ -114,13 +115,14 @@ public function getElementFromMessage(
         array $types = [JobRunExtractorInterface::ASSET_TYPE]
     ): ?ElementInterface {
         $elementDescriptor = $message->getElement();
-        if (!$elementDescriptor || !in_array($elementDescriptor->getType(), $types, true)) {
+        if (!$elementDescriptor) {
             return null;
         }
 
-        $element = $this->getElement(
+        $element = $this->getElementByType(
             $elementDescriptor->getType(),
-            $elementDescriptor->getId()
+            $elementDescriptor->getId(),
+            $types
         );
 
         if (!$element) {
@@ -130,6 +132,31 @@ public function getElementFromMessage(
         return $element;
     }
 
+    public function getElementsFromMessage(
+        GenericExecutionEngineMessageInterface $message,
+        array $types = [JobRunExtractorInterface::ASSET_TYPE]
+    ): array {
+
+        $elementsToProcess = [];
+        $jobRun = $this->getJobRun($message);
+
+        /** @var ElementDescriptor[] $elementDescriptors */
+        $elementDescriptors = $jobRun->getJob()?->getSelectedElements();
+
+        foreach ($elementDescriptors as $elementDescriptor) {
+            $element = $this->getElementByType(
+                $elementDescriptor->getType(),
+                $elementDescriptor->getId(),
+                $types
+            );
+            if($element !== null) {
+                $elementsToProcess[] = $element;
+            }
+        }
+
+        return $elementsToProcess;
+    }
+
     private function getElement(string $type, int $id): ?ElementInterface
     {
         $element = Service::getElementById($type, $id);
@@ -140,4 +167,17 @@ private function getElement(string $type, int $id): ?ElementInterface
 
         return $element;
     }
+
+    private function getElementByType(
+        string $elementType,
+        int $elementId,
+        array $typesToLookFor = [JobRunExtractorInterface::ASSET_TYPE]): ?ElementInterface
+    {
+
+        if (!in_array($elementType, $typesToLookFor, true)) {
+            return null;
+        }
+
+        return $this->getElement($elementType, $elementId);
+    }
 }
diff --git a/bundles/GenericExecutionEngineBundle/src/Extractor/JobRunExtractorInterface.php b/bundles/GenericExecutionEngineBundle/src/Extractor/JobRunExtractorInterface.php
index 2fb30d3f2c1..5c42c7efa65 100644
--- a/bundles/GenericExecutionEngineBundle/src/Extractor/JobRunExtractorInterface.php
+++ b/bundles/GenericExecutionEngineBundle/src/Extractor/JobRunExtractorInterface.php
@@ -49,4 +49,10 @@ public function getElementFromMessage(
         GenericExecutionEngineMessageInterface $message,
         array $types = [JobRunExtractorInterface::ASSET_TYPE]
     ): ?ElementInterface;
+
+    /** @return ElementInterface[]  */
+    public function getElementsFromMessage(
+        GenericExecutionEngineMessageInterface $message,
+        array $types = [JobRunExtractorInterface::ASSET_TYPE]
+    ): array;
 }
diff --git a/bundles/GenericExecutionEngineBundle/src/Messenger/Handler/AbstractAutomationActionHandler.php b/bundles/GenericExecutionEngineBundle/src/Messenger/Handler/AbstractAutomationActionHandler.php
index f1987b565e7..d9158678bb9 100644
--- a/bundles/GenericExecutionEngineBundle/src/Messenger/Handler/AbstractAutomationActionHandler.php
+++ b/bundles/GenericExecutionEngineBundle/src/Messenger/Handler/AbstractAutomationActionHandler.php
@@ -33,7 +33,6 @@
 use Symfony\Contracts\Service\Attribute\Required;
 use Symfony\Contracts\Translation\TranslatorInterface;
 use Throwable;
-use UnexpectedValueException;
 use function array_key_exists;
 
 abstract class AbstractAutomationActionHandler
@@ -190,24 +189,23 @@ protected function isRunning(
     protected function getSubjectFromMessage(
         GenericExecutionEngineMessageInterface $message,
         array $types = [JobRunExtractorInterface::OBJECT_TYPE, JobRunExtractorInterface::ASSET_TYPE]
-    ): AbstractElement {
-        /** @var AbstractElement|null $subject */
-        $subject = $this->jobRunExtractor->getElementFromMessage(
-            $message,
-            $types
-        );
-
-        if (!$subject) {
-            throw new UnexpectedValueException(
-                sprintf(
-                    'No subject type found. Expected types: %s, found %s, %s',
-                    implode(', ', $types),
-                    $message->getElement()?->getId(),
-                    $message->getElement()?->getType()
-                )
-            );
-        }
+    ): ?AbstractElement {
+        /** @var AbstractElement $subject */
+        $subject = $this->jobRunExtractor->getElementFromMessage($message, $types);
 
         return $subject;
     }
+
+    /**
+     * @return AbstractElement[]
+     */
+    protected function getSubjectsFromMessage(
+        GenericExecutionEngineMessageInterface $message,
+        array $types = [JobRunExtractorInterface::OBJECT_TYPE, JobRunExtractorInterface::ASSET_TYPE]
+    ): array {
+        /** @var AbstractElement[] $subjects */
+        $subjects = $this->jobRunExtractor->getElementsFromMessage($message, $types);
+
+        return $subjects;
+    }
 }
diff --git a/bundles/GenericExecutionEngineBundle/src/Messenger/Messages/AbstractExecutionEngineMessage.php b/bundles/GenericExecutionEngineBundle/src/Messenger/Messages/AbstractExecutionEngineMessage.php
index 9b36e70bb93..e45d584d9a6 100644
--- a/bundles/GenericExecutionEngineBundle/src/Messenger/Messages/AbstractExecutionEngineMessage.php
+++ b/bundles/GenericExecutionEngineBundle/src/Messenger/Messages/AbstractExecutionEngineMessage.php
@@ -21,10 +21,14 @@
 
 abstract class AbstractExecutionEngineMessage implements GenericExecutionEngineMessageInterface
 {
+    /**
+     * @param ElementDescriptor[] $elements
+     */
     public function __construct(
         protected int $jobRunId,
         protected int $currentJobStep,
-        protected ?ElementDescriptor $element = null
+        protected ?ElementDescriptor $element = null,
+        protected array $elements = []
     ) {
     }
 
diff --git a/bundles/GenericExecutionEngineBundle/src/Model/JobStep.php b/bundles/GenericExecutionEngineBundle/src/Model/JobStep.php
index 41d69617a86..de2f5ad38e4 100644
--- a/bundles/GenericExecutionEngineBundle/src/Model/JobStep.php
+++ b/bundles/GenericExecutionEngineBundle/src/Model/JobStep.php
@@ -16,13 +16,16 @@
 
 namespace Pimcore\Bundle\GenericExecutionEngineBundle\Model;
 
+use Pimcore\Bundle\GenericExecutionEngineBundle\Utils\Enums\SelectionProcessingMode;
+
 final class JobStep implements JobStepInterface
 {
     public function __construct(
         private readonly string $name,
         private readonly string $messageFQCN,
         private readonly string $condition,
-        private readonly array $config
+        private readonly array $config,
+        private readonly SelectionProcessingMode $selectionProcessingMode = SelectionProcessingMode::FOR_EACH
     ) {
     }
 
@@ -45,4 +48,9 @@ public function getCondition(): string
     {
         return $this->condition;
     }
+
+    public function getSelectionProcessingMode(): SelectionProcessingMode
+    {
+        return $this->selectionProcessingMode;
+    }
 }
diff --git a/bundles/GenericExecutionEngineBundle/src/Model/JobStepInterface.php b/bundles/GenericExecutionEngineBundle/src/Model/JobStepInterface.php
index b7e0779a6bc..74c188c263f 100644
--- a/bundles/GenericExecutionEngineBundle/src/Model/JobStepInterface.php
+++ b/bundles/GenericExecutionEngineBundle/src/Model/JobStepInterface.php
@@ -16,6 +16,8 @@
 
 namespace Pimcore\Bundle\GenericExecutionEngineBundle\Model;
 
+use Pimcore\Bundle\GenericExecutionEngineBundle\Utils\Enums\SelectionProcessingMode;
+
 interface JobStepInterface
 {
     public function getName(): string;
@@ -25,4 +27,6 @@ public function getMessageFQCN(): string;
     public function getConfig(): array;
 
     public function getCondition(): string;
+
+    public function getSelectionProcessingMode(): SelectionProcessingMode;
 }
diff --git a/bundles/GenericExecutionEngineBundle/src/Exception/InvalidErrorHandlingModeException.php b/bundles/GenericExecutionEngineBundle/src/Utils/Enums/SelectionProcessingMode.php
similarity index 72%
rename from bundles/GenericExecutionEngineBundle/src/Exception/InvalidErrorHandlingModeException.php
rename to bundles/GenericExecutionEngineBundle/src/Utils/Enums/SelectionProcessingMode.php
index 4005869e468..984b926f223 100644
--- a/bundles/GenericExecutionEngineBundle/src/Exception/InvalidErrorHandlingModeException.php
+++ b/bundles/GenericExecutionEngineBundle/src/Utils/Enums/SelectionProcessingMode.php
@@ -14,13 +14,10 @@
  *  @license    http://www.pimcore.org/license     GPLv3 and PCL
  */
 
-namespace Pimcore\Bundle\GenericExecutionEngineBundle\Exception;
+namespace Pimcore\Bundle\GenericExecutionEngineBundle\Utils\Enums;
 
-use RuntimeException;
-
-/**
- * @internal
- */
-final class InvalidErrorHandlingModeException extends RuntimeException
+enum SelectionProcessingMode: string
 {
+    case ONCE = 'once';
+    case FOR_EACH = 'for_each';
 }
diff --git a/doc/19_Development_Tools_and_Details/08_Generic_Execution_Engine/01_Usage.md b/doc/19_Development_Tools_and_Details/08_Generic_Execution_Engine/01_Usage.md
index 1d687625bf2..b660e4ed1c4 100644
--- a/doc/19_Development_Tools_and_Details/08_Generic_Execution_Engine/01_Usage.md
+++ b/doc/19_Development_Tools_and_Details/08_Generic_Execution_Engine/01_Usage.md
@@ -13,8 +13,8 @@ Create a job as follows
 $job = new Job(
   'my-first-job',
   [
-    new JobStep('Step 1', MyFirstTestMessage::class, '', []),
-    new JobStep('Step 2', MySecondTestMessage::class, '', []),
+    new JobStep('Step 1', MyFirstTestMessage::class, '', [], StepSelectionMode::FOR_EACH),
+    new JobStep('Step 2', MySecondTestMessage::class, '', [], StepSelectionMode::FOR_EACH),
   ],
   [new ElementDescriptor('object', 234)],
   [
@@ -74,3 +74,17 @@ $this->jobRunRepository->updateLogLocalized(
 ## Job Run Error Logs
 Job Run Error Logs are the entity for storing log information about Job Run. There is the
 [`Pimcore\Bundle\GenericExecutionEngineBundle\Repository\JobRunErrorLogRepositoryInterface`] for all kind of CRUD operations on these logs.
+
+## Accessing JobStep in Handler
+The `JobStep` object can be accessed in the handler via the `getJobStep` method. This can be useful if you need to access the step configuration in the handler.
+
+```php
+$jobRun = $this->getJobRun($message);
+$steps = $jobRun->getJob()?->getSteps();
+if($steps !== null) {
+    $step = $steps[$jobRun->getCurrentStep()] ?? null;
+    if($step) {
+        return $step->getSelectionMode();
+    }
+}
+```
diff --git a/doc/19_Development_Tools_and_Details/08_Generic_Execution_Engine/04_Step_Configuration.md b/doc/19_Development_Tools_and_Details/08_Generic_Execution_Engine/04_Step_Configuration.md
new file mode 100644
index 00000000000..78c42cfec04
--- /dev/null
+++ b/doc/19_Development_Tools_and_Details/08_Generic_Execution_Engine/04_Step_Configuration.md
@@ -0,0 +1,36 @@
+# Step configuration
+
+The configuration of a step is done via the `JobStep` object. 
+The `JobStep` is part of the `Job` object and contains the following properties:
+
+- `name`: The name of the step.
+- `messageClass`: The message class that should be dispatched.
+- `configuration`: The configuration of the step.
+- `selectionMode`: The selection mode of the step.
+
+## Name
+The name of the step is a string that is used to identify the step in the job run log.
+
+## Message class
+The message class is the class of the message that should be dispatched when the step is executed.
+
+## Configuration
+The configuration of a step is an array that can contain any kind of data that is necessary for the step handler to execute the step.
+
+## Selection Processing mode
+The selection processing mode of a step is an enum that defines how the step should process the selected elements.
+
+- `SelectionProcessingMode::FOR_EACH`: The step is executed for each selected element.
+For example if you pass 10 selected elements to the job, the message gets dispatched 10 times and the step handler therefore is executed 10 times.
+Use `getSubjectFromMessage()` in `AbstractAutomationActionHandler` method to access the current element in the handler.
+
+- `SelectionProcessingMode::FOR_ALL`: The step is executed once for all selected elements.
+If you pass 10 selected elements to the job, the message gets dispatched once and the step handler therefore is executed once.
+Use `getSubjectsFromMessage()` in `AbstractAutomationActionHandler` method to access all elements in the handler.
+
+For further information on how to access the JobStep configuration in the step handler, see [01_Usage](./01_Usage.md).
+
+
+
+
+


Test patch: