<?php

declare(strict_types=1);

namespace Drupal\Tests\flowdrop_stategraph\Kernel;

use Drupal\Tests\flowdrop\Kernel\WorkflowExecutionTestBase;
use Drupal\flowdrop_interrupt\Entity\FlowDropInterruptInterface;
use Drupal\flowdrop_interrupt\Service\InterruptManagerInterface;
use Drupal\flowdrop_job\Constants\JobStatus;
use Drupal\flowdrop_pipeline\Constants\PipelineStatus;

/**
 * Pins interrupt identity to a single execution across loop iterations.
 *
 * Reproduces the field report: a cyclic StateGraph workflow with a
 * confirmation node in the loop body only paused on the FIRST iteration.
 * `createLoopIterationJob()` copied the original job's metadata wholesale,
 * so the iteration-2 clone inherited `metadata.interrupt_id` — pointing at
 * an interrupt that was already resolved — and the runtime routed the clone
 * to resume() instead of process(). The clone completed in the same tick
 * with iteration 1's answer: every HITL gate after iteration 1 silently
 * auto-approved, and the trigger-gated consumer could be scheduled twice
 * for one upstream emission (data-edge clone + confirmation fan-out clone),
 * duplicating side effects.
 *
 * Same bug family as the iteration-scoping fixtures in
 * {@see StateGraphBranchSemanticsTest}: per-execution state leaking across
 * loop-iteration boundaries via clones. Here the leaked state is interrupt
 * identity rather than branch decisions.
 *
 * Deterministic two-iteration cycle, no LLM: ForEach drives the loop and
 * the confirmation node gates the body on its True branch.
 *
 * @group flowdrop_stategraph
 */
class StateGraphInterruptScopingTest extends WorkflowExecutionTestBase {

  /**
   * Every loop iteration through a confirmation node must pause anew.
   *
   * Mirrors the field topology: the loop driver feeds the body over a data
   * edge AND the confirmation over a data edge; the confirmation's True
   * branch triggers the body; the body loops back.
   *
   * Before the fix: resolving the iteration-1 interrupt completed the whole
   * run — the iteration-2 confirmation clone auto-approved with iteration
   * 1's resolution (one interrupt total, no second pause).
   */
  public function testEachLoopIterationCreatesItsOwnInterrupt(): void {
    $workflow = $this->createStateGraphWorkflow('sg_interrupt_scope', [
      $this->repeatNode('seed', 'item', 2),
      $this->forEachNode('fe'),
      $this->confirmationNode('confirm', 'Run this iteration?'),
      $this->loggerNode('body', 'loop body'),
    ], [
      $this->edge('seed', 'items', 'fe', 'items'),
      // The body mirrors the field report's consumer: a data input from the
      // cycle driver (this is what pre-creates the iteration-2 body clone —
      // the precondition for the double-schedule) plus a True trigger from
      // the confirmation.
      $this->edge('fe', 'current_item', 'body', 'message'),
      $this->edge('fe', 'current_item', 'confirm', 'message'),
      $this->edge('confirm', 'True', 'body', 'trigger'),
      $this->edge('body', 'message', 'fe', 'loop_back'),
    ], [
      // Iterator state must survive the pause/resume boundary — the memory
      // checkpointer resets it on every resume, replaying item 0 forever.
      'stategraph' => [
        'checkpointer_type' => 'entity',
        'create_checkpoints' => TRUE,
        'max_iterations' => 20,
      ],
    ]);

    // Iteration 1 reaches the confirmation: the run pauses on a HITL
    // interrupt owned by the iteration-1 confirmation job.
    $result = $this->launchAndWait($workflow);
    $this->assertSame(PipelineStatus::PAUSED, $result->status);

    $pipeline = $this->loadPipeline($result->pipelineId);
    $interrupts = $this->interruptsForPipeline($result->pipelineId);
    $this->assertCount(1, $interrupts, $this->jobTrail($pipeline));
    $first = reset($interrupts);
    $this->assertNull($first->getResolvedAt());

    // Approve iteration 1. The body runs, loops back, and iteration 2 must
    // reach the confirmation AS A NEW QUESTION: a second interrupt entity,
    // a second pause — not an instant completion with iteration 1's answer.
    $this->interruptManager()->resolveInterrupt((string) $first->uuid(), TRUE);

    $pipeline = $this->loadPipeline($result->pipelineId);
    $this->assertSame(PipelineStatus::PAUSED, $pipeline->getStatus(),
      'Iteration 2 must pause on its own confirmation, not auto-approve '
      . 'with iteration 1\'s resolution: ' . $this->jobTrail($pipeline));

    $interrupts = $this->interruptsForPipeline($result->pipelineId);
    $this->assertCount(2, $interrupts,
      'Each loop iteration must create its own interrupt entity.');
    $second = $this->pendingInterrupt($interrupts);
    $this->assertNotNull($second, 'The iteration-2 interrupt must be pending.');
    $this->assertNotSame($first->getJobId(), $second->getJobId(),
      'The iteration-2 interrupt must be owned by the iteration-2 clone, '
      . 'not the iteration-1 job.');

    // Approve iteration 2: the items are exhausted and the run completes.
    $this->interruptManager()->resolveInterrupt((string) $second->uuid(), TRUE);

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

    // The trigger-gated consumer executed exactly once per resolved
    // interrupt — no duplicate body clone from the confirmation's fan-out.
    $bodyRuns = 0;
    $confirmRuns = 0;
    $confirmInterruptIds = [];
    foreach ($pipeline->getJobsByStatus(JobStatus::COMPLETED) as $job) {
      if ($job->getNodeId() === 'body') {
        $bodyRuns++;
      }
      if ($job->getNodeId() === 'confirm') {
        $confirmRuns++;
        $confirmInterruptIds[] = $job->getOutputData()['interrupt_id'] ?? NULL;
      }
    }
    $this->assertSame(2, $bodyRuns,
      'The trigger-gated consumer must execute exactly once per resolved interrupt.');
    $this->assertSame(2, $confirmRuns);
    $this->assertCount(2, array_unique($confirmInterruptIds),
      'Each confirmation run must consume its OWN interrupt — a shared '
      . 'interrupt_id means the clone replayed iteration 1\'s resolution.');
  }

  /**
   * Loads all interrupts for a pipeline, keyed by entity id.
   *
   * @param string $pipelineId
   *   The pipeline id.
   *
   * @return \Drupal\flowdrop_interrupt\Entity\FlowDropInterruptInterface[]
   *   The interrupt entities.
   */
  protected function interruptsForPipeline(string $pipelineId): array {
    $storage = $this->container->get('entity_type.manager')
      ->getStorage('flowdrop_interrupt');
    $storage->resetCache();
    $interrupts = [];
    foreach ($storage->loadByProperties(['pipeline_id' => $pipelineId]) as $id => $interrupt) {
      $this->assertInstanceOf(FlowDropInterruptInterface::class, $interrupt);
      $interrupts[$id] = $interrupt;
    }
    return $interrupts;
  }

  /**
   * Returns the single pending interrupt from a set, or NULL.
   *
   * @param \Drupal\flowdrop_interrupt\Entity\FlowDropInterruptInterface[] $interrupts
   *   The interrupt entities.
   *
   * @return \Drupal\flowdrop_interrupt\Entity\FlowDropInterruptInterface|null
   *   The pending interrupt or NULL.
   */
  protected function pendingInterrupt(array $interrupts): ?FlowDropInterruptInterface {
    foreach ($interrupts as $interrupt) {
      if ($interrupt->getResolvedAt() === NULL) {
        return $interrupt;
      }
    }
    return NULL;
  }

  /**
   * Returns the interrupt manager.
   */
  protected function interruptManager(): InterruptManagerInterface {
    return $this->container->get('flowdrop_interrupt.manager');
  }

}
