feat: 🔥Added real time observation.

CHANGELOG.md

@@ -1,3 +1,10 @@
+# 2.1.0
+
+- :star2: Support for php-etl 2.1's new operations (`Grouping\BatchOperation`, `IfOperation`, `SwitchOperation`).
+- :star2: Live execution graph observability: t
+- :star2: Optional real-time updates over Mercure. The graph degrades to polling / static without it.
+- :collision: Removed the old Mermaid `graph_reload` Stimulus controller (`assets/`), superseded by the new Cytoscape widget.
+
 # 2.0.0
 
 - :star2: Support for php-etl 2 (**Breaking Change**)

Controller/ExecutionObservabilityController.php

@@ -0,0 +1,172 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Oliverde8\PhpEtlBundle\Controller;
+
+use Oliverde8\PhpEtlBundle\Entity\EtlExecution;
+use Oliverde8\PhpEtlBundle\Graph\ChainGraphBuilder;
+use Oliverde8\PhpEtlBundle\Graph\RunStateNormalizer;
+use Oliverde8\PhpEtlBundle\Repository\EtlExecutionRepository;
+use Oliverde8\PhpEtlBundle\Security\EtlExecutionVoter;
+use Oliverde8\PhpEtlBundle\Services\ChainProcessorsManager;
+use Oliverde8\PhpEtlBundle\Services\ChainWorkDirManager;
+use Oliverde8\PhpEtlBundle\Services\ExecutionContextFactory;
+use Symfony\Bundle\FrameworkBundle\Controller\AbstractController;
+use Symfony\Component\HttpFoundation\JsonResponse;
+use Symfony\Component\HttpFoundation\Request;
+use Symfony\Component\Routing\Attribute\Route;
+
+/**
+ * Framework-native (not EasyAdmin-coupled) JSON endpoints that power the live
+ * execution graph. Any Symfony frontend (EasyAdmin, Sylius, custom) mounts
+ * these routes and reuses them; the shipped JS controller talks to them.
+ *
+ *  - GET .../graph  → static topology + the last persisted run-state (static fallback)
+ *  - GET .../state  → latest persisted run-state (poll fallback when no Mercure)
+ *  - GET .../logs   → incremental log tail (offset in lines)
+ *
+ * All three are read-only and guarded by {@see EtlExecutionVoter::VIEW}.
+ */
+class ExecutionObservabilityController extends AbstractController
+{
+    private const int LOG_BATCH = 1000;
+
+    public function __construct(
+        private readonly EtlExecutionRepository $executions,
+        private readonly ChainGraphBuilder $graphBuilder,
+        private readonly RunStateNormalizer $stateNormalizer,
+        private readonly ChainProcessorsManager $chainProcessorManager,
+        private readonly ChainWorkDirManager $workDirManager,
+        private readonly ExecutionContextFactory $executionContextFactory,
+    ) {
+    }
+
+    #[Route('/etl/executions/{id}/graph', name: 'oliverde8_etl_execution_graph', requirements: ['id' => '\d+'], methods: ['GET'])]
+    public function graph(int $id): JsonResponse
+    {
+        $execution = $this->findExecution($id);
+
+        $data = [
+            'execution' => $this->executionMeta($execution),
+            'graph' => ['nodes' => [], 'edges' => []],
+            'state' => $this->stateNormalizer->normalize($execution->getStepStats()),
+            'error' => null,
+        ];
+
+        try {
+            $processor = $this->chainProcessorManager->getProcessor($execution->getName(), $this->options($execution));
+            $data['graph'] = $this->graphBuilder->build($processor);
+        } catch (\Throwable $e) {
+            // The definition may have changed or been removed since the run: still
+            // return the (topology-less) persisted state so the UI degrades cleanly.
+            $data['error'] = $e->getMessage();
+        }
+
+        return new JsonResponse($data);
+    }
+
+    #[Route('/etl/executions/{id}/state', name: 'oliverde8_etl_execution_state', requirements: ['id' => '\d+'], methods: ['GET'])]
+    public function state(int $id): JsonResponse
+    {
+        $execution = $this->findExecution($id);
+
+        return new JsonResponse([
+            'status' => $execution->getStatus(),
+            'finished' => $this->isFinished($execution),
+            'state' => $this->stateNormalizer->normalize($execution->getStepStats()),
+        ]);
+    }
+
+    #[Route('/etl/executions/{id}/logs', name: 'oliverde8_etl_execution_logs', requirements: ['id' => '\d+'], methods: ['GET'])]
+    public function logs(int $id, Request $request): JsonResponse
+    {
+        $execution = $this->findExecution($id);
+
+        $offset = max(0, $request->query->getInt('offset'));
+        $lines = $this->readLogLines($execution);
+        $slice = \array_slice($lines, $offset, self::LOG_BATCH);
+
+        return new JsonResponse([
+            'lines' => array_values($slice),
+            'offset' => $offset + \count($slice),
+            'more' => \count($lines) > $offset + \count($slice),
+            'finished' => $this->isFinished($execution),
+        ]);
+    }
+
+    private function findExecution(int $id): EtlExecution
+    {
+        $execution = $this->executions->find($id);
+        if (!$execution instanceof EtlExecution) {
+            throw $this->createNotFoundException("Unknown ETL execution $id");
+        }
+
+        $this->denyAccessUnlessGranted(EtlExecutionVoter::VIEW, $execution);
+
+        return $execution;
+    }
+
+    /** @return array<string, mixed> */
+    private function options(EtlExecution $execution): array
+    {
+        $options = json_decode((string) $execution->getInputOptions(), true);
+
+        return \is_array($options) ? $options : [];
+    }
+
+    private function isFinished(EtlExecution $execution): bool
+    {
+        return \in_array(
+            $execution->getStatus(),
+            [EtlExecution::STATUS_SUCCESS, EtlExecution::STATUS_FAILURE],
+            true,
+        );
+    }
+
+    /** @return array<string, mixed> */
+    private function executionMeta(EtlExecution $execution): array
+    {
+        return [
+            'id' => $execution->getId(),
+            'name' => $execution->getName(),
+            'username' => $execution->getUsername(),
+            'status' => $execution->getStatus(),
+            'finished' => $this->isFinished($execution),
+            'createTime' => $execution->getCreateTime()->format(\DATE_ATOM),
+            'startTime' => $execution->getStartTime()?->format(\DATE_ATOM),
+            'endTime' => $execution->getEndTime()?->format(\DATE_ATOM),
+        ];
+    }
+
+    /** @return string[] */
+    private function readLogLines(EtlExecution $execution): array
+    {
+        // While running, the log lives in the local tmp work dir; after the
+        // context is finalised it may have been copied to the execution filesystem.
+        $localLog = $this->workDirManager->getLocalTmpWorkDir($execution, false).'/execution.log';
+        if (is_file($localLog)) {
+            $content = @file($localLog, \FILE_IGNORE_NEW_LINES);
+
+            return false === $content ? [] : $content;
+        }
+
+        try {
+            $fileSystem = $this->executionContextFactory->get(['etl' => ['execution' => $execution]])->getFileSystem();
+            if ($fileSystem->fileExists('execution.log')) {
+                $stream = $fileSystem->readStream('execution.log');
+                $lines = [];
+                while (false !== ($line = fgets($stream))) {
+                    $lines[] = rtrim($line, "\r\n");
+                }
+                fclose($stream);
+
+                return $lines;
+            }
+        } catch (\Throwable) {
+            // Best-effort tail: never let a missing/remote log break the endpoint.
+        }
+
+        return [];
+    }
+}

DependencyInjection/Oliverde8PhpEtlExtension.php

@@ -22,5 +22,26 @@ public function load(array $configs, ContainerBuilder $container): void
         $loader->load('service-rule-transformers.yml');
         $loader->load('service-operation-factories.yml');
         $loader->load('service-operations-v2.yml');
+
+        // ChainBuilderV2 operations that only exist on newer php-etl versions
+        // (Grouping/BatchOperation, IfOperation, SwitchOperation — all added in
+        // 2.1). The whole file loads or none of it does, so every class it
+        // references must be checked here — a service definition pointing at a
+        // missing class fails container compilation, unlike a plain instanceof
+        // check. This keeps the bundle working against older versions that
+        // don't have them yet.
+        if (class_exists(\Oliverde8\Component\PhpEtl\ChainOperation\Grouping\BatchOperation::class)
+            && class_exists(\Oliverde8\Component\PhpEtl\ChainOperation\IfOperation::class)
+            && class_exists(\Oliverde8\Component\PhpEtl\ChainOperation\SwitchOperation::class)
+        ) {
+            $loader->load('service-operations-v2-optional.yml');
+        }
+
+        // Optional real-time layer: only wire the Mercure publisher when the
+        // component is actually installed. Without it the bundle keeps the
+        // no-op publisher and the graph degrades to polling / static.
+        if (interface_exists(\Symfony\Component\Mercure\HubInterface::class)) {
+            $loader->load('services-mercure.yml');
+        }
     }
 }

Graph/ChainGraph.php

@@ -0,0 +1,50 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Oliverde8\PhpEtlBundle\Graph;
+
+/**
+ * Framework-agnostic description of a chain's static topology: the nodes
+ * (operations / splits) and the directed edges between them. Serialises to
+ * {"nodes": [...], "edges": [{"source": id, "target": id}, ...]} for any
+ * frontend (EasyAdmin, Sylius, custom) to render.
+ */
+final class ChainGraph implements \JsonSerializable
+{
+    /** @var GraphNode[] */
+    private array $nodes = [];
+
+    /** @var array<int, array{source: string, target: string}> */
+    private array $edges = [];
+
+    public function addNode(GraphNode $node): void
+    {
+        $this->nodes[] = $node;
+    }
+
+    public function addEdge(string $source, string $target): void
+    {
+        $this->edges[] = ['source' => $source, 'target' => $target];
+    }
+
+    /** @return GraphNode[] */
+    public function getNodes(): array
+    {
+        return $this->nodes;
+    }
+
+    /** @return array<int, array{source: string, target: string}> */
+    public function getEdges(): array
+    {
+        return $this->edges;
+    }
+
+    public function jsonSerialize(): array
+    {
+        return [
+            'nodes' => $this->nodes,
+            'edges' => $this->edges,
+        ];
+    }
+}

Graph/ChainGraphBuilder.php

@@ -0,0 +1,99 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Oliverde8\PhpEtlBundle\Graph;
+
+use Oliverde8\Component\PhpEtl\ChainOperation\SubChainsAwareOperationInterface;
+use Oliverde8\Component\PhpEtl\ChainProcessorInterface;
+
+/**
+ * Builds the static {@see ChainGraph} topology from a chain processor.
+ *
+ * Mirrors the traversal of the core MermaidStaticOutput (php-etl) but emits a
+ * framework-agnostic node/edge structure instead of Mermaid text, and uses
+ * dotted-path node ids that match {@see RunStateNormalizer} so live/persisted
+ * run-state can be overlaid on the topology.
+ *
+ * Edge semantics follow the core: within a chain, link N connects to link N+1;
+ * a branch-holding operation (split, merge, ...) connects to the first node of
+ * each branch, and the main line continues from that operation's own node
+ * (branches do not rejoin automatically).
+ *
+ * {@see SubChainsAwareOperationInterface} is only implemented by php-etl 2.1+
+ * (Split and Merge so far). Checking `instanceof` against it is safe even when
+ * the bundle runs against php-etl 2.0 — PHP evaluates instanceof as false for
+ * a non-existent class rather than erroring — so this bundle keeps working
+ * with either version; branch-holding operations older than 2.1 just render
+ * as a single opaque node, same as before.
+ */
+final class ChainGraphBuilder
+{
+    public function build(ChainProcessorInterface $processor): ChainGraph
+    {
+        $graph = new ChainGraph();
+        $this->addNodes($processor, '', $graph);
+        $this->addEdges($processor, '', null, $graph);
+
+        return $graph;
+    }
+
+    private function addNodes(ChainProcessorInterface $processor, string $prefix, ChainGraph $graph): void
+    {
+        $names = $processor->getChainLinkNames();
+
+        foreach ($processor->getChainLinks() as $index => $link) {
+            $id = '' === $prefix ? (string) $index : "$prefix.$index";
+            $isBranching = $link instanceof SubChainsAwareOperationInterface;
+            $type = $this->typeOf($link);
+
+            // ChainConfig::addLink() keys named links by name and unnamed links by a
+            // separate auto-increment counter, so an unnamed link's key does not
+            // necessarily equal its position here — checking is_string() (rather than
+            // comparing to the position) is the only reliable way to tell "this is a
+            // real name" from "this is an auto-assigned index", named or not.
+            $rawName = $names[$index] ?? null;
+            $name = is_string($rawName) ? $rawName : $type;
+
+            $graph->addNode(new GraphNode(
+                $id,
+                $name,
+                $type,
+                $isBranching ? GraphNode::KIND_SPLIT : GraphNode::KIND_OPERATION,
+            ));
+
+            if ($isBranching) {
+                foreach ($link->getChainProcessors() as $branch => $subProcessor) {
+                    $this->addNodes($subProcessor, "$id.$branch", $graph);
+                }
+            }
+        }
+    }
+
+    private function addEdges(ChainProcessorInterface $processor, string $prefix, ?string $previous, ChainGraph $graph): void
+    {
+        foreach ($processor->getChainLinks() as $index => $link) {
+            $id = '' === $prefix ? (string) $index : "$prefix.$index";
+
+            if (null !== $previous) {
+                $graph->addEdge($previous, $id);
+            }
+            $previous = $id;
+
+            if ($link instanceof SubChainsAwareOperationInterface) {
+                // Each branch starts from this node; the main line ($previous) stays
+                // on it so the next top-level link follows from here.
+                foreach ($link->getChainProcessors() as $branch => $subProcessor) {
+                    $this->addEdges($subProcessor, "$id.$branch", $id, $graph);
+                }
+            }
+        }
+    }
+
+    private function typeOf(object $operation): string
+    {
+        $parts = explode('\\', $operation::class);
+
+        return end($parts);
+    }
+}