<?php

declare(strict_types=1);

namespace Drupal\flowdrop_workflow_executor\Service;

use Drupal\Core\Queue\QueueFactory;
use Drupal\flowdrop\DTO\NodeMetadata\NodePort;
use Drupal\flowdrop_orchestration\Service\Orchestrator\OrchestratorHelperInterface;
use Drupal\flowdrop_pipeline\Constants\PipelineStatus;
use Drupal\flowdrop_pipeline\Entity\FlowDropPipelineInterface;
use Drupal\flowdrop_pipeline\Entity\FlowDropPipelineRuntimeInterface;
use Drupal\flowdrop_pipeline\Service\JobGenerationService;
use Drupal\flowdrop_runtime\Service\Storage\PipelineStorageRegistryInterface;
use Drupal\flowdrop_workflow\WorkflowDefinitionInterface;
use Drupal\flowdrop_workflow\Service\WorkflowSchemaResolverInterface;
use Drupal\flowdrop_workflow\Service\WorkflowValidatorInterface;
use Drupal\flowdrop_workflow_executor\DTO\LaunchOptions;
use Drupal\flowdrop_workflow_executor\DTO\LaunchResult;
use Drupal\flowdrop_workflow_executor\Exception\InvalidLaunchInputException;
use Drupal\flowdrop_workflow_executor\Exception\WorkflowLaunchException;
use Drupal\flowdrop_workflow_executor\Exception\WorkflowLaunchValidationException;
use Psr\Log\LoggerInterface;

/**
 * Default implementation of {@see WorkflowLauncherInterface}.
 *
 * The contract on the interface is the BC commitment. This class is free
 * to change.
 *
 * @internal
 */
final class WorkflowLauncher implements WorkflowLauncherInterface {

  /**
   * Queue used to dispatch async-orchestrator runs to a worker.
   */
  private const PIPELINE_EXECUTION_QUEUE = 'flowdrop_runtime_pipeline_execution';

  public function __construct(
    private readonly WorkflowSchemaResolverInterface $schemaResolver,
    private readonly JobGenerationService $jobGenerationService,
    private readonly OrchestratorHelperInterface $orchestratorHelper,
    private readonly PipelineStorageRegistryInterface $pipelineStorageRegistry,
    private readonly DeferredPipelineProcessor $deferredPipelineProcessor,
    private readonly QueueFactory $queueFactory,
    private readonly WorkflowValidatorInterface $validator,
    private readonly LoggerInterface $logger,
  ) {}

  /**
   * {@inheritdoc}
   */
  public function launch(
    WorkflowDefinitionInterface $workflow,
    array $inputs = [],
    ?LaunchOptions $options = NULL,
  ): LaunchResult {
    $options ??= new LaunchOptions();

    $this->validateDefinition($workflow);
    $this->validateInputs($workflow, $inputs);
    $initialData = $this->schemaResolver->resolveInputs($workflow, $inputs);

    [$orchestratorType, $bundle] = $this->resolveOrchestratorAndBundle($workflow);

    $pipeline = $this->createPipeline($workflow, $bundle, $initialData, $options->sessionId, $orchestratorType);

    try {
      $this->jobGenerationService->generateJobs($pipeline);
    }
    catch (\Throwable $e) {
      $this->logger->error('Job generation failed for launched pipeline @id: @message', [
        '@id' => $pipeline->id(),
        '@message' => $e->getMessage(),
      ]);
      throw new WorkflowLaunchException(
        'Failed to generate jobs for launched workflow.',
        0,
        $e,
      );
    }

    // Two dispatch paths. Async (queue worker) is durable: work survives this
    // process. Otherwise (sync/stategraph) the pipeline runs in this PHP
    // process — after response flush via DeferredPipelineProcessor, or inline
    // when the caller opted into `wait`. The choice of path follows the
    // workflow author's declared orchestrator; the public API does not let
    // the caller override it.
    $isAsync = $orchestratorType === OrchestratorHelperInterface::ASYNCHRONOUS;
    if ($options->wait && $isAsync) {
      // The queue is the author's declared durability choice; `wait` never
      // overrides orchestrators.
      $this->logger->notice('LaunchOptions::wait ignored for workflow @workflow: it declares the asynchronous orchestrator.', [
        '@workflow' => (string) $workflow->id(),
      ]);
    }
    $queued = $this->dispatch($pipeline, $isAsync, $options->wait);

    if (!$queued && $options->wait) {
      // The run executed inline; reload so the result carries the final
      // status. Execution failures landed on the pipeline, not as
      // exceptions — same contract as the deferred path.
      $fresh = $this->pipelineStorageRegistry->getStorage($bundle)->load($pipeline->id());
      if ($fresh instanceof FlowDropPipelineRuntimeInterface) {
        $pipeline = $fresh;
      }
    }

    $this->logger->info('Launched workflow @workflow as pipeline @pipeline (orchestrator: @orch, queued: @queued, waited: @waited)', [
      '@workflow' => (string) $workflow->id(),
      '@pipeline' => (string) $pipeline->id(),
      '@orch' => $orchestratorType,
      '@queued' => $queued ? 'true' : 'false',
      '@waited' => (!$queued && $options->wait) ? 'true' : 'false',
    ]);

    return new LaunchResult(
      pipelineId: (string) $pipeline->id(),
      status: $pipeline->getStatus(),
      queued: $queued,
    );
  }

  /**
   * Validates the workflow definition's structural integrity before launch.
   *
   * The save and import paths run the shared {@see WorkflowValidatorInterface}
   * choke-point, but a definition can reach storage without it — a workflow
   * created before a rule existed, written through the entity API directly, or
   * applied via config import. Re-running the validator here fails fast with a
   * clear error rather than letting an invalid definition execute. That matters
   * most for the exposure rules (R7): the runtime defends itself defensively —
   * a non-exposed input edge is ignored, a non-exposed output is stripped — so
   * an invalid edge would otherwise run to a silently wrong result instead of
   * an error the caller can act on.
   *
   * Validation is identical to the save path (all rules) and runs once per
   * launch, which is negligible against the pipeline it precedes.
   *
   * @throws \Drupal\flowdrop_workflow_executor\Exception\WorkflowLaunchValidationException
   *   When the definition fails validation — a 4xx-class condition the HTTP
   *   boundary maps to 422 with the per-error detail.
   */
  private function validateDefinition(WorkflowDefinitionInterface $workflow): void {
    $result = $this->validator->validate(
      $workflow->getNodes(),
      $workflow->getEdges(),
      $workflow->getInputPorts(),
      $workflow->getOutputPorts(),
    );
    if ($result->isValid()) {
      return;
    }

    $details = implode('; ', array_map(
      static fn ($error): string => $error->getCode() . ': ' . $error->getMessage(),
      $result->getErrors(),
    ));
    $this->logger->warning('Refused to launch invalid workflow @workflow: @details', [
      '@workflow' => (string) $workflow->id(),
      '@details' => $details,
    ]);
    throw new WorkflowLaunchValidationException(
      'Workflow ' . (string) $workflow->id()
      . ' failed validation and cannot be launched: ' . $details,
      $result,
    );
  }

  /**
   * Validates caller inputs against the workflow's declared input ports.
   *
   * Rejects unknown port names (which the schema resolver would otherwise
   * silently drop) and missing required ports. Type validation against the
   * declared schema is a follow-up — for now we trust port types are
   * compatible with the supplied values, and surface errors via downstream
   * execution.
   *
   * @throws \Drupal\flowdrop_workflow_executor\Exception\InvalidLaunchInputException
   */
  private function validateInputs(WorkflowDefinitionInterface $workflow, array $inputs): void {
    $declared = [];
    $required = [];
    foreach ($workflow->getInputPorts() as $port) {
      if (!is_array($port) || !isset($port['name'])) {
        continue;
      }
      $portDto = NodePort::fromArray($port);
      $declared[] = $portDto->name;
      if ($portDto->required) {
        $required[] = $portDto->name;
      }
    }

    $unknown = array_values(array_diff(array_keys($inputs), $declared));
    if ($unknown !== []) {
      throw new InvalidLaunchInputException(
        'Unknown input(s) for workflow ' . (string) $workflow->id() . ': '
        . implode(', ', $unknown),
      );
    }

    $missing = array_values(array_diff($required, array_keys($inputs)));
    if ($missing !== []) {
      throw new InvalidLaunchInputException(
        'Missing required input(s) for workflow ' . (string) $workflow->id() . ': '
        . implode(', ', $missing),
      );
    }
  }

  /**
   * Resolves the orchestrator type (and matching pipeline bundle).
   *
   * The workflow's declared `orchestrator_settings` (in its metadata)
   * wins; defaults to ASYNCHRONOUS for headless launches.
   *
   * @return array{0: string, 1: string}
   *   [orchestrator namespaced id, pipeline bundle].
   */
  private function resolveOrchestratorAndBundle(WorkflowDefinitionInterface $workflow): array {
    $metadata = $workflow->getMetadata();
    $orchestratorSettings = $metadata['orchestrator_settings'] ?? [];

    $orchestrator = $this->orchestratorHelper->getInstanceFromSettings(
      $orchestratorSettings,
      OrchestratorHelperInterface::ASYNCHRONOUS,
    );
    // The OrchestratorPluginInterface exposes a plugin id; namespaced form
    // is `<provider>:<id>`. We use the bundle resolver below rather than
    // string-matching the plugin id ourselves.
    $bundle = $this->pipelineStorageRegistry->getBundleForOrchestrator($orchestrator);

    // The orchestrator plugin manager already namespaces plugin ids as
    // `<provider>:<id>` during discovery, so getPluginId() is the
    // namespaced type. (Re-prepending the provider here used to double the
    // namespace — `flowdrop_runtime:flowdrop_runtime:asynchronous` — which
    // broke both the async queue-dispatch detection and the
    // orchestrator_type stamp the executor resolver routes by.)
    $orchestratorType = $orchestrator->getPluginId();

    return [$orchestratorType, $bundle];
  }

  /**
   * Creates and persists the pipeline entity in PENDING state.
   *
   * The resolved namespaced orchestrator type is stamped into the
   * pipeline's execution context so post-creation execution paths
   * (DeferredPipelineProcessor, interrupt resume) route the pipeline to
   * the executor its workflow declared.
   */
  private function createPipeline(
    WorkflowDefinitionInterface $workflow,
    string $bundle,
    array $initialData,
    ?string $sessionId,
    string $orchestratorType,
  ): FlowDropPipelineRuntimeInterface {
    $storage = $this->pipelineStorageRegistry->getStorage($bundle);

    $createValues = [
      'bundle' => $bundle,
      'label' => 'Launch ' . (string) $workflow->id() . ' ' . gmdate('Y-m-d H:i:s'),
      'workflow_id' => (string) $workflow->id(),
      'input_data' => $initialData,
      'status' => PipelineStatus::PENDING,
    ];
    if ($sessionId !== NULL && $sessionId !== '') {
      $createValues['session_id'] = $sessionId;
    }

    /** @var \Drupal\flowdrop_pipeline\Entity\FlowDropPipelineRuntimeInterface $pipeline */
    $pipeline = $storage->create($createValues);
    // The execution_context field is JSON-encoded by its setter; it cannot
    // be passed as a raw create value.
    $pipeline->setExecutionContext(['orchestrator_type' => $orchestratorType]);
    $storage->save($pipeline);

    return $pipeline;
  }

  /**
   * Dispatches a created+generated pipeline to its execution path.
   *
   * @return bool
   *   TRUE if execution is queued to a worker; FALSE if it runs in this
   *   process (post-response destruct, or inline when $wait is set).
   */
  private function dispatch(FlowDropPipelineRuntimeInterface $pipeline, bool $isAsync, bool $wait): bool {
    if ($isAsync) {
      // markAsStarted before enqueue so a polling client sees RUNNING
      // immediately rather than briefly PENDING. Mirrors the form's
      // resumePipeline pattern.
      $pipeline->markAsStarted();
      if ($pipeline instanceof FlowDropPipelineInterface) {
        $pipeline->save();
      }
      $this->queueFactory->get(self::PIPELINE_EXECUTION_QUEUE)->createItem([
        'pipeline_id' => $pipeline->id(),
        'action' => 'execute',
      ]);
      return TRUE;
    }

    if ($wait) {
      // Sync (or stategraph) with wait — run inline, in this request,
      // before launch() returns.
      $this->deferredPipelineProcessor->executeNow($pipeline);
      return FALSE;
    }

    // Sync (or stategraph) — run in this PHP process after response flush.
    $this->deferredPipelineProcessor->schedule($pipeline);
    return FALSE;
  }

}
