feat: Live update execution diagram as etl runs

Author: oliverde8

Repository/EtlExecutionRepository.php

@@ -14,15 +14,33 @@
  */
 class EtlExecutionRepository extends ServiceEntityRepository
 {
-    public function __construct(ManagerRegistry $registry)
+    public function __construct(protected readonly ManagerRegistry $registry)
     {
         parent::__construct($registry, EtlExecution::class);
     }
 
     public function save(EtlExecution $execution)
     {
         $this->_em->persist($execution);
-        $this->_em->flush();
+        $this->_em->flush($execution);
+    }
+
+    public function updateStepStats(EtlExecution $execution, string $stepStats)
+    {
+        /** @var \Doctrine\DBAL\Connection $connection */
+        $connection = $this->registry->getConnection();
+
+        $query = $connection->createQueryBuilder()
+            ->update('EtlExecution', 'e')
+            ->set('e.stepStats', ':stepStats')
+            ->where('e.id = :executionId')
+            ->setParameter('stepStats', $stepStats)
+            ->setParameter('executionId', $execution->getId());
+        $query->execute();
+
+        if ($connection->getTransactionNestingLevel() > 0) {
+            $query->getConnection()->commit();
+        }
     }
 
     public function getCountInStatus(\DateTime $startTime, \DateTime $endTime, string $status): int

Services/ChainProcessorsManager.php

@@ -127,13 +127,14 @@ public function executeFromEtlEntity(EtlExecution $execution, iterable $iterator
             // Start the process.
             $observerProcessTime = 0;
             $processor->process($iterator, $params, function (array $operationStates, int $processedItems, int $returnedItems, bool $hasFinished = false) use ($observerCallback, &$observerProcessTime, $execution) {
-                $observerCallback($operationStates, $processedItems, $returnedItems, $hasFinished);
+                if ($observerCallback) {
+                    $observerCallback($operationStates, $processedItems, $returnedItems, $hasFinished);
+                }
 
                 if ((time() - $observerProcessTime) > 5 || $hasFinished) {
-                    $execution = $this->etlExecutionRepository->find($execution->getId());
-                    $execution->setStepStats(json_encode($operationStates));
-
-                    $this->etlExecutionRepository->save($execution);
+                    $jsonStates = json_encode($operationStates);
+                    $execution->setStepStats($jsonStates);
+                    $this->etlExecutionRepository->updateStepStats($execution, $jsonStates);
                     $observerProcessTime = time();
                 }
             });

assets/controllers/graph_reload.js

@@ -0,0 +1,43 @@
+import { Controller } from '@hotwired/stimulus';
+import mermaid from 'mermaid';
+
+export default class extends Controller {
+  static values = { continueUpdate: Boolean, refreshInterval: Number, apiUrl: String }
+
+  async connect() {
+    console.log(this.continueUpdateValue);
+    console.log(this.refreshIntervalValue);
+    console.log(this.apiUrlValue);
+    // console.log(mermaid);
+
+    mermaid.initialize({startOnLoad: false});
+    this.mermaid = await mermaid.run({
+      nodes: [this.element],
+    });
+
+    if (this.continueUpdateValue) {
+      this.startTimer(this.refreshIntervalValue);
+    }
+  }
+
+  startTimer(timer) {
+    let that = this;
+    setTimeout(function () {
+      fetch(that.apiUrlValue, {method: "get"})
+        .then(response => {
+          return response.json();
+        }).then(async data => {
+        if (data.continueUpdate === 'false') {
+          location.reload()
+          return;
+        }
+
+        console.log(data);
+        const {svg, bindFunctions} = await mermaid.render('run-graph', data.graph);
+        that.element.innerHTML = svg;
+
+        that.startTimer(data.refreshInterval);
+      })
+    }, timer * 1000);
+  }
+}

assets/package.json

@@ -0,0 +1,26 @@
+{
+  "name": "@oliverde8/php-etl-graph-reload",
+  "description": "",
+  "version": "1.0.0",
+    "license": "MIT",
+    "symfony": {
+        "controllers": {
+            "oliverde8-etl-graph-reload": {
+                "main": "controllers/graph_reload.js",
+                "name": "oliverde8-etl-graph-reload",
+                "webpackMode": "eager",
+                "fetch": "eager",
+                "enabled": true
+            }
+        },
+        "importmap": {
+            "@hotwired/stimulus": "^3.0.0"
+        }
+    },
+    "dependencies": {
+      "mermaid": "^1.13"
+    },
+    "peerDependencies": {
+        "@hotwired/stimulus": "^3.0.0"
+    }
+}

composer.json

@@ -3,6 +3,7 @@
   "type": "library",
   "description": "Allow usage of the PHP-ETL library in symfony framework.",
   "license": "MIT",
+  "keywords": ["symfony-ux"],
   "authors": [
     {
       "name": "Oliver de Cramer",