<?php

declare(strict_types=1);

namespace Drupal\flowdrop_interrupt\EventSubscriber;

use Drupal\Core\Entity\EntityTypeManagerInterface;
use Drupal\flowdrop_interrupt\DTO\ExecutionStamp;
use Drupal\flowdrop_interrupt\Enum\InterruptType;
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 Psr\Log\LoggerInterface;
use Symfony\Component\EventDispatcher\EventSubscriberInterface;

/**
 * Converts a budget pause into a pending system Pause signal.
 *
 * A run that stops on its execution budget (`timeout` / `max_iterations`)
 * pauses with ready jobs remaining and a stamped
 * {@see PausedReason::CONTEXT_KEY} — but, unlike a HITL pause, with no
 * interrupt entity, so neither resume subscriber could ever fire. This
 * subscriber creates the missing inward Pause signal (initiator NULL =
 * system), making the standard resume chain work verbatim: `POST
 * .../resume` (or inbox resolution) resolves the signal,
 * InterruptResolvedSubscriber re-executes the pipeline with a fresh
 * budget window.
 *
 * Gates:
 * - status must be 'paused' (completed/failed runs are not resumable);
 * - a budget reason must be stamped (HITL pauses carry none and already
 *   have their own interrupt entity);
 * - no inward signal may already be pending (idempotence, and an operator
 *   Pause raised mid-run must not be shadowed by a system one).
 *
 * The orchestrators live without any interrupt dependency; this is the
 * signal-machinery side of the bridge, mirroring
 * {@see PipelineSignalPollSubscriber}.
 *
 * @internal
 */
final class BudgetPauseSignalSubscriber implements EventSubscriberInterface {

  public function __construct(
    private readonly InterruptManagerInterface $interruptManager,
    private readonly InterruptMessageBusInterface $messageBus,
    private readonly EntityTypeManagerInterface $entityTypeManager,
    private readonly LoggerInterface $logger,
  ) {}

  /**
   * {@inheritdoc}
   */
  public static function getSubscribedEvents(): array {
    return [
      PipelineCompletedEvent::NAME => 'onPipelineCompleted',
    ];
  }

  /**
   * Raises a system Pause signal for a budget-paused pipeline.
   *
   * @param \Drupal\flowdrop_runtime\Event\PipelineCompletedEvent $event
   *   The pipeline-completed event.
   */
  public function onPipelineCompleted(PipelineCompletedEvent $event): void {
    if ($event->getStatus() !== PipelineStatus::PAUSED) {
      return;
    }

    $reason = $event->getPipelineContext()[PausedReason::CONTEXT_KEY] ?? NULL;
    if (!in_array($reason, PausedReason::budgetReasons(), TRUE)) {
      return;
    }

    $pipelineId = $event->getPipelineId();
    if ($pipelineId === NULL || $pipelineId === '') {
      return;
    }

    if ($this->interruptManager->findPendingInwardSignal($pipelineId) !== NULL) {
      return;
    }

    try {
      $signal = $this->messageBus->dispatch(
        new InwardSignalMessage(
          type: InterruptType::Pause,
          message: sprintf('Execution budget reached (%s); resolve to resume with a fresh budget window', $reason),
          targetPipelineId: $pipelineId,
          // NULL initiator marks the signal as system-raised.
          initiatorUid: NULL,
          reason: $reason,
        ),
        // Budget signals have no originating node — same marker convention
        // as PipelineSignalController.
        new ExecutionStamp(
          workflowId: $this->resolveWorkflowId($pipelineId),
          nodeId: 'signal:' . InterruptType::Pause->value,
        ),
      );

      $this->logger->info('Raised system Pause signal @signal for budget-paused pipeline @pipeline (reason: @reason)', [
        '@signal' => $signal->uuid(),
        '@pipeline' => $pipelineId,
        '@reason' => $reason,
      ]);
    }
    catch (\Throwable $e) {
      // Never let signal bookkeeping break the run that just finished
      // pausing — the pipeline stays paused either way; only the
      // one-click resume affordance is lost.
      $this->logger->error('Failed to raise system Pause signal for pipeline @pipeline: @message', [
        '@pipeline' => $pipelineId,
        '@message' => $e->getMessage(),
      ]);
    }
  }

  /**
   * Resolves the workflow ID for the interrupt's execution stamp.
   *
   * @param string $pipelineId
   *   The pipeline entity ID.
   *
   * @return string
   *   The workflow ID, or '' when the pipeline cannot be loaded.
   */
  private function resolveWorkflowId(string $pipelineId): string {
    $pipeline = $this->entityTypeManager
      ->getStorage('flowdrop_pipeline')
      ->load($pipelineId);
    return $pipeline instanceof FlowDropPipelineRuntimeInterface
      ? $pipeline->getWorkflowId()
      : '';
  }

}
