<?php

declare(strict_types=1);

namespace Drupal\Tests\flowdrop_interrupt\Unit\EventSubscriber;

use Drupal\Core\Entity\EntityStorageInterface;
use Drupal\Core\Entity\EntityTypeManagerInterface;
use Drupal\flowdrop_interrupt\DTO\ExecutionStamp;
use Drupal\flowdrop_interrupt\Entity\FlowDropInterruptInterface;
use Drupal\flowdrop_interrupt\Enum\InterruptType;
use Drupal\flowdrop_interrupt\EventSubscriber\BudgetPauseSignalSubscriber;
use Drupal\flowdrop_interrupt\Message\InwardSignalMessage;
use Drupal\flowdrop_interrupt\Service\InterruptManagerInterface;
use Drupal\flowdrop_interrupt\Service\InterruptMessageBusInterface;
use Drupal\flowdrop_pipeline\Constants\PausedReason;
use Drupal\flowdrop_pipeline\Constants\PipelineStatus;
use Drupal\flowdrop_pipeline\Entity\FlowDropPipelineRuntimeInterface;
use Drupal\flowdrop_runtime\Event\PipelineCompletedEvent;
use Drupal\Tests\UnitTestCase;
use Prophecy\Argument;
use Prophecy\PhpUnit\ProphecyTrait;
use Prophecy\Prophecy\ObjectProphecy;
use Psr\Log\LoggerInterface;

/**
 * Tests BudgetPauseSignalSubscriber's gating and signal creation.
 *
 * The subscriber must raise a system Pause signal ONLY for a paused run
 * carrying a stamped budget reason and no already-pending inward signal —
 * never for completed/failed runs, HITL pauses (no reason), or on top of
 * an operator's own signal.
 *
 * @coversDefaultClass \Drupal\flowdrop_interrupt\EventSubscriber\BudgetPauseSignalSubscriber
 * @group flowdrop_interrupt
 */
class BudgetPauseSignalSubscriberTest extends UnitTestCase {

  use ProphecyTrait;

  /**
   * Mocked interrupt manager.
   *
   * @var \Prophecy\Prophecy\ObjectProphecy
   */
  private ObjectProphecy $interruptManager;

  /**
   * Mocked message bus.
   *
   * @var \Prophecy\Prophecy\ObjectProphecy
   */
  private ObjectProphecy $messageBus;

  /**
   * Mocked entity type manager.
   *
   * @var \Prophecy\Prophecy\ObjectProphecy
   */
  private ObjectProphecy $entityTypeManager;

  /**
   * Subscriber under test.
   */
  private BudgetPauseSignalSubscriber $subscriber;

  /**
   * {@inheritdoc}
   */
  protected function setUp(): void {
    parent::setUp();
    $this->interruptManager = $this->prophesize(InterruptManagerInterface::class);
    $this->messageBus = $this->prophesize(InterruptMessageBusInterface::class);
    $this->entityTypeManager = $this->prophesize(EntityTypeManagerInterface::class);
    $this->subscriber = new BudgetPauseSignalSubscriber(
      $this->interruptManager->reveal(),
      $this->messageBus->reveal(),
      $this->entityTypeManager->reveal(),
      $this->prophesize(LoggerInterface::class)->reveal(),
    );
  }

  /**
   * A non-paused outcome never raises a signal.
   *
   * @dataProvider providerNonPausedStatuses
   */
  public function testNonPausedStatusIsIgnored(string $status): void {
    $this->messageBus->dispatch(Argument::cetera())->shouldNotBeCalled();

    $this->subscriber->onPipelineCompleted(new PipelineCompletedEvent(
      [PausedReason::CONTEXT_KEY => PausedReason::TIME_BUDGET],
      'exec_1',
      $status,
      1000,
      'pipe_1',
    ));
  }

  /**
   * Data provider: terminal statuses that must be ignored.
   *
   * @return array<string, array{string}>
   *   The statuses.
   */
  public static function providerNonPausedStatuses(): array {
    return [
      'completed' => [PipelineStatus::COMPLETED],
      'failed' => [PipelineStatus::FAILED],
    ];
  }

  /**
   * A pause without a stamped budget reason (HITL) raises no signal.
   *
   * @dataProvider providerNonBudgetReasons
   */
  public function testPauseWithoutBudgetReasonIsIgnored(array $context): void {
    $this->messageBus->dispatch(Argument::cetera())->shouldNotBeCalled();

    $this->subscriber->onPipelineCompleted(new PipelineCompletedEvent(
      $context,
      'exec_1',
      PipelineStatus::PAUSED,
      1000,
      'pipe_1',
    ));
  }

  /**
   * Data provider: contexts that must not be treated as budget pauses.
   *
   * @return array<string, array{array<string, mixed>}>
   *   The contexts.
   */
  public static function providerNonBudgetReasons(): array {
    return [
      'no reason (HITL pause)' => [[]],
      'unknown reason value' => [[PausedReason::CONTEXT_KEY => 'something_else']],
    ];
  }

  /**
   * A paused run without a pipeline entity ID raises no signal.
   */
  public function testMissingPipelineIdIsIgnored(): void {
    $this->messageBus->dispatch(Argument::cetera())->shouldNotBeCalled();

    $this->subscriber->onPipelineCompleted(new PipelineCompletedEvent(
      [PausedReason::CONTEXT_KEY => PausedReason::TIME_BUDGET],
      'exec_1',
      PipelineStatus::PAUSED,
      1000,
      // No pipeline id (non-entity-backed run).
    ));
  }

  /**
   * An already-pending inward signal blocks a second (system) one.
   *
   * Covers both idempotence (event replay) and the operator-pause case:
   * an operator Pause raised mid-run must not be shadowed.
   */
  public function testExistingPendingSignalBlocksDispatch(): void {
    $existing = $this->prophesize(FlowDropInterruptInterface::class);
    $this->interruptManager->findPendingInwardSignal('pipe_1')
      ->willReturn($existing->reveal());
    $this->messageBus->dispatch(Argument::cetera())->shouldNotBeCalled();

    $this->subscriber->onPipelineCompleted(new PipelineCompletedEvent(
      [PausedReason::CONTEXT_KEY => PausedReason::MAX_ITERATIONS],
      'exec_1',
      PipelineStatus::PAUSED,
      1000,
      'pipe_1',
    ));
  }

  /**
   * A budget pause raises a system Pause signal carrying the reason.
   *
   * @dataProvider providerBudgetReasons
   */
  public function testBudgetPauseRaisesSystemPauseSignal(string $reason): void {
    $this->interruptManager->findPendingInwardSignal('pipe_1')->willReturn(NULL);

    $pipeline = $this->prophesize(FlowDropPipelineRuntimeInterface::class);
    $pipeline->getWorkflowId()->willReturn('wf_1');
    $storage = $this->prophesize(EntityStorageInterface::class);
    $storage->load('pipe_1')->willReturn($pipeline->reveal());
    $this->entityTypeManager->getStorage('flowdrop_pipeline')
      ->willReturn($storage->reveal());

    $signal = $this->prophesize(FlowDropInterruptInterface::class);
    $signal->uuid()->willReturn('uuid-signal');

    $this->messageBus->dispatch(
      Argument::that(function (InwardSignalMessage $message) use ($reason): bool {
        return $message->getType() === InterruptType::Pause
          && $message->targetPipelineId === 'pipe_1'
          && $message->initiatorUid === NULL
          && $message->reason === $reason;
      }),
      Argument::that(function (ExecutionStamp $stamp): bool {
        return $stamp->workflowId === 'wf_1'
          && $stamp->nodeId === 'signal:pause';
      }),
    )->shouldBeCalledOnce()->willReturn($signal->reveal());

    $this->subscriber->onPipelineCompleted(new PipelineCompletedEvent(
      [PausedReason::CONTEXT_KEY => $reason],
      'exec_1',
      PipelineStatus::PAUSED,
      1000,
      'pipe_1',
    ));
  }

  /**
   * Data provider: the two budget reasons.
   *
   * @return array<string, array{string}>
   *   The reasons.
   */
  public static function providerBudgetReasons(): array {
    return [
      'time budget' => [PausedReason::TIME_BUDGET],
      'max iterations' => [PausedReason::MAX_ITERATIONS],
    ];
  }

  /**
   * A bus failure is swallowed and logged — the pause itself must stand.
   */
  public function testDispatchFailureIsSwallowed(): void {
    $this->interruptManager->findPendingInwardSignal('pipe_1')->willReturn(NULL);

    $pipeline = $this->prophesize(FlowDropPipelineRuntimeInterface::class);
    $pipeline->getWorkflowId()->willReturn('wf_1');
    $storage = $this->prophesize(EntityStorageInterface::class);
    $storage->load('pipe_1')->willReturn($pipeline->reveal());
    $this->entityTypeManager->getStorage('flowdrop_pipeline')
      ->willReturn($storage->reveal());

    $this->messageBus->dispatch(Argument::cetera())
      ->willThrow(new \LogicException('no handler'));

    $this->subscriber->onPipelineCompleted(new PipelineCompletedEvent(
      [PausedReason::CONTEXT_KEY => PausedReason::TIME_BUDGET],
      'exec_1',
      PipelineStatus::PAUSED,
      1000,
      'pipe_1',
    ));

    // Reaching this point without an exception is the assertion.
    $this->addToAssertionCount(1);
  }

}
