<?php

declare(strict_types=1);

namespace Drupal\Tests\flowdrop_stategraph\Kernel;

use Drupal\Tests\flowdrop\Kernel\WorkflowExecutionTestBase;
use Drupal\flowdrop_job\Constants\JobStatus;
use Drupal\flowdrop_orchestration\Service\Orchestrator\OrchestratorHelperInterface;
use Drupal\flowdrop_pipeline\Constants\PipelineStatus;
use Drupal\flowdrop_stategraph\Service\StateGraphOrchestrator;
use Drupal\flowdrop_workflow\WorkflowDefinitionInterface;

/**
 * Proves launched & resumed pipelines run under their declared orchestrator.
 *
 * The stategraph-executor-parity plan exists because pre-built pipelines
 * (deferred launch, interrupt resume) were hardwired to the synchronous
 * job loop, silently dropping checkpointing, reducers, and conditional-edge
 * routing for stategraph-declared workflows. These tests pin the fix
 * end-to-end across the real container — launcher → resolver →
 * StateGraphOrchestrator::executePipeline — which no unit test of a single
 * hop can protect.
 *
 * The observable StateGraph signature asserted here is the execution-context
 * stamp (orchestrator_type / stategraph_config) plus state_checkpoint
 * entities on the pipeline's thread: the sync loop produces neither.
 *
 * @group flowdrop_stategraph
 */
class StateGraphExecutorParityTest extends WorkflowExecutionTestBase {

  /**
   * Plan item 11: a launched stategraph workflow runs under StateGraph.
   *
   * Launch via the public WorkflowLauncherInterface with `wait`, then
   * assert the run demonstrably executed under StateGraphOrchestrator —
   * not the sync loop the launcher used to hardwire.
   */
  public function testWaitLaunchRunsUnderStateGraph(): void {
    $workflow = $this->createEntityCheckpointWorkflow('sg_parity_launch', [
      $this->loggerNode('log_first', 'first'),
      $this->loggerNode('log_second', 'second'),
    ], [
      ['id' => 'e1', 'source' => 'log_first', 'target' => 'log_second'],
    ]);

    $result = $this->launchAndWait($workflow);

    // `wait` ran the pipeline inline; the handle carries the final status.
    $this->assertFalse($result->queued);
    $this->assertSame(PipelineStatus::COMPLETED, $result->status);

    $pipeline = $this->loadPipeline($result->pipelineId);
    $this->assertSame(PipelineStatus::COMPLETED, $pipeline->getStatus());

    // The stategraph stamp: written by executePipeline(), never by the
    // sync loop.
    $context = $pipeline->getExecutionContext();
    $this->assertSame(
      OrchestratorHelperInterface::STATEGRAPH,
      $context[StateGraphOrchestrator::CONTEXT_ORCHESTRATOR_TYPE] ?? NULL,
    );
    $config = $context[StateGraphOrchestrator::CONTEXT_STATEGRAPH_CONFIG] ?? NULL;
    $this->assertIsArray($config);
    $this->assertSame('entity', $config['checkpointerType']);
    // No session on the launch → the isolated per-pipeline thread default.
    $expectedThread = 'pipeline_' . $pipeline->id();
    $this->assertSame($expectedThread, $config['threadId']);

    // Entity checkpoints on the pipeline's thread: the StateGraph
    // execution signature.
    $this->assertNotEmpty(
      $this->checkpointsForThread($expectedThread),
      'StateGraph wrote checkpoints for the launched pipeline.',
    );

    // Both jobs actually executed.
    $this->assertCount(2, $pipeline->getJobsByStatus(JobStatus::COMPLETED));
  }

  /**
   * Plan item 12: HITL resume goes through StateGraph, not the sync loop.
   *
   * A confirmation node pauses the run; resolving its interrupt fires
   * InterruptResolvedSubscriber, which must route the resume through
   * PipelineExecutorResolver to StateGraphOrchestrator::executePipeline.
   * The resume is proven to run under StateGraph by the checkpoint trail
   * advancing on the same thread after resolution.
   */
  public function testInterruptResumeRunsUnderStateGraph(): void {
    $workflow = $this->createEntityCheckpointWorkflow('sg_parity_resume', [
      $this->loggerNode('log_before', 'before'),
      $this->confirmationNode('confirm'),
      $this->loggerNode('log_after', 'after'),
    ], [
      ['id' => 'e1', 'source' => 'log_before', 'target' => 'confirm'],
      ['id' => 'e2', 'source' => 'confirm', 'target' => 'log_after'],
    ]);

    $result = $this->launchAndWait($workflow);

    // The confirmation node pauses the pipeline on a HITL interrupt.
    $this->assertSame(PipelineStatus::PAUSED, $result->status);

    $pipeline = $this->loadPipeline($result->pipelineId);
    $thread = 'pipeline_' . $pipeline->id();
    $checkpointsBeforeResume = count($this->checkpointsForThread($thread));
    $this->assertGreaterThan(
      0,
      $checkpointsBeforeResume,
      'The pre-pause run executed under StateGraph.',
    );

    $interrupts = $this->container->get('entity_type.manager')
      ->getStorage('flowdrop_interrupt')
      ->loadByProperties(['pipeline_id' => (string) $pipeline->id()]);
    $this->assertCount(1, $interrupts);
    $interrupt = reset($interrupts);

    // Resolving dispatches InterruptResolvedEvent; the subscriber resumes
    // the pipeline inline in this process.
    $this->container->get('flowdrop_interrupt.manager')
      ->resolveInterrupt($interrupt->uuid(), TRUE);

    $pipeline = $this->loadPipeline($result->pipelineId);
    $this->assertSame(PipelineStatus::COMPLETED, $pipeline->getStatus());

    // The resume advanced the same thread's checkpoint trail — i.e. it ran
    // under StateGraph with the stamped config, not the plain sync loop
    // (which writes no checkpoints).
    $this->assertGreaterThan(
      $checkpointsBeforeResume,
      count($this->checkpointsForThread($thread)),
      'The resume wrote further checkpoints on the stamped thread.',
    );

    // All three jobs completed across pause + resume.
    $this->assertCount(3, $pipeline->getJobsByStatus(JobStatus::COMPLETED));
  }

  /**
   * Creates a stategraph-declared workflow with the entity checkpointer.
   *
   * Cross-request continuity and observable checkpoints need the entity
   * checkpointer; `memory` would not survive into asserts.
   *
   * @param string $id
   *   The workflow entity id.
   * @param array<int, array<string, mixed>> $nodes
   *   The workflow nodes.
   * @param array<int, array<string, mixed>> $edges
   *   The workflow edges.
   *
   * @return \Drupal\flowdrop_workflow\WorkflowDefinitionInterface
   *   The saved workflow.
   */
  protected function createEntityCheckpointWorkflow(string $id, array $nodes, array $edges): WorkflowDefinitionInterface {
    return $this->createStateGraphWorkflow($id, $nodes, $edges, [
      'stategraph' => [
        'checkpointer_type' => 'entity',
        'create_checkpoints' => TRUE,
        'max_iterations' => 20,
      ],
    ]);
  }

  /**
   * Loads the checkpoint entities for a thread.
   *
   * @param string $threadId
   *   The thread id.
   *
   * @return array<int|string, \Drupal\Core\Entity\EntityInterface>
   *   The checkpoint entities.
   */
  protected function checkpointsForThread(string $threadId): array {
    $storage = $this->container->get('entity_type.manager')->getStorage('state_checkpoint');
    $storage->resetCache();
    return $storage->loadByProperties(['thread_id' => $threadId]);
  }

}
