<?php

declare(strict_types=1);

namespace Drupal\flowdrop_workflow_executor\Service;

use Drupal\Core\DestructableInterface;
use Drupal\Core\Entity\EntityTypeManagerInterface;
use Drupal\flowdrop_pipeline\Entity\FlowDropPipelineRuntimeInterface;
use Psr\Log\LoggerInterface;
use Symfony\Component\DependencyInjection\ContainerInterface;

/**
 * Runs scheduled pipelines after the HTTP response is flushed.
 *
 * Mirrors {@see \Drupal\flowdrop_session\Service\DeferredMessageProcessor}
 * but operates on pipelines rather than messages. Tagged
 * `needs_destruction`, so `destruct()` is only called if the service was
 * actually used during the request — no overhead on requests that never
 * launch a workflow.
 *
 * The pipeline executor resolver is resolved lazily via the container to
 * keep the launcher's wire-up uncoupled from orchestrator boot order. Each
 * pipeline executes under the orchestrator its workflow declared — a
 * stategraph-declared pipeline runs through StateGraphOrchestrator, not
 * the plain synchronous job loop.
 *
 * @internal
 */
class DeferredPipelineProcessor implements DestructableInterface {

  /**
   * Pipeline IDs scheduled for deferred execution.
   *
   * @var array<int|string>
   */
  protected array $pendingPipelineIds = [];

  public function __construct(
    protected readonly EntityTypeManagerInterface $entityTypeManager,
    protected readonly ContainerInterface $container,
    protected readonly LoggerInterface $logger,
  ) {}

  /**
   * Schedules a pipeline for synchronous execution after response flush.
   *
   * The pipeline must already exist with its jobs generated. The
   * orchestrator will pick it up from PENDING/RUNNING and execute its
   * jobs to completion (or to the next interrupt) in the same PHP
   * process, after `fastcgi_finish_request()`.
   */
  public function schedule(FlowDropPipelineRuntimeInterface $pipeline): void {
    $id = $pipeline->id();
    if ($id !== NULL) {
      $this->pendingPipelineIds[] = $id;
      $this->logger->info('Scheduled deferred execution for pipeline @id', [
        '@id' => $id,
      ]);
    }
  }

  /**
   * Whether any pipelines are queued for deferred execution this request.
   */
  public function hasPendingPipelines(): bool {
    return $this->pendingPipelineIds !== [];
  }

  /**
   * Executes a pipeline immediately, in this process, before returning.
   *
   * The synchronous counterpart of schedule() — used by the launcher's
   * `wait` option. Same execution and error semantics as the deferred
   * path: the executor is resolved per the pipeline's declared
   * orchestrator, and failures land on the pipeline's status rather than
   * propagating to the caller.
   */
  public function executeNow(FlowDropPipelineRuntimeInterface $pipeline): void {
    $this->executeLoaded($pipeline);
  }

  /**
   * {@inheritdoc}
   */
  public function destruct(): void {
    if ($this->pendingPipelineIds === []) {
      return;
    }

    // Flush the response before doing any heavy work, so callers don't
    // observe orchestration time on their HTTP round-trip.
    if (function_exists('fastcgi_finish_request')) {
      fastcgi_finish_request();
    }

    $this->logger->info('Running @count deferred pipeline(s)', [
      '@count' => count($this->pendingPipelineIds),
    ]);

    $storage = $this->entityTypeManager->getStorage('flowdrop_pipeline');

    foreach ($this->pendingPipelineIds as $pipelineId) {
      $this->runOne($storage, $pipelineId);
    }

    $this->pendingPipelineIds = [];
  }

  /**
   * Runs a single scheduled pipeline.
   *
   * @param \Drupal\Core\Entity\EntityStorageInterface $storage
   *   The pipeline storage.
   * @param int|string $pipelineId
   *   The pipeline id to execute.
   */
  protected function runOne($storage, int|string $pipelineId): void {
    $pipeline = $storage->load($pipelineId);
    if (!$pipeline instanceof FlowDropPipelineRuntimeInterface) {
      $this->logger->warning('Pipeline @id not found for deferred execution', [
        '@id' => $pipelineId,
      ]);
      return;
    }
    $this->executeLoaded($pipeline);
  }

  /**
   * Executes a loaded pipeline under its declared orchestrator.
   *
   * @param \Drupal\flowdrop_pipeline\Entity\FlowDropPipelineRuntimeInterface $pipeline
   *   The pipeline to execute.
   */
  protected function executeLoaded(FlowDropPipelineRuntimeInterface $pipeline): void {
    try {
      // Lazy-resolve to keep the launcher service graph independent of
      // the orchestrator's own dependencies. The resolver routes to the
      // executor matching the pipeline's declared orchestrator.
      /** @var \Drupal\flowdrop_runtime\Service\Orchestrator\PipelineExecutorResolver $resolver */
      $resolver = $this->container->get('flowdrop_runtime.pipeline_executor_resolver');
      $resolver->getExecutorForPipeline($pipeline)->executePipeline($pipeline);

      $this->logger->info('Execution completed for pipeline @id', [
        '@id' => $pipeline->id(),
      ]);
    }
    catch (\Throwable $e) {
      // Execution failures land on the pipeline's status via the
      // orchestrator; log here so operators see them in watchdog.
      $this->logger->error('Execution failed for pipeline @id: @message (@class)', [
        '@id' => $pipeline->id(),
        '@message' => $e->getMessage(),
        '@class' => get_class($e),
      ]);
    }
  }

}
