<?php

declare(strict_types=1);

namespace Drupal\Tests\flowdrop_runtime\Unit\Service\Orchestrator;

use Drupal\Core\DependencyInjection\ContainerBuilder;
use Drupal\Core\Logger\LoggerChannelFactoryInterface;
use Drupal\Tests\UnitTestCase;
use Drupal\flowdrop_job\Constants\JobStatus;
use Drupal\flowdrop_job\FlowDropJobInterface;
use Drupal\flowdrop_orchestration\DTO\Orchestrator\OrchestrationRequest;
use Drupal\flowdrop_orchestration\DTO\Orchestrator\OrchestrationResponse;
use Drupal\flowdrop_pipeline\Entity\FlowDropPipelineRuntimeInterface;
use Drupal\flowdrop_pipeline\Service\JobGenerationService;
use Drupal\flowdrop_runtime\Service\Orchestrator\AbstractOrchestratorBase;
use Drupal\flowdrop_runtime\Service\Runtime\NodeRuntimeServiceInterface;
use Drupal\flowdrop_runtime\Service\Storage\JobStorageInterface;
use Drupal\flowdrop_runtime\Service\Storage\PipelineStorageRegistryInterface;
use Psr\Log\LoggerInterface;
use Symfony\Component\EventDispatcher\EventDispatcherInterface;

/**
 * Pins the "many sources → one input port" resolution contract.
 *
 * When several edges target the SAME input port, the engine keeps a SINGLE
 * value — never an aggregated array — and that value MUST come from the latest
 * executor (the source with the highest execution_order). This last-executor-
 * wins rule is the intended contract: it flows naturally into StateGraph loop
 * order (each re-execution completes later, so it wins) and keeps the
 * error-recovery alternate path easy to reason about without array semantics.
 * Any resolution path that does NOT pick the latest executor is a bug.
 *
 * The resolution lives in
 * {@see AbstractOrchestratorBase::buildRuntimeInputData()} and is shared by
 * every pipeline orchestrator — the synchronous orchestrators directly, and
 * the queue-backed asynchronous orchestrator transitively (it executes
 * through the synchronous pipeline orchestrator). That single shared
 * implementation, with its execution-order sort, first-writer-wins, and
 * collision warning, is pinned here.
 *
 * A second path,
 * {@see \Drupal\flowdrop_stategraph\Service\StateGraphOrchestrator::buildStateGraphRuntimeInputData()},
 * applies the same last-executor-wins rule (it now stamps execution_order on
 * completion); it requires a StateGraph state object and is covered separately.
 *
 * These tests also act as the tripwire for the proposed opt-in
 * `'multiple' => true` fan-in port (upstream request #1): turning overwrite
 * into ordered-array collection must update them on purpose, never silently.
 *
 * @group flowdrop_runtime
 */
class MultiSourcePortResolutionTest extends UnitTestCase {

  /**
   * {@inheritdoc}
   */
  protected function setUp(): void {
    parent::setUp();

    // AbstractOrchestratorBase gets its logger injected directly (see the
    // helper below); a minimal container is still set for any mock that
    // resolves services through \Drupal during the test.
    $logger = $this->createMock(LoggerInterface::class);
    $loggerFactory = $this->createMock(LoggerChannelFactoryInterface::class);
    $loggerFactory->method('get')->willReturn($logger);
    $container = new ContainerBuilder();
    $container->set('logger.factory', $loggerFactory);
    \Drupal::setContainer($container);
  }

  /**
   * A lone source populates its target port with the source's output value.
   */
  public function testSingleSourcePopulatesTargetPort(): void {
    $source = $this->makeJob('node_a', ['result' => 'alpha'], 1, 10);
    $target = $this->makeTargetJob('node_z', 100, [
      $this->edge('node_a', 'node_a-output-result', 'node_z-input-data'),
    ]);
    $pipeline = $this->makePipeline([$source]);

    $worker = $this->makeWorker();
    $inputs = $worker->call($target, $pipeline);

    $this->assertSame(['data' => 'alpha'], $inputs);
  }

  /**
   * Two sources → one port: a single value survives, from the latest executor.
   *
   * The surviving value is pinned exactly so a future fan-in change (which
   * would yield ['first', 'second']) breaks this test on purpose.
   */
  public function testTwoSourcesIntoOnePortKeepSingleValueByExecutionOrder(): void {
    // node_early ran first (order 1), node_late ran second (order 2). The
    // latest executor wins, so the port resolves to node_late → 'second'.
    $early = $this->makeJob('node_early', ['out' => 'first'], 1, 10);
    $late = $this->makeJob('node_late', ['out' => 'second'], 2, 11);
    $target = $this->makeTargetJob('node_z', 100, [
      $this->edge('node_early', 'node_early-output-out', 'node_z-input-data'),
      $this->edge('node_late', 'node_late-output-out', 'node_z-input-data'),
    ]);
    $pipeline = $this->makePipeline([$early, $late]);

    $inputs = $this->makeWorker()->call($target, $pipeline);

    // Exactly ONE value on the port — not an array of both.
    $this->assertArrayHasKey('data', $inputs);
    $this->assertIsNotArray($inputs['data']);
    // Latest executor wins: execution-order sort + first-writer-wins.
    $this->assertSame(['data' => self::WINNER_BY_EXEC_ORDER], $inputs);
  }

  /**
   * No execution order recorded: winner falls back to source-node-ID ordering.
   */
  public function testTwoSourcesFallBackToNodeIdOrderWhenNoExecutionOrder(): void {
    // Neither source carries an execution_order (passed as NULL).
    $a = $this->makeJob('node_a', ['out' => 'from_a'], NULL, 10);
    $b = $this->makeJob('node_b', ['out' => 'from_b'], NULL, 11);
    $target = $this->makeTargetJob('node_z', 100, [
      $this->edge('node_a', 'node_a-output-out', 'node_z-input-data'),
      $this->edge('node_b', 'node_b-output-out', 'node_z-input-data'),
    ]);
    $pipeline = $this->makePipeline([$a, $b]);

    $inputs = $this->makeWorker()->call($target, $pipeline);

    $this->assertSame(['data' => self::WINNER_BY_NODE_ID], $inputs);
  }

  /**
   * Characterizes fan-in resolution across every completed-source combination.
   *
   * Four sources (A, B, C, D) wire into ONE input port — the wiring is fixed;
   * only the runtime status of each source varies. This is the executable form
   * of the fan-in truth table: 2^4 = 16 rows, one per subset of COMPLETED
   * sources.
   *
   * Modelling notes that make the table correct:
   *  - A source is either a CANDIDATE (COMPLETED) or absent. FAILED and PENDING
   *    are indistinguishable here: both are simply not in the pipeline's
   *    COMPLETED set, so neither contributes a value. Each row therefore lists
   *    only the completed sources.
   *  - Completion order is A→B→C→D, so execution_order follows node rank
   *    (A=1 … D=4). Among the completed candidates the LATEST executor wins
   *    (highest execution_order), so the winner is always the highest-ranked
   *    completed source — never an array of several.
   *  - Zero completed sources → the port is never populated and the target
   *    receives no input ([]).
   *
   * This pins the VALUE resolution only. WHICH row you land on at runtime is a
   * separate, timing-governed question owned by
   * {@see \Drupal\flowdrop_pipeline\Service\JobGenerationService::areJobDependenciesMet()}
   * (OR-within-port readiness fires the target once one source completes).
   * A future opt-in array fan-in (upstream request #1) must rewrite these
   * expectations on purpose, never silently.
   *
   * @param array<int, string> $completed
   *   The source letters that are COMPLETED, in completion order.
   * @param string|null $winner
   *   The letter of the source whose value should win, or NULL when no source
   *   has completed and the port stays unpopulated.
   *
   * @dataProvider provideFanInCombinations
   */
  public function testFanInResolvesToLatestCompletedSource(array $completed, ?string $winner): void {
    $ranks = ['a' => 1, 'b' => 2, 'c' => 3, 'd' => 4];

    $jobs = [];
    foreach ($completed as $letter) {
      $rank = $ranks[$letter];
      $jobs[] = $this->makeJob("node_$letter", ['out' => "val_$letter"], $rank, 10 + $rank);
    }

    // Fixed wiring: all four sources always target the same input port.
    $target = $this->makeTargetJob('node_z', 100, [
      $this->edge('node_a', 'node_a-output-out', 'node_z-input-data'),
      $this->edge('node_b', 'node_b-output-out', 'node_z-input-data'),
      $this->edge('node_c', 'node_c-output-out', 'node_z-input-data'),
      $this->edge('node_d', 'node_d-output-out', 'node_z-input-data'),
    ]);
    $pipeline = $this->makePipeline($jobs);

    $inputs = $this->makeWorker()->call($target, $pipeline);

    if ($winner === NULL) {
      // No completed source → port unpopulated, target receives no input.
      $this->assertSame([], $inputs);
      return;
    }

    // Exactly ONE value on the port, from the latest completed executor — never
    // an aggregated array of every completed source.
    $this->assertArrayHasKey('data', $inputs);
    $this->assertIsNotArray($inputs['data']);
    $this->assertSame(['data' => "val_$winner"], $inputs);
  }

  /**
   * The fan-in truth table: every subset of completed sources → the winner.
   *
   * @return array<string, array{0: array<int, string>, 1: string|null}>
   *   Keyed by the human-readable row label; each case is [completed, winner].
   */
  public static function provideFanInCombinations(): array {
    return [
      'none completed → no input'   => [[], NULL],
      'A only → A'                  => [['a'], 'a'],
      'B only → B'                  => [['b'], 'b'],
      'C only → C'                  => [['c'], 'c'],
      'D only → D'                  => [['d'], 'd'],
      'A,B → B (latest)'            => [['a', 'b'], 'b'],
      'A,C → C (latest)'            => [['a', 'c'], 'c'],
      'A,D → D (latest)'            => [['a', 'd'], 'd'],
      'B,C → C (latest)'            => [['b', 'c'], 'c'],
      'B,D → D (latest)'            => [['b', 'd'], 'd'],
      'C,D → D (latest)'            => [['c', 'd'], 'd'],
      'A,B,C → C (latest)'          => [['a', 'b', 'c'], 'c'],
      'A,B,D → D (latest)'          => [['a', 'b', 'd'], 'd'],
      'A,C,D → D (latest)'          => [['a', 'c', 'd'], 'd'],
      'B,C,D → D (latest)'          => [['b', 'c', 'd'], 'd'],
      'A,B,C,D → D (latest)'        => [['a', 'b', 'c', 'd'], 'd'],
    ];
  }

  /**
   * Trigger edges carry no data and must not populate any port.
   */
  public function testTriggerEdgesAreIgnoredForDataFlow(): void {
    $source = $this->makeJob('node_a', ['result' => 'alpha'], 1, 10);
    $target = $this->makeTargetJob('node_z', 100, [
      $this->edge('node_a', 'node_a-output-result', 'node_z-input-data', TRUE),
    ]);
    $pipeline = $this->makePipeline([$source]);

    $inputs = $this->makeWorker()->call($target, $pipeline);

    $this->assertSame([], $inputs);
  }

  /**
   * Distinct target ports each receive their own source value (no clobbering).
   */
  public function testDistinctPortsAreIndependent(): void {
    $a = $this->makeJob('node_a', ['out' => 'alpha'], 1, 10);
    $b = $this->makeJob('node_b', ['out' => 'beta'], 2, 11);
    $target = $this->makeTargetJob('node_z', 100, [
      $this->edge('node_a', 'node_a-output-out', 'node_z-input-left'),
      $this->edge('node_b', 'node_b-output-out', 'node_z-input-right'),
    ]);
    $pipeline = $this->makePipeline([$a, $b]);

    $inputs = $this->makeWorker()->call($target, $pipeline);

    // assertEquals (not assertSame): only key/value pairs matter here, the
    // population order is an incidental side effect of edge processing order.
    $this->assertEquals(['left' => 'alpha', 'right' => 'beta'], $inputs);
  }

  /**
   * A collision (2nd source into a port) emits a warning and keeps one value.
   */
  public function testCollisionLogsWarning(): void {
    $logger = $this->createMock(LoggerInterface::class);
    $logger->expects($this->once())
      ->method('warning')
      ->with($this->stringContains('Multiple sources target'));

    $early = $this->makeJob('node_early', ['out' => 'first'], 1, 10);
    $late = $this->makeJob('node_late', ['out' => 'second'], 2, 11);
    $target = $this->makeTargetJob('node_z', 100, [
      $this->edge('node_early', 'node_early-output-out', 'node_z-input-data'),
      $this->edge('node_late', 'node_late-output-out', 'node_z-input-data'),
    ]);
    $pipeline = $this->makePipeline([$early, $late]);

    $inputs = $this->makeWorker($logger)->call($target, $pipeline);

    $this->assertArrayHasKey('data', $inputs);
    $this->assertIsNotArray($inputs['data']);
    $this->assertSame(['data' => self::WINNER_BY_EXEC_ORDER], $inputs);
  }

  /**
   * Pinned winner of an execution-order tiebreak: the LATEST-executed source.
   *
   * Source node_late (execution order 2) beats node_early (order 1) →
   * 'second'. This is the last-executor-wins contract; any other winner is a
   * bug.
   */
  private const WINNER_BY_EXEC_ORDER = 'second';

  /**
   * Pinned winner when no execution order is present: a deterministic fallback.
   *
   * With no order to compare, edges sort by source node ID (ascending) and the
   * first one written wins, so 'node_a' → 'from_a'. Arbitrary but stable.
   */
  private const WINNER_BY_NODE_ID = 'from_a';

  // --- helpers --------------------------------------------------------------

  /**
   * Builds an orchestrator exposing the protected buildRuntimeInputData().
   *
   * The method only consumes $this->logger and $this->extractPortName().
   * AbstractOrchestratorBase declares no abstract methods, so a concrete
   * anonymous subclass built through the real constructor (with the injected
   * logger and mocks for the remaining deps) exercises the shared
   * implementation every live pipeline orchestrator uses.
   *
   * @param \Psr\Log\LoggerInterface|null $logger
   *   An optional logger; a no-op mock is used when omitted.
   *
   * @return object
   *   An AbstractOrchestratorBase subclass with a public call() shim.
   */
  private function makeWorker(?LoggerInterface $logger = NULL): object {
    $logger ??= $this->createMock(LoggerInterface::class);

    return new class(
      $this->createMock(PipelineStorageRegistryInterface::class),
      $this->createMock(JobStorageInterface::class),
      $this->createMock(NodeRuntimeServiceInterface::class),
      $this->createMock(JobGenerationService::class),
      $this->createMock(EventDispatcherInterface::class),
      $logger,
    ) extends AbstractOrchestratorBase {

      /**
       * Public shim over the protected method under test.
       *
       * @return array<string, mixed>
       *   The resolved runtime inputs.
       */
      public function call(FlowDropJobInterface $job, FlowDropPipelineRuntimeInterface $pipeline): array {
        return $this->buildRuntimeInputData($job, $pipeline);
      }

      /**
       * {@inheritdoc}
       */
      public function getType(): string {
        return 'test';
      }

      /**
       * {@inheritdoc}
       */
      public function orchestrate(OrchestrationRequest $request): OrchestrationResponse {
        throw new \BadMethodCallException('Not used in this test.');
      }

      /**
       * {@inheritdoc}
       */
      public function getCapabilities(): array {
        return [];
      }

    };
  }

  /**
   * Builds a completed upstream job mock.
   *
   * @param string $nodeId
   *   The node ID the job ran for.
   * @param array<string, mixed> $output
   *   The job's output data (port => value).
   * @param int|null $executionOrder
   *   The execution_order metadata, or NULL to omit it.
   * @param int $id
   *   The job entity ID.
   */
  private function makeJob(string $nodeId, array $output, ?int $executionOrder, int $id): FlowDropJobInterface {
    $metadata = [];
    if ($executionOrder !== NULL) {
      $metadata['execution_order'] = $executionOrder;
    }

    $job = $this->createMock(FlowDropJobInterface::class);
    $job->method('getNodeId')->willReturn($nodeId);
    $job->method('getOutputData')->willReturn($output);
    $job->method('getMetadata')->willReturn($metadata);
    $job->method('id')->willReturn($id);

    return $job;
  }

  /**
   * Builds the target job whose incoming edges are being resolved.
   *
   * @param string $nodeId
   *   The target node ID.
   * @param int $id
   *   The target job entity ID.
   * @param array<int, array<string, mixed>> $incomingEdges
   *   The incoming edge definitions.
   */
  private function makeTargetJob(string $nodeId, int $id, array $incomingEdges): FlowDropJobInterface {
    $job = $this->createMock(FlowDropJobInterface::class);
    $job->method('getNodeId')->willReturn($nodeId);
    $job->method('id')->willReturn($id);
    $job->method('getMetadata')->willReturn(['incoming_edges' => $incomingEdges]);
    $job->method('getOutputData')->willReturn([]);

    return $job;
  }

  /**
   * Builds a pipeline mock returning the given completed jobs.
   *
   * @param array<int, FlowDropJobInterface> $completedJobs
   *   The completed upstream jobs.
   */
  private function makePipeline(array $completedJobs): FlowDropPipelineRuntimeInterface {
    $pipeline = $this->createMock(FlowDropPipelineRuntimeInterface::class);
    $pipeline->method('getJobsByStatus')->willReturnCallback(
      fn(string $status): array => $status === JobStatus::COMPLETED ? $completedJobs : []
    );
    $pipeline->method('id')->willReturn(1);

    return $pipeline;
  }

  /**
   * Builds an incoming-edge definition.
   */
  private function edge(string $source, string $sourceHandle, string $targetHandle, bool $isTrigger = FALSE): array {
    return [
      'source' => $source,
      'source_handle' => $sourceHandle,
      'target_handle' => $targetHandle,
      'is_trigger' => $isTrigger,
    ];
  }

}
