<?php

declare(strict_types=1);

namespace Drupal\flowdrop_workflow\Service;

use Drupal\Component\Plugin\Exception\PluginException;
use Drupal\flowdrop\Constants\ReservedName;
use Drupal\flowdrop\DTO\ValidationError;
use Drupal\flowdrop\DTO\ValidationResult;
use Drupal\flowdrop\DTO\ValidationWarning;
use Drupal\flowdrop\DTO\JsApi\Edge as TransportEdge;
use Drupal\flowdrop\Plugin\FlowDropNodeProcessor\FlowDropNodeProcessorInterface;
use Drupal\flowdrop\Plugin\FlowDropNodeProcessor\TerminalNodeProcessorInterface;
use Drupal\flowdrop\Service\FlowDropNodeProcessorPluginManager;
use Drupal\flowdrop\Utility\EdgePortClassifier;
use Drupal\flowdrop\Utility\NodeAnchor;
use Drupal\flowdrop\Utility\PortExposure;
use Drupal\flowdrop_node_type\Service\NodeMetadataResolver;

/**
 * Default WorkflowValidator implementation.
 *
 * @internal
 */
final class WorkflowValidator implements WorkflowValidatorInterface {

  public function __construct(
    private readonly FlowDropNodeProcessorPluginManager $pluginManager,
    private readonly NodeMetadataResolver $metadataResolver,
  ) {}

  /**
   * {@inheritdoc}
   */
  public function validate(array $nodes, array $edges, array $inputPorts, array $outputPorts): ValidationResult {
    $errors = [];
    $warnings = [];

    $nodesById = $this->indexNodes($nodes);

    $this->checkStructure($nodes, $edges, $errors);
    $this->checkPluginExistence($nodes, $errors);
    $this->checkConfig($nodes, $errors, $warnings);
    $this->checkEdgeEndpointsExist($edges, $nodesById, $errors);
    $this->checkEdgeExposure($edges, $nodesById, $errors);

    // Plugin instances are memoised across both sides so each distinct plugin
    // is constructed at most once per validation, regardless of how many
    // ports it exposes. A workflow with no exposed ports constructs nothing.
    $instances = [];
    $this->checkExposure($inputPorts, 'input', $nodesById, $instances, $errors);
    $this->checkExposure($outputPorts, 'output', $nodesById, $instances, $errors);
    $this->checkTerminalReachability($edges, $nodesById, $warnings);

    return $errors === []
      ? ValidationResult::success($warnings)
      : ValidationResult::failure($errors, $warnings);
  }

  /**
   * R5: structural limits — node/edge counts and node id presence.
   *
   * @param array<int, array<string, mixed>> $nodes
   *   The workflow nodes.
   * @param array<int, array<string, mixed>> $edges
   *   The workflow edges.
   * @param array<int, \Drupal\flowdrop\DTO\ValidationError> $errors
   *   Accumulator for validation errors, by reference.
   */
  private function checkStructure(array $nodes, array $edges, array &$errors): void {
    if (count($nodes) > self::MAX_NODES) {
      $errors[] = new ValidationError(
        'workflow',
        sprintf('Workflow exceeds the maximum node count (%d).', self::MAX_NODES),
        self::CODE_TOO_MANY_NODES,
      );
    }
    if (count($edges) > self::MAX_EDGES) {
      $errors[] = new ValidationError(
        'workflow',
        sprintf('Workflow exceeds the maximum edge count (%d).', self::MAX_EDGES),
        self::CODE_TOO_MANY_EDGES,
      );
    }
    foreach ($nodes as $index => $node) {
      if ($this->nodeId($node) === '') {
        $errors[] = new ValidationError(
          'node.' . $index,
          sprintf('Node at position %d has no identifier.', $index),
          self::CODE_NODE_MISSING_ID,
        );
      }
    }
  }

  /**
   * R1: every node's declared executor plugin must resolve.
   *
   * Uses hasDefinition() rather than constructing the plugin — existence is a
   * cheap definition lookup. Nodes with no declared executor plugin are left
   * to node-type validation (a later phase); only a non-empty id that fails to
   * resolve is an error here.
   *
   * @param array<int, array<string, mixed>> $nodes
   *   The workflow nodes.
   * @param array<int, \Drupal\flowdrop\DTO\ValidationError> $errors
   *   Accumulator for validation errors, by reference.
   */
  private function checkPluginExistence(array $nodes, array &$errors): void {
    foreach ($nodes as $node) {
      $pluginId = $this->executorPlugin($node);
      if ($pluginId === '') {
        continue;
      }
      if (!$this->pluginManager->hasDefinition($pluginId)) {
        $nodeId = $this->nodeId($node);
        $errors[] = new ValidationError(
          'node.' . ($nodeId !== '' ? $nodeId : '?'),
          sprintf('Node %s references node processor %s, which is not available. Install the module that provides it.', $nodeId !== '' ? $nodeId : '(unidentified)', $pluginId),
          self::CODE_PLUGIN_MISSING,
        );
      }
    }
  }

  /**
   * R6: a node's user config is valid against the node type's config schema.
   *
   * The config schema is derived from the live node type + plugin (the single
   * source of truth), keyed by the node's `data.metadata.node_type_id`
   * anchor — the stored metadata cache is never trusted. Required keys must be
   * present and
   * known keys must satisfy their constraints (both block). Keys absent from
   * the schema are warnings only: the runtime ignores them, so they cannot
   * affect execution.
   *
   * @param array<int, array<string, mixed>> $nodes
   *   The workflow nodes.
   * @param array<int, \Drupal\flowdrop\DTO\ValidationError> $errors
   *   Accumulator for blocking errors, by reference.
   * @param array<int, \Drupal\flowdrop\DTO\ValidationWarning> $warnings
   *   Accumulator for non-blocking warnings, by reference.
   */
  private function checkConfig(array $nodes, array &$errors, array &$warnings): void {
    // Memoise the derived config schema per node type: many nodes can share a
    // type, and deriving the schema instantiates the plugin.
    $schemaCache = [];

    foreach ($nodes as $node) {
      $nodeTypeId = NodeAnchor::read($node);
      if ($nodeTypeId === '') {
        // No anchor to resolve a schema from; identity issues are R1/R5.
        continue;
      }

      $nodeId = $this->nodeId($node);
      $locatorNode = $nodeId !== '' ? $nodeId : '?';
      $config = $node['data']['config'] ?? [];

      if (!is_array($config)) {
        $errors[] = new ValidationError(
          'node.' . $locatorNode . '.config',
          sprintf('Node %s has a non-object config.', $nodeId !== '' ? $nodeId : '(unidentified)'),
          self::CODE_CONFIG_INVALID,
        );
        continue;
      }

      if (!array_key_exists($nodeTypeId, $schemaCache)) {
        $metadata = $this->metadataResolver->buildForNodeTypeId($nodeTypeId);
        $schemaCache[$nodeTypeId] = $metadata?->toArray()['configSchema'] ?? NULL;
      }
      $configSchema = $schemaCache[$nodeTypeId];
      if (!is_array($configSchema)) {
        // Node type or plugin not resolvable; R1 reports a missing plugin.
        continue;
      }

      $properties = is_array($configSchema['properties'] ?? NULL) ? $configSchema['properties'] : [];
      $required = is_array($configSchema['required'] ?? NULL) ? $configSchema['required'] : [];
      $locatorBase = 'node.' . $locatorNode . '.config';

      // Required config keys must be present. The derived schema only marks a
      // key required when it cannot also be supplied by an input edge, so this
      // never flags params that arrive at runtime.
      foreach ($required as $reqName) {
        if (!array_key_exists((string) $reqName, $config)) {
          $errors[] = new ValidationError(
            $locatorBase . '.' . $reqName,
            sprintf('Required config "%s" is missing on node %s.', $reqName, $nodeId !== '' ? $nodeId : '(unidentified)'),
            self::CODE_CONFIG_REQUIRED,
          );
        }
      }

      // Per-key validation.
      foreach ($config as $key => $value) {
        $key = (string) $key;
        if (!isset($properties[$key])) {
          $warnings[] = new ValidationWarning(
            $locatorBase . '.' . $key,
            sprintf('Config key "%s" is not declared by node type "%s" and will be ignored at runtime.', $key, $nodeTypeId),
          );
          continue;
        }
        if (is_array($properties[$key])) {
          $this->validateConfigValue($key, $value, $properties[$key], $locatorBase, $errors);
        }
      }
    }
  }

  /**
   * R8: every edge endpoint references a node that exists.
   *
   * The editor cannot produce a dangling edge — deleting a node removes its
   * connected edges atomically, and the swap flow drops edges whose ports do
   * not carry over. Orphans therefore only enter through hand-edited JSON,
   * imports or migrations. This rejects them at save/import time so the
   * corruption is reported rather than persisted; the read path repairs the
   * same condition by pruning (see {@see WorkflowDTO::fromArray}).
   *
   * An empty source/target is treated as missing: such an edge references no
   * node and is equally unusable.
   *
   * @param array<int, array<string, mixed>> $edges
   *   The workflow edges.
   * @param array<string, array<string, mixed>> $nodesById
   *   Nodes indexed by identifier.
   * @param array<int, \Drupal\flowdrop\DTO\ValidationError> $errors
   *   Accumulator for blocking errors, by reference.
   */
  private function checkEdgeEndpointsExist(array $edges, array $nodesById, array &$errors): void {
    foreach ($edges as $index => $edge) {
      $transportEdge = TransportEdge::fromArray($edge);
      $source = $transportEdge->getSource();
      if (!isset($nodesById[$source])) {
        $errors[] = new ValidationError(
          'edge.' . $index,
          sprintf('Edge references source node "%s", which does not exist. Remove the edge or restore the node.', $source),
          self::CODE_EDGE_SOURCE_MISSING,
        );
      }

      $target = $transportEdge->getTarget();
      if (!isset($nodesById[$target])) {
        $errors[] = new ValidationError(
          'edge.' . $index,
          sprintf('Edge references target node "%s", which does not exist. Remove the edge or restore the node.', $target),
          self::CODE_EDGE_TARGET_MISSING,
        );
      }
    }
  }

  /**
   * R7: every edge connects exposed ports only.
   *
   * In v2 exposure is semantic: a not-exposed port is not wireable. The editor
   * enforces this by withholding the handle, but a hand-edited graph or stale
   * imported JSON can still carry an edge to/from a port the author left
   * not-exposed (e.g. an `error` output that ships not-exposed). This closes
   * that gap by rejecting such edges at save/import time.
   *
   * Effective exposure is read through {@see PortExposure}: the destination
   * node's `data.config.ports` override, else the port's metadata
   * `exposedByDefault`. Ports not declared by the node type's metadata
   * (dynamic, user-defined ports) are out of scope — they are inherently
   * wireable and carry no exposure default to check against.
   *
   * @param array<int, array<string, mixed>> $edges
   *   The workflow edges.
   * @param array<string, array<string, mixed>> $nodesById
   *   Nodes indexed by identifier.
   * @param array<int, \Drupal\flowdrop\DTO\ValidationError> $errors
   *   Accumulator for blocking errors, by reference.
   */
  private function checkEdgeExposure(array $edges, array $nodesById, array &$errors): void {
    // Memoise each node type's metadata exposure defaults: nodeTypeId =>
    // ['input' => [portId => bool], 'output' => [portId => bool]]. NULL marks
    // a type whose metadata could not be resolved (R1 reports the cause).
    $defaults = [];

    foreach ($edges as $index => $edge) {
      $locator = 'edge.' . $index;

      $transportEdge = TransportEdge::fromArray($edge);
      $targetHandle = $transportEdge->getTargetHandle();
      $this->checkEndpointExposed(
        $transportEdge->getTarget(),
        EdgePortClassifier::portName($targetHandle, ReservedName::PORT_INPUT),
        ReservedName::PORT_INPUT,
        $nodesById,
        $defaults,
        $locator,
        self::CODE_EDGE_TARGET_NOT_EXPOSED,
        'lands on',
        $errors,
      );

      $sourceHandle = $transportEdge->getSourceHandle();
      $this->checkEndpointExposed(
        $transportEdge->getSource(),
        EdgePortClassifier::portName($sourceHandle, ReservedName::PORT_OUTPUT),
        ReservedName::PORT_OUTPUT,
        $nodesById,
        $defaults,
        $locator,
        self::CODE_EDGE_SOURCE_NOT_EXPOSED,
        'emits from',
        $errors,
      );
    }
  }

  /**
   * Flags one edge endpoint when it addresses a not-exposed declared port.
   *
   * @param string $nodeId
   *   The endpoint node id.
   * @param string $port
   *   The port name parsed from the handle (empty when unparseable → skip).
   * @param string $direction
   *   ReservedName::PORT_INPUT or ::PORT_OUTPUT.
   * @param array<string, array<string, mixed>> $nodesById
   *   Nodes indexed by identifier.
   * @param array<string, array{input: array<string, bool>, output: array<string, bool>}|null> $defaults
   *   Memoised per-node-type exposure defaults, by reference.
   * @param string $locator
   *   The error locator for this edge.
   * @param string $code
   *   The validation error code to raise.
   * @param string $verb
   *   Human phrasing ("lands on" / "emits from") for the message.
   * @param array<int, \Drupal\flowdrop\DTO\ValidationError> $errors
   *   Accumulator for blocking errors, by reference.
   */
  private function checkEndpointExposed(string $nodeId, string $port, string $direction, array $nodesById, array &$defaults, string $locator, string $code, string $verb, array &$errors): void {
    // No identifiable node or port → nothing to enforce (structural concerns
    // are R5; an unparseable handle carries no port to check).
    if ($nodeId === '' || $port === '' || !isset($nodesById[$nodeId])) {
      return;
    }

    $portDefaults = $this->portExposureDefaults($nodesById[$nodeId], $defaults);
    $key = $direction === ReservedName::PORT_OUTPUT ? 'output' : 'input';
    // A port absent from the metadata defaults is dynamic/user-defined: out of
    // scope (inherently wireable, no default to check).
    if ($portDefaults === NULL || !array_key_exists($port, $portDefaults[$key])) {
      return;
    }

    $config = $nodesById[$nodeId]['data']['config'] ?? [];
    $configPorts = is_array($config) && is_array($config[ReservedName::PORTS] ?? NULL) ? $config[ReservedName::PORTS] : [];
    if (!PortExposure::isExposed($configPorts, $direction, $port, $portDefaults[$key][$port])) {
      $errors[] = new ValidationError(
        $locator,
        sprintf('Edge %s the not-exposed %s port "%s" of node "%s". Expose the port or remove the edge.', $verb, $key, $port, $nodeId),
        $code,
      );
    }
  }

  /**
   * Builds (memoised) a node type's per-port exposure defaults from metadata.
   *
   * @param array<string, mixed> $node
   *   The node whose type's defaults are wanted.
   * @param array<string, array{input: array<string, bool>, output: array<string, bool>}|null> $defaults
   *   Memoised defaults keyed by node type id, by reference.
   *
   * @return array{input: array<string, bool>, output: array<string, bool>}|null
   *   The exposure-default maps, or NULL when the type can't be resolved.
   */
  private function portExposureDefaults(array $node, array &$defaults): ?array {
    $nodeTypeId = NodeAnchor::read($node);
    if ($nodeTypeId === '') {
      return NULL;
    }
    if (array_key_exists($nodeTypeId, $defaults)) {
      return $defaults[$nodeTypeId];
    }

    $metadata = $this->metadataResolver->buildForNodeTypeId($nodeTypeId);
    if ($metadata === NULL) {
      return $defaults[$nodeTypeId] = NULL;
    }

    $map = ['input' => [], 'output' => []];
    foreach ($metadata->inputs as $port) {
      $map['input'][$port->id] = $port->exposedByDefault;
    }
    foreach ($metadata->outputs as $port) {
      $map['output'][$port->id] = $port->exposedByDefault;
    }
    return $defaults[$nodeTypeId] = $map;
  }

  /**
   * Validates a single config value against its JSON-schema property.
   *
   * Conservative on purpose: only flags unambiguous violations (wrong type,
   * value outside enum/bounds/length, pattern mismatch) so a save is never
   * blocked by a borderline case. A NULL value is treated as unset.
   *
   * @param string $key
   *   The config key.
   * @param mixed $value
   *   The config value.
   * @param array<string, mixed> $schema
   *   The property schema for this key.
   * @param string $locatorBase
   *   The locator prefix (e.g. "node.x.config").
   * @param array<int, \Drupal\flowdrop\DTO\ValidationError> $errors
   *   Accumulator for blocking errors, by reference.
   */
  private function validateConfigValue(string $key, mixed $value, array $schema, string $locatorBase, array &$errors): void {
    if ($value === NULL) {
      return;
    }
    $locator = $locatorBase . '.' . $key;

    $expectedType = $schema['type'] ?? 'mixed';
    if (is_string($expectedType) && !$this->isValidType($value, $expectedType)) {
      $errors[] = new ValidationError(
        $locator,
        sprintf('Config "%s" must be of type "%s".', $key, $expectedType),
        self::CODE_CONFIG_INVALID,
      );
      return;
    }

    if (isset($schema['enum']) && is_array($schema['enum']) && !in_array($value, $schema['enum'], TRUE)) {
      $errors[] = new ValidationError(
        $locator,
        sprintf('Config "%s" has a value outside its allowed set.', $key),
        self::CODE_CONFIG_INVALID,
      );
      return;
    }

    if (is_int($value) || is_float($value)) {
      $num = (float) $value;
      if (isset($schema['minimum']) && is_numeric($schema['minimum']) && $num < (float) $schema['minimum']) {
        $errors[] = new ValidationError($locator, sprintf('Config "%s" is below the minimum of %s.', $key, $schema['minimum']), self::CODE_CONFIG_INVALID);
        return;
      }
      if (isset($schema['maximum']) && is_numeric($schema['maximum']) && $num > (float) $schema['maximum']) {
        $errors[] = new ValidationError($locator, sprintf('Config "%s" exceeds the maximum of %s.', $key, $schema['maximum']), self::CODE_CONFIG_INVALID);
        return;
      }
    }

    if (is_string($value)) {
      $len = mb_strlen($value);
      if (isset($schema['minLength']) && is_numeric($schema['minLength']) && $len < (int) $schema['minLength']) {
        $errors[] = new ValidationError($locator, sprintf('Config "%s" is shorter than the minimum length of %d.', $key, (int) $schema['minLength']), self::CODE_CONFIG_INVALID);
        return;
      }
      if (isset($schema['maxLength']) && is_numeric($schema['maxLength']) && $len > (int) $schema['maxLength']) {
        $errors[] = new ValidationError($locator, sprintf('Config "%s" exceeds the maximum length of %d.', $key, (int) $schema['maxLength']), self::CODE_CONFIG_INVALID);
        return;
      }
      if (isset($schema['pattern']) && is_string($schema['pattern']) && $schema['pattern'] !== '') {
        $delimited = '#' . str_replace('#', '\\#', $schema['pattern']) . '#';
        // preg_match returns 0 for an explicit non-match; FALSE means the
        // schema's own pattern is invalid, which is not the author's fault.
        if (@preg_match($delimited, $value) === 0) {
          $errors[] = new ValidationError($locator, sprintf('Config "%s" does not match the required pattern.', $key), self::CODE_CONFIG_INVALID);
        }
      }
    }
  }

  /**
   * Checks a value against a JSON-schema scalar type (lenient).
   *
   * @param mixed $value
   *   The value.
   * @param string $expectedType
   *   The JSON schema type.
   *
   * @return bool
   *   TRUE if the value is acceptable for the type.
   */
  private function isValidType(mixed $value, string $expectedType): bool {
    return match ($expectedType) {
      'string' => is_string($value),
      'integer', 'int' => is_int($value),
      'number' => is_int($value) || is_float($value),
      'boolean', 'bool' => is_bool($value),
      'array' => is_array($value),
      'object' => is_array($value) || is_object($value),
      'null' => $value === NULL,
      default => TRUE,
    };
  }

  /**
   * R4: exposure-map integrity for one side.
   *
   * @param list<array{name?: string, node_id?: string, port?: string}> $entries
   *   The exposure entries for this side.
   * @param string $side
   *   The exposure side ('input' or 'output').
   * @param array<string, array<string, mixed>> $nodesById
   *   Nodes indexed by identifier.
   * @param array<string, \Drupal\flowdrop\Plugin\FlowDropNodeProcessor\FlowDropNodeProcessorInterface|null> $instances
   *   Memoised plugin instances, by reference. NULL marks a plugin that could
   *   not be constructed (already reported by R1).
   * @param array<int, \Drupal\flowdrop\DTO\ValidationError> $errors
   *   Accumulator for validation errors, by reference.
   */
  private function checkExposure(array $entries, string $side, array $nodesById, array &$instances, array &$errors): void {
    $seenNames = [];

    foreach ($entries as $index => $entry) {
      $locator = sprintf('schema.%s.%d', $side, $index);
      $name = (string) ($entry['name'] ?? '');
      $nodeId = (string) ($entry['node_id'] ?? '');
      $port = (string) ($entry['port'] ?? '');

      if (!preg_match('/^[a-z0-9_-]+$/', $name)) {
        $errors[] = new ValidationError(
          $locator,
          sprintf('Exposed name "%s" must match [a-z0-9_-]+.', $name),
          self::CODE_NAME_FORMAT,
        );
      }

      if (isset($seenNames[$name])) {
        $errors[] = new ValidationError(
          $locator,
          sprintf('Exposed name "%s" is duplicated.', $name),
          self::CODE_NAME_DUPLICATE,
        );
      }
      $seenNames[$name] = TRUE;

      if (!isset($nodesById[$nodeId])) {
        $errors[] = new ValidationError(
          $locator,
          sprintf('Node "%s" referenced by exposure "%s" no longer exists.', $nodeId, $name),
          self::CODE_NODE_MISSING,
        );
        continue;
      }

      $plugin = $this->resolvePlugin($nodesById[$nodeId], $instances);
      if (!$plugin instanceof FlowDropNodeProcessorInterface) {
        // No (resolvable) plugin to check the port against. A missing plugin
        // is already reported once by R1; don't double-report per port.
        continue;
      }

      $schema = $side === 'input' ? $plugin->getParameterSchema() : $plugin->getOutputSchema();
      $properties = $schema['properties'] ?? [];
      if (!isset($properties[$port])) {
        $errors[] = new ValidationError(
          $locator,
          sprintf('Port "%s" is not declared by plugin "%s" (exposure "%s").', $port, $this->executorPlugin($nodesById[$nodeId]), $name),
          self::CODE_PORT_MISSING,
        );
      }
    }
  }

  /**
   * Warns about nodes wired downstream of a terminal node.
   *
   * A terminal node (TerminalNodeProcessorInterface, e.g. Stop) ends the run
   * when reached, so any node wired after it can never execute. Non-blocking:
   * the graph is still valid — the downstream nodes are simply dead — so this
   * is a warning, not an error.
   *
   * Terminal-ness is read from the plugin definition's class via
   * is_subclass_of (no instantiation), keeping the validator's laziness
   * contract: a plugin is constructed only for exposed-port checks, never here.
   * Only edge SOURCES are inspected — a node with nothing wired after it can
   * carry no dead downstream — so an edgeless workflow looks nothing up.
   *
   * @param array<int, array<string, mixed>> $edges
   *   The workflow edges.
   * @param array<string, array<string, mixed>> $nodesById
   *   Nodes indexed by identifier.
   * @param array<int, \Drupal\flowdrop\DTO\ValidationWarning> $warnings
   *   Accumulator for non-blocking warnings, by reference.
   */
  private function checkTerminalReachability(array $edges, array $nodesById, array &$warnings): void {
    $isTerminal = [];
    foreach ($edges as $index => $edge) {
      $transportEdge = TransportEdge::fromArray($edge);
      $source = $transportEdge->getSource();
      // A dangling source is R8's concern, not ours.
      if (!isset($nodesById[$source])) {
        continue;
      }
      $isTerminal[$source] ??= $this->isTerminalNode($nodesById[$source]);
      if ($isTerminal[$source]) {
        $warnings[] = new ValidationWarning(
          'edge.' . $index,
          sprintf('Node "%s" is wired after terminal node "%s"; it can never execute because the run stops there.', $transportEdge->getTarget(), $source),
        );
      }
    }
  }

  /**
   * Whether a node's processor is a terminal node, without constructing it.
   *
   * @param array<string, mixed> $node
   *   The node.
   *
   * @return bool
   *   TRUE if the node's executor plugin implements
   *   TerminalNodeProcessorInterface.
   */
  private function isTerminalNode(array $node): bool {
    $pluginId = $this->executorPlugin($node);
    if ($pluginId === '') {
      return FALSE;
    }
    $definition = $this->pluginManager->getDefinition($pluginId, FALSE);
    if (!is_array($definition) || !isset($definition['class'])) {
      return FALSE;
    }
    $class = $definition['class'];
    return is_string($class) && is_subclass_of($class, TerminalNodeProcessorInterface::class);
  }

  /**
   * Returns the memoised plugin instance for a node, or NULL.
   *
   * @param array<string, mixed> $node
   *   The node.
   * @param array<string, \Drupal\flowdrop\Plugin\FlowDropNodeProcessor\FlowDropNodeProcessorInterface|null> $instances
   *   Memoised plugin instances, by reference.
   */
  private function resolvePlugin(array $node, array &$instances): ?FlowDropNodeProcessorInterface {
    $pluginId = $this->executorPlugin($node);
    if ($pluginId === '') {
      return NULL;
    }
    if (array_key_exists($pluginId, $instances)) {
      return $instances[$pluginId];
    }

    $instance = NULL;
    try {
      $candidate = $this->pluginManager->createInstance($pluginId);
      if ($candidate instanceof FlowDropNodeProcessorInterface) {
        $instance = $candidate;
      }
    }
    catch (PluginException) {
      // Leave $instance NULL; R1 reports the missing plugin.
    }
    return $instances[$pluginId] = $instance;
  }

  /**
   * Indexes nodes by their identifier.
   *
   * @param array<int, array<string, mixed>> $nodes
   *   The nodes to index.
   *
   * @return array<string, array<string, mixed>>
   *   Nodes keyed by identifier; nodes without an id are skipped.
   */
  private function indexNodes(array $nodes): array {
    $indexed = [];
    foreach ($nodes as $node) {
      $id = $this->nodeId($node);
      if ($id !== '') {
        $indexed[$id] = $node;
      }
    }
    return $indexed;
  }

  /**
   * Extracts a node's identifier.
   *
   * @param array<string, mixed> $node
   *   The node.
   */
  private function nodeId(array $node): string {
    return (string) ($node['id'] ?? '');
  }

  /**
   * Resolves a node's executor plugin id.
   *
   * Resolves the authoritative plugin declared by the node type entity (keyed
   * by the persisted `metadata.node_type_id` anchor). Returns an empty string
   * when the node type can no longer be resolved.
   *
   * @param array<string, mixed> $node
   *   The node.
   */
  private function executorPlugin(array $node): string {
    $nodeTypeId = NodeAnchor::read($node);
    if ($nodeTypeId !== '') {
      $resolved = $this->metadataResolver->resolveExecutorPlugin($nodeTypeId);
      if ($resolved !== NULL) {
        return $resolved;
      }
    }
    return '';
  }

}
