Code Coverage
 
Lines
Functions and Methods
Classes and Traits
Total
98.32%
117 / 119
71.43%
5 / 7
 CRAP
0.00%
0 / 1
WorkflowDagEngine
98.31%
116 / 118
71.43%
5 / 7
 25
0.00%
0 / 1
 __construct
100.00%
5 / 5
100.00%
1 / 1
 1
 executeBeforeHook
100.00%
18 / 18
100.00%
1 / 1
 5
 executeAfterHook
100.00%
14 / 14
100.00%
1 / 1
 4
 shouldSkipExecution
66.67%
2 / 3
0.00%
0 / 1
 3.33
 runSingleWorkflow
100.00%
50 / 50
100.00%
1 / 1
 4
 traverseDag
93.33%
14 / 15
0.00%
0 / 1
 5.01
 executeNode
100.00%
13 / 13
100.00%
1 / 1
 3
1<?php
2
3declare(strict_types=1);
4
5/** @license For full copyright and license information, please see the LICENSE.md file. */
6
7namespace App\Modules\Automation\Application\Service;
8
9defined('AMMONLY_APP') || exit('Direct script access is forbidden.');
10
11use App\Core\Engine\Domain\Exception\WorkflowValidationException;
12use App\Core\Engine\Domain\Model\PermissionContext;
13use App\Core\Engine\Domain\Service\WorkflowEngineInterface;
14use App\Modules\Automation\Application\Service\Dag\WorkflowActionExecutor;
15use App\Modules\Automation\Application\Service\Dag\WorkflowConditionEvaluator;
16use App\Modules\Automation\Application\Service\Dag\WorkflowDagParser;
17use App\Modules\Automation\Domain\Model\Workflow;
18use App\Modules\Automation\Domain\Model\WorkflowExecutionContext;
19use App\Modules\Automation\Domain\Repository\WorkflowRepositoryInterface;
20use PDO;
21use Throwable;
22
23/**
24 * Workflow Directed Acyclic Graph (DAG) Execution Engine.
25 *
26 * Traverses visual Drawflow node trees, evaluates conditional branches,
27 * resolves relations, creates records with prefixes, and dispatches queued emails.
28 * Delegates node parsing, condition evaluation, and action execution to specialized subcomponents.
29 *
30 * @package App\Modules\Automation\Application\Service
31 */
32final readonly class WorkflowDagEngine implements WorkflowEngineInterface
33{
34    public const string OUTPUT_1 = 'output_1';
35    public const string OUTPUT_2 = 'output_2';
36    private const string STATUS_SUCCESS = 'success';
37    private const string STATUS_FAILED = 'failed';
38    private const string STATUS_BLOCKED = 'blocked_by_guard';
39
40    private WorkflowDagParser $parser;
41    private WorkflowConditionEvaluator $conditionEvaluator;
42    private WorkflowActionExecutor $actionExecutor;
43
44    /**
45     * WorkflowDagEngine constructor.
46     *
47     * @param WorkflowRepositoryInterface      $workflowRepo       Workflow persistence repository.
48     * @param PDO|null                         $pdo                Database connection handle.
49     * @param string                           $tablePrefix        Database table prefix.
50     * @param WorkflowDagParser|null           $parser             Optional DAG graph parser.
51     * @param WorkflowConditionEvaluator|null  $conditionEvaluator Optional condition evaluator.
52     * @param WorkflowActionExecutor|null      $actionExecutor     Optional action executor.
53     */
54    public function __construct(
55        private WorkflowRepositoryInterface $workflowRepo,
56        private ?PDO $pdo = null,
57        private string $tablePrefix = 'a_',
58        ?WorkflowDagParser $parser = null,
59        ?WorkflowConditionEvaluator $conditionEvaluator = null,
60        ?WorkflowActionExecutor $actionExecutor = null
61    ) {
62        $this->parser = $parser ?? new WorkflowDagParser();
63        $this->conditionEvaluator = $conditionEvaluator
64            ?? new WorkflowConditionEvaluator($this->pdo, $this->tablePrefix);
65        $this->actionExecutor = $actionExecutor
66            ?? new WorkflowActionExecutor($this->pdo, $this->tablePrefix);
67    }
68
69    /**
70     * {@inheritdoc}
71     */
72    public function executeBeforeHook(
73        string            $moduleName,
74        string            $triggerType,
75        array             $data,
76        ?array            $snapshot,
77        PermissionContext $context
78    ): array {
79        $workflows = $this->workflowRepo->findActiveByTrigger($moduleName, $triggerType);
80        if ($workflows === []) {
81            return $data;
82        }
83
84        $recordIdRaw = $snapshot['id'] ?? $data['id'] ?? null;
85        $recordId = $recordIdRaw !== null ? (int) $recordIdRaw : null;
86        $currentData = $data;
87
88        foreach ($workflows as $workflow) {
89            if ($this->shouldSkipExecution($workflow, $recordId)) {
90                continue;
91            }
92
93            $currentData = $this->runSingleWorkflow(
94                $workflow,
95                $triggerType,
96                $moduleName,
97                $recordId,
98                $currentData,
99                $snapshot,
100                $context
101            );
102        }
103
104        return $currentData;
105    }
106
107    /**
108     * Executes after-save triggers asynchronously or synchronously.
109     *
110     * @param string                 $moduleName  Module name.
111     * @param string                 $triggerType Stage ('after_create', 'after_update', 'after_delete').
112     * @param int                    $recordId    Primary key of the saved record.
113     * @param array<string, mixed>   $data        Saved data payload.
114     * @param array<string, mixed>|null $snapshot Previous database snapshot.
115     * @param PermissionContext      $context     Security context.
116     */
117    public function executeAfterHook(
118        string            $moduleName,
119        string            $triggerType,
120        int               $recordId,
121        array             $data,
122        ?array            $snapshot,
123        PermissionContext $context
124    ): void {
125        $workflows = $this->workflowRepo->findActiveByTrigger($moduleName, $triggerType);
126        if ($workflows === []) {
127            return;
128        }
129
130        foreach ($workflows as $workflow) {
131            if ($this->shouldSkipExecution($workflow, $recordId)) {
132                continue;
133            }
134
135            $this->runSingleWorkflow(
136                $workflow,
137                $triggerType,
138                $moduleName,
139                $recordId,
140                $data,
141                $snapshot,
142                $context
143            );
144        }
145    }
146
147    /**
148     * Determines whether execution should be skipped based on frequency settings.
149     *
150     * @param Workflow $workflow Target workflow.
151     * @param int|null $recordId Target record ID.
152     * @return bool True if execution should be skipped.
153     */
154    private function shouldSkipExecution(Workflow $workflow, ?int $recordId): bool
155    {
156        if ($workflow->executionFrequency === 'once_per_record' && $recordId !== null) {
157            return $this->workflowRepo->hasRunForRecord((int) $workflow->id, $recordId);
158        }
159
160        return false;
161    }
162
163    /**
164     * Executes a single active workflow over the record payload.
165     *
166     * @param Workflow               $workflow    Target workflow definition.
167     * @param string                 $triggerType Trigger event type.
168     * @param string                 $moduleName  Module machine name.
169     * @param int|null               $recordId    Target record ID.
170     * @param array<string, mixed>   $data        Current data payload.
171     * @param array<string, mixed>|null $snapshot Database snapshot.
172     * @param PermissionContext      $context     Security context.
173     * @return array<string, mixed> Potentially mutated data payload.
174     */
175    private function runSingleWorkflow(
176        Workflow          $workflow,
177        string            $triggerType,
178        string            $moduleName,
179        ?int              $recordId,
180        array             $data,
181        ?array            $snapshot,
182        PermissionContext $context
183    ): array {
184        $startTime = microtime(true);
185        $runId = bin2hex(random_bytes(8));
186        $execContext = new WorkflowExecutionContext(
187            runId: $runId,
188            workflowId: (int) $workflow->id,
189            triggerType: $triggerType,
190            moduleName: $moduleName,
191            recordId: $recordId,
192            currentData: $data,
193            previousSnapshot: $snapshot,
194            actorUserId: $context->actorUserId
195        );
196
197        $nodes = $this->parser->parseGraphNodes($workflow);
198        if ($nodes === []) {
199            return $data;
200        }
201
202        try {
203            $execContext = $this->traverseDag($nodes, $execContext, $workflow);
204            $durationMs = (int) round((microtime(true) - $startTime) * 1000);
205
206            $this->workflowRepo->recordRun(
207                (int) $workflow->id,
208                $recordId,
209                $triggerType,
210                self::STATUS_SUCCESS,
211                $durationMs,
212                null,
213                ['record_id' => $recordId]
214            );
215            $this->workflowRepo->incrementRuns((int) $workflow->id);
216
217            return $execContext->currentData;
218        } catch (WorkflowValidationException $e) {
219            $durationMs = (int) round((microtime(true) - $startTime) * 1000);
220            $this->workflowRepo->recordRun(
221                (int) $workflow->id,
222                $recordId,
223                $triggerType,
224                self::STATUS_BLOCKED,
225                $durationMs,
226                $e->getMessage()
227            );
228            throw $e;
229        } catch (Throwable $e) {
230            $durationMs = (int) round((microtime(true) - $startTime) * 1000);
231            $this->workflowRepo->recordRun(
232                (int) $workflow->id,
233                $recordId,
234                $triggerType,
235                self::STATUS_FAILED,
236                $durationMs,
237                $e->getMessage()
238            );
239            throw $e;
240        }
241    }
242
243    /**
244     * Traverses the DAG nodes breadth-first from the initial trigger node.
245     *
246     * @param array<string, array<string, mixed>> $nodes       All canvas nodes.
247     * @param WorkflowExecutionContext            $execContext Active context.
248     * @param Workflow                            $workflow    Workflow model.
249     * @return WorkflowExecutionContext Resulting context.
250     */
251    private function traverseDag(
252        array                    $nodes,
253        WorkflowExecutionContext $execContext,
254        Workflow                 $workflow
255    ): WorkflowExecutionContext {
256        $triggerNodeId = $this->parser->findTriggerNodeId($nodes);
257        if ($triggerNodeId === null) {
258            return $execContext;
259        }
260
261        $queue = [$triggerNodeId];
262        $visited = [];
263
264        while ($queue !== []) {
265            $nodeId = array_shift($queue);
266            if (isset($visited[$nodeId]) || !isset($nodes[$nodeId])) {
267                continue;
268            }
269            $visited[$nodeId] = true;
270
271            $node = $nodes[$nodeId];
272            $nodeType = (string) ($node['name'] ?? ($node['data']['node_type'] ?? ''));
273
274            $branchToFollow = self::OUTPUT_1;
275            $execContext = $this->executeNode($nodeType, $node, $execContext, $workflow, $branchToFollow);
276
277            $this->parser->enqueueNextNodes($node, $branchToFollow, $visited, $queue);
278        }
279
280        return $execContext;
281    }
282
283    /**
284     * Executes an individual node logic and decides which branch port to follow.
285     *
286     * @param string                   $nodeType       Node type machine name.
287     * @param array<string, mixed>     $node           Full node definition.
288     * @param WorkflowExecutionContext $execContext    Active context.
289     * @param Workflow                 $workflow       Workflow model.
290     * @param string                   $branchToFollow Output branch chosen ('output_1' or 'output_2').
291     * @return WorkflowExecutionContext Updated context.
292     */
293    private function executeNode(
294        string                   $nodeType,
295        array                    $node,
296        WorkflowExecutionContext $execContext,
297        Workflow                 $workflow,
298        string                   &$branchToFollow
299    ): WorkflowExecutionContext {
300        if (str_contains($nodeType, 'condition') || $nodeType === 'condition_if') {
301            return $this->conditionEvaluator->executeConditionNode(
302                $nodeType,
303                $node,
304                $execContext,
305                $branchToFollow
306            );
307        }
308
309        return $this->actionExecutor->executeActionNode(
310            $nodeType,
311            $node,
312            $execContext,
313            $workflow
314        );
315    }
316}