<?php

declare(strict_types=1);

namespace Drupal\flowdrop_runtime\Plugin\QueueWorker;

use Drupal\flowdrop_pipeline\Entity\FlowDropPipelineInterface;
use Drupal\Core\Queue\QueueWorkerBase;
use Drupal\Core\Queue\RequeueException;
use Drupal\Core\Queue\SuspendQueueException;
use Drupal\flowdrop_job\FlowDropJobInterface;
use Drupal\flowdrop_runtime\Service\Runtime\NodeRuntimeService;
use Drupal\flowdrop_runtime\Service\Orchestrator\AsynchronousOrchestrator;
use Drupal\flowdrop_runtime\DTO\Runtime\NodeExecutionContext;
use Drupal\Core\Entity\EntityTypeManagerInterface;
use Drupal\Core\Logger\LoggerChannelFactoryInterface;
use Drupal\Core\Plugin\ContainerFactoryPluginInterface;
use Symfony\Component\DependencyInjection\ContainerInterface;

/**
 * Queue worker for job execution via runtime.
 *
 * @QueueWorker(
 * id = "flowdrop_runtime_job",
 * title = @Translation("FlowDrop Runtime Job"),
 * cron = {"time" = 60}
 * )
 */
class JobExecutionWorker extends QueueWorkerBase implements ContainerFactoryPluginInterface {

  /**
   * Constructs a JobExecutionWorker.
   *
   * @param array $configuration
   *   A configuration array containing information about the plugin instance.
   * @param string $plugin_id
   *   The plugin_id for the plugin instance.
   * @param mixed $plugin_definition
   *   The plugin implementation definition.
   * @param \Drupal\flowdrop_runtime\Service\Runtime\NodeRuntimeService $nodeRuntime
   *   The node runtime service.
   * @param \Drupal\flowdrop_runtime\Service\Orchestrator\AsynchronousOrchestrator $asynchronousOrchestrator
   *   The asynchronous orchestrator service.
   * @param \Drupal\Core\Entity\EntityTypeManagerInterface $entityTypeManager
   *   The entity type manager.
   * @param \Drupal\Core\Logger\LoggerChannelFactoryInterface $loggerFactory
   *   The logger factory.
   */
  public function __construct(
    array $configuration,
    string $plugin_id,
    $plugin_definition,
    protected NodeRuntimeService $nodeRuntime,
    protected AsynchronousOrchestrator $asynchronousOrchestrator,
    protected EntityTypeManagerInterface $entityTypeManager,
    protected LoggerChannelFactoryInterface $loggerFactory,
  ) {
    parent::__construct($configuration, $plugin_id, $plugin_definition);
  }

  /**
   * {@inheritdoc}
   */
  public static function create(
    ContainerInterface $container,
    array $configuration,
    $plugin_id,
    $plugin_definition,
  ): static {
    return new static(
      $configuration,
      $plugin_id,
      $plugin_definition,
      $container->get("flowdrop_runtime.node_runtime"),
      $container->get("flowdrop_runtime.asynchronous_orchestrator"),
      $container->get("entity_type.manager"),
      $container->get("logger.factory"),
    );
  }

  /**
   * {@inheritdoc}
   */
  public function processItem($data) {
    try {
      if (!isset($data["job_id"])) {
        throw new \InvalidArgumentException("Job ID is required");
      }

      $job_id = $data["job_id"];
      $action = $data["action"] ?? "execute";

      // Load the job.
      $job = $this->entityTypeManager->getStorage("flowdrop_job")->load($job_id);
      if (!$job instanceof FlowDropJobInterface) {
        throw new \RuntimeException("Job {$job_id} not found");
      }

      $this->loggerFactory->get("flowdrop_runtime")->info("Processing job @id with action @action", [
        "@id" => $job_id,
        "@action" => $action,
      ]);

      switch ($action) {
        case "execute":
          $this->executeJob($job);
          break;

        case "retry":
          $this->retryJob($job);
          break;

        default:
          throw new \InvalidArgumentException("Unknown action: {$action}");
      }

      $this->loggerFactory->get("flowdrop_runtime")->info("Successfully processed job @id", ["@id" => $job_id]);

    }
    catch (\Exception $e) {
      $this->loggerFactory->get("flowdrop_runtime")->error("Failed to process job execution: @message", [
        "@message" => $e->getMessage(),
      ]);

      // Requeue the item if it's a temporary failure.
      if ($this->isTemporaryFailure($e)) {
        throw new RequeueException($e->getMessage());
      }

      // Suspend the queue if it's a permanent failure.
      if ($this->isPermanentFailure($e)) {
        throw new SuspendQueueException($e->getMessage());
      }

      // For other exceptions, just log and continue.
      $this->loggerFactory->get("flowdrop_runtime")->error("Job execution failed: @message", [
        "@message" => $e->getMessage(),
      ]);
    }
  }

  /**
   * Execute a job.
   *
   * @param \Drupal\flowdrop_job\FlowDropJobInterface $job
   *   The job to execute.
   */
  protected function executeJob(FlowDropJobInterface $job): void {
    try {
      // Validate the job before execution.
      $this->validateJob($job);

      // Prepare execution context.
      $context = $this->prepareExecutionContext($job);

      // Get config from metadata (design-time parameters).
      // Config is stored in metadata with merged defaults and user values.
      $metadata = $job->getMetadata();
      $config = $metadata["config"] ?? [];

      // Build runtime inputs from upstream nodes.
      $inputs = $this->prepareInputs($job);

      // Update job's input_data with runtime inputs for tracking/debugging.
      // This captures what data actually flowed into the job at execution time.
      if (!empty($inputs)) {
        $job->setInputData($inputs);
        $job->save();
      }

      // Get node type from metadata.
      $nodeTypeId = $job->getMetadataValue("node_type_id", "default");
      if ($nodeTypeId === "default") {
        $nodeTypeId = $this->getNodeType($job);
      }

      // Execute the job using node runtime.
      $result = $this->nodeRuntime->executeNode(
        $job->id(),
        $job->getNodeId(),
        $nodeTypeId,
        $inputs,
        $config,
        $context
      );

      // Handle successful completion.
      $this->asynchronousOrchestrator->handleJobCompletion($job, $result->getOutput()->toArray());

    }
    catch (\Exception $e) {
      // Handle job failure.
      $this->asynchronousOrchestrator->handleJobFailure($job, $e->getMessage());
      throw $e;
    }
  }

  /**
   * Retry a failed job.
   *
   * @param \Drupal\flowdrop_job\FlowDropJobInterface $job
   *   The job to retry.
   */
  protected function retryJob(FlowDropJobInterface $job): void {
    try {
      // Execute the job with retry logic.
      $this->executeJob($job);

    }
    catch (\Exception $e) {
      // Handle job failure.
      $this->asynchronousOrchestrator->handleJobFailure($job, $e->getMessage());
      throw $e;
    }
  }

  /**
   * Validate job before execution.
   *
   * @param \Drupal\flowdrop_job\FlowDropJobInterface $job
   *   The job to validate.
   *
   * @throws \InvalidArgumentException
   *   If the job is invalid.
   */
  protected function validateJob(FlowDropJobInterface $job): void {
    // Check required fields.
    if (empty($job->getNodeId())) {
      throw new \InvalidArgumentException("Job must have a node ID");
    }

    // Check job status.
    if (!in_array($job->getStatus(), ["pending", "running"], TRUE)) {
      throw new \InvalidArgumentException("Job must be in pending or running status");
    }

    // Validate metadata contains required node information.
    // Node data and config are now stored in metadata (design-time config).
    $metadata = $job->getMetadata();
    if (empty($metadata) || !isset($metadata["node_type_id"])) {
      throw new \InvalidArgumentException("Job must have valid metadata with node_type_id");
    }
  }

  /**
   * Prepare execution context for a job.
   *
   * @param \Drupal\flowdrop_job\FlowDropJobInterface $job
   *   The job to prepare context for.
   *
   * @return \Drupal\flowdrop_runtime\DTO\Runtime\NodeExecutionContext
   *   The execution context.
   */
  protected function prepareExecutionContext(FlowDropJobInterface $job): NodeExecutionContext {
    // Get pipeline to access workflow and initial data.
    $pipeline = $job->getPipeline();
    $workflowId = "";
    $pipelineId = $job->id() ?? "";
    $initialData = [];
    $metadata = [
      "job_id" => $job->id(),
      "node_id" => $job->getNodeId(),
      "execution_time" => time(),
      "retry_count" => $job->getRetryCount(),
      "max_retries" => $job->getMaxRetries(),
    ];

    if ($pipeline !== NULL) {
      $pipelineId = $pipeline->id() ?? "";
      $initialData = $pipeline->getInputData();
      $workflow = $pipeline->getWorkflow();
      if ($workflow !== NULL) {
        $workflowId = $workflow->id() ?? "";
      }
    }

    return new NodeExecutionContext(
      workflowId: $workflowId,
      pipelineId: $pipelineId,
      initialData: $initialData,
      metadata: $metadata
    );
  }

  /**
   * Prepare inputs for a job.
   *
   * Builds runtime input data from upstream node outputs via incoming edges.
   * Context is passed separately via NodeExecutionContext to
   * NodeRuntimeService, which injects it into processors implementing
   * ExecutionContextAwareInterface.
   *
   * @param \Drupal\flowdrop_job\FlowDropJobInterface $job
   *   The job to prepare inputs for.
   *
   * @return array<string, mixed>
   *   The prepared inputs as array.
   */
  protected function prepareInputs(FlowDropJobInterface $job): array {
    $inputData = [];

    // Build runtime input data from incoming edges (upstream node outputs).
    $pipeline = $job->getPipeline();
    if ($pipeline !== NULL) {
      $inputData = $this->buildRuntimeInputData($job, $pipeline);
    }

    return $inputData;
  }

  /**
   * Prepare config for a job.
   *
   * @param \Drupal\flowdrop_job\FlowDropJobInterface $job
   *   The job to prepare config for.
   *
   * @return array<string, mixed>
   *   The prepared config as array.
   */
  protected function prepareConfig(FlowDropJobInterface $job): array {
    return [
      "timeout" => 300,
      "max_retries" => $job->getMaxRetries(),
      "memory_limit" => 128,
    ];
  }

  /**
   * Get the node type for a job.
   *
   * @param \Drupal\flowdrop_job\FlowDropJobInterface $job
   *   The job to get node type for.
   *
   * @return string
   *   The node type.
   */
  protected function getNodeType(FlowDropJobInterface $job): string {
    $metadata = $job->getMetadata();
    return $metadata["node_type_id"] ?? $metadata["executor_plugin_id"] ?? "unknown";
  }

  /**
   * Builds runtime input data for a job from incoming edges.
   *
   * @param \Drupal\flowdrop_job\FlowDropJobInterface $job
   *   The job to build input data for.
   * @param \Drupal\flowdrop_pipeline\Entity\FlowDropPipelineInterface $pipeline
   *   The pipeline this job belongs to.
   *
   * @return array<string, mixed>
   *   Runtime input data mapped by port names.
   */
  protected function buildRuntimeInputData(
    FlowDropJobInterface $job,
    FlowDropPipelineInterface $pipeline,
  ): array {
    $runtimeData = [];
    $logger = $this->loggerFactory->get("flowdrop_runtime");

    $metadata = $job->getMetadata();
    $incomingEdges = $metadata["incoming_edges"] ?? [];

    $completedJobs = $pipeline->getJobsByStatus("completed");
    $completedJobsMap = [];
    foreach ($completedJobs as $completedJob) {
      $completedJobsMap[$completedJob->getNodeId()] = $completedJob;
    }

    // Track which target ports have been set and by which sources.
    // This helps detect conflicts when multiple sources target the same port.
    $portSources = [];

    // Build execution order map from job metadata for sorting by "latest".
    $executionOrderMap = [];
    foreach ($completedJobs as $completedJob) {
      $jobMetadata = $completedJob->getMetadata();
      $order = $jobMetadata["execution_order"] ?? 0;
      if ($order > 0) {
        $executionOrderMap[$completedJob->getNodeId()] = $order;
      }
    }

    // Sort edges by execution order (latest first = higher counter),
    // then by node ID as fallback.
    // This ensures that when multiple sources target the same port,
    // the most recently executed job's value is used, which is more
    // intuitive than alphabetical ordering.
    // The execution counter increments with each job completion,
    // avoiding timestamp collisions.
    usort($incomingEdges, function (array $a, array $b) use ($executionOrderMap): int {
      $sourceA = $a["source"] ?? "";
      $sourceB = $b["source"] ?? "";

      $orderA = $executionOrderMap[$sourceA] ?? 0;
      $orderB = $executionOrderMap[$sourceB] ?? 0;

      // If both have execution order, sort by latest
      // (descending = higher counter first).
      if ($orderA > 0 && $orderB > 0) {
        return $orderB <=> $orderA;
      }

      // If only one has execution order, prioritize it.
      if ($orderA > 0) {
        return -1;
      }
      if ($orderB > 0) {
        return 1;
      }

      // Fallback to alphabetical by node ID if no execution order available.
      return strcmp($sourceA, $sourceB);
    });

    foreach ($incomingEdges as $edge) {
      // Skip trigger edges - they're only for execution control.
      if ($edge["is_trigger"] ?? FALSE) {
        continue;
      }

      $sourceNodeId = $edge["source"];
      $sourceHandle = $edge["source_handle"] ?? "";
      $targetHandle = $edge["target_handle"] ?? "";

      if (!isset($completedJobsMap[$sourceNodeId])) {
        $logger->debug("Source job @source not completed for data flow to job @job", [
          "@source" => $sourceNodeId,
          "@job" => $job->id(),
        ]);
        continue;
      }

      $sourceJob = $completedJobsMap[$sourceNodeId];
      $sourceOutput = $sourceJob->getOutputData();

      $sourcePortName = $this->extractPortName($sourceHandle, "output");
      $targetPortName = $this->extractPortName($targetHandle, "input");

      if ($sourcePortName !== NULL && $targetPortName !== NULL) {
        if (isset($sourceOutput[$sourcePortName])) {
          // Check for conflicts: multiple sources targeting the same port.
          if (isset($portSources[$targetPortName])) {
            $previousSource = $portSources[$targetPortName];
            $prevOrder = $executionOrderMap[$previousSource] ?? 0;
            $currOrder = $executionOrderMap[$sourceNodeId] ?? 0;

            $resolutionMethod = "latest execution order (execution counter)";
            if ($prevOrder === 0 && $currOrder === 0) {
              $resolutionMethod = "alphabetical order (no execution order available)";
            }
            elseif ($prevOrder === $currOrder && $prevOrder > 0) {
              $resolutionMethod = "alphabetical order (execution order equal)";
            }

            $logger->warning(
              "Multiple sources target the same input port '@target_port' on job '@job_id' (node: '@node_id'). " .
              "Previous source: '@prev_source' (order: @prev_order), current source: '@curr_source' (order: @curr_order). " .
              "Using value from '@curr_source' (resolved by @method).",
              [
                "@target_port" => $targetPortName,
                "@job_id" => $job->id(),
                "@node_id" => $job->getNodeId(),
                "@prev_source" => $previousSource,
                "@prev_order" => $prevOrder > 0 ? (string) $prevOrder : "N/A",
                "@curr_source" => $sourceNodeId,
                "@curr_order" => $currOrder > 0 ? (string) $currOrder : "N/A",
                "@method" => $resolutionMethod,
              ]
            );
          }
          $runtimeData[$targetPortName] = $sourceOutput[$sourcePortName];
          $portSources[$targetPortName] = $sourceNodeId;

          $logger->debug("Data flow: job @source[@source_port] -> job @target[@target_port]", [
            "@source" => $sourceJob->id(),
            "@source_port" => $sourcePortName,
            "@target" => $job->id(),
            "@target_port" => $targetPortName,
          ]);
        }
      }
    }

    return $runtimeData;
  }

  /**
   * Extracts port name from a handle.
   *
   * @param string $handle
   *   The handle string.
   * @param string $direction
   *   The direction ("output" or "input").
   *
   * @return string|null
   *   The port name, or NULL if parsing failed.
   */
  protected function extractPortName(string $handle, string $direction): ?string {
    if (empty($handle)) {
      return NULL;
    }

    $pattern = "/-" . $direction . "-/";
    $parts = preg_split($pattern, $handle, 2);

    if ($parts !== FALSE && count($parts) === 2) {
      return $parts[1];
    }

    return NULL;
  }

  /**
   * Check if the exception represents a temporary failure.
   *
   * @param \Exception $e
   *   The exception to check.
   *
   * @return bool
   *   TRUE if the failure is temporary.
   */
  protected function isTemporaryFailure(\Exception $e): bool {
    // Network errors, database connection issues, etc.
    $temporary_exceptions = [
      "PDOException",
      "DatabaseException",
      "ConnectionException",
      "TimeoutException",
    ];

    foreach ($temporary_exceptions as $exception_class) {
      if ($e instanceof $exception_class) {
        return TRUE;
      }
    }

    // Check for specific error messages that indicate temporary issues.
    $temporary_messages = [
      "connection",
      "timeout",
      "temporary",
      "retry",
      "busy",
      "locked",
      "rate limit",
      "quota exceeded",
    ];

    $message = strtolower($e->getMessage());
    foreach ($temporary_messages as $keyword) {
      if (strpos($message, $keyword) !== FALSE) {
        return TRUE;
      }
    }

    return FALSE;
  }

  /**
   * Check if the exception represents a permanent failure.
   *
   * @param \Exception $e
   *   The exception to check.
   *
   * @return bool
   *   TRUE if the failure is permanent.
   */
  protected function isPermanentFailure(\Exception $e): bool {
    // Invalid data, missing dependencies, etc.
    $permanent_exceptions = [
      "InvalidArgumentException",
      "RuntimeException",
      "LogicException",
    ];

    foreach ($permanent_exceptions as $exception_class) {
      if ($e instanceof $exception_class) {
        return TRUE;
      }
    }

    // Check for specific error messages that indicate permanent issues.
    $permanent_messages = [
      "not found",
      "invalid",
      "missing",
      "required",
      "not allowed",
      "authentication failed",
      "authorization failed",
      "permission denied",
    ];

    $message = strtolower($e->getMessage());
    foreach ($permanent_messages as $keyword) {
      if (strpos($message, $keyword) !== FALSE) {
        return TRUE;
      }
    }

    return FALSE;
  }

}
