<?php

declare(strict_types=1);

namespace Drupal\flowdrop_runtime\Service\Orchestrator;

use Drupal\flowdrop_orchestration\Service\Orchestrator\OrchestratorHelperInterface;
use Drupal\flowdrop_pipeline\Entity\FlowDropPipelineRuntimeInterface;
use Psr\Log\LoggerInterface;

/**
 * Resolves the executor a pre-built pipeline should run under.
 *
 * A pipeline that already exists (deferred launch, interrupt resume) must
 * execute under the orchestrator its workflow declared — running a
 * stategraph-declared pipeline through the plain synchronous job loop
 * silently drops checkpointing, reducers, conditional-edge routing, and the
 * agent loop. This service is the single place that mapping lives.
 *
 * Resolution order:
 * 1. The namespaced orchestrator type stamped on the pipeline's execution
 *    context at creation time (`orchestrator_type`).
 * 2. The `orchestrator_settings` declared in the workflow's metadata.
 * 3. The synchronous pipeline orchestrator — today's behavior, kept as the
 *    fallback, but no longer silent: a stamped/declared type that cannot be
 *    honored is logged.
 *
 * @internal
 *   This class is not part of the public API and may change without notice.
 */
class PipelineExecutorResolver {

  /**
   * Execution-context key holding the namespaced orchestrator type.
   *
   * Kept in sync with StateGraphOrchestrator::CONTEXT_ORCHESTRATOR_TYPE;
   * defined here too so flowdrop_runtime does not depend on
   * flowdrop_stategraph.
   */
  public const CONTEXT_ORCHESTRATOR_TYPE = "orchestrator_type";

  public function __construct(
    protected readonly OrchestratorHelperInterface $orchestratorHelper,
    protected readonly PipelineExecutorInterface $synchronousPipelineExecutor,
    protected readonly LoggerInterface $logger,
  ) {}

  /**
   * Resolves the executor for a pipeline by its declared orchestrator.
   *
   * @param \Drupal\flowdrop_pipeline\Entity\FlowDropPipelineRuntimeInterface $pipeline
   *   The pipeline to execute or resume.
   *
   * @return \Drupal\flowdrop_runtime\Service\Orchestrator\PipelineExecutorInterface
   *   The executor matching the pipeline's declared orchestrator, or the
   *   synchronous pipeline orchestrator when no better match exists.
   */
  public function getExecutorForPipeline(
    FlowDropPipelineRuntimeInterface $pipeline,
  ): PipelineExecutorInterface {
    $type = $this->resolveDeclaredType($pipeline);

    // "default" defers the choice to the helper (site default, event-based)
    // — resolve it before classifying, so the log lines below report the
    // actual orchestrator rather than the literal "default".
    if ($type === 'default') {
      $type = $this->orchestratorHelper->resolveDefault();
    }

    if ($type === NULL
      || $type === OrchestratorHelperInterface::SYNCHRONOUS_PIPELINE
      // The legacy synchronous orchestrator shares the sync pipeline
      // loop's semantics (in-process, no checkpointing) — executing its
      // pre-built pipelines through the sync pipeline executor is
      // faithful, not a substitution worth logging.
      || $type === OrchestratorHelperInterface::SYNCHRONOUS) {
      return $this->synchronousPipelineExecutor;
    }

    $plugin = $this->orchestratorHelper->getInstance(
      $type,
      OrchestratorHelperInterface::SYNCHRONOUS_PIPELINE,
    );

    if ($plugin instanceof PipelineExecutorAwareInterface) {
      return $plugin->getPipelineExecutor();
    }

    if ($type === OrchestratorHelperInterface::ASYNCHRONOUS) {
      // Expected, not a substitution: the async orchestrator executes fresh
      // launches through its queue worker and has never had a pre-built
      // pipeline executor — resumes of async-declared pipelines have always
      // run through the sync loop.
      $this->logger->info(
        'Pipeline @id declares the asynchronous orchestrator; resuming through the synchronous pipeline orchestrator as usual.',
        ['@id' => (string) $pipeline->id()],
      );
      return $this->synchronousPipelineExecutor;
    }

    $this->logger->warning(
      'Pipeline @id declares orchestrator @type, which cannot execute pre-built pipelines; falling back to the synchronous pipeline orchestrator.',
      ['@id' => (string) $pipeline->id(), '@type' => $type],
    );

    return $this->synchronousPipelineExecutor;
  }

  /**
   * Finds the namespaced orchestrator type the pipeline should run under.
   *
   * @param \Drupal\flowdrop_pipeline\Entity\FlowDropPipelineRuntimeInterface $pipeline
   *   The pipeline.
   *
   * @return string|null
   *   The namespaced orchestrator type, or NULL when nothing is declared.
   */
  protected function resolveDeclaredType(
    FlowDropPipelineRuntimeInterface $pipeline,
  ): ?string {
    $context = $pipeline->getExecutionContext();
    $stamped = $context[self::CONTEXT_ORCHESTRATOR_TYPE] ?? NULL;
    if (is_string($stamped) && $stamped !== '') {
      return $stamped;
    }

    $workflow = $pipeline->getWorkflow();
    if ($workflow === NULL) {
      return NULL;
    }
    $settings = $workflow->getMetadata()['orchestrator_settings'] ?? [];
    $declared = is_array($settings) ? ($settings['type'] ?? NULL) : NULL;

    return is_string($declared) && $declared !== '' ? $declared : NULL;
  }

}
