<?php

declare(strict_types=1);

namespace Drupal\flowdrop_interrupt\EventSubscriber;

use Drupal\flowdrop_interrupt\Enum\InterruptDirection;
use Drupal\flowdrop_interrupt\Enum\InterruptStatus;
use Drupal\flowdrop_interrupt\Enum\InterruptType;
use Drupal\flowdrop_interrupt\Service\InterruptManagerInterface;
use Drupal\flowdrop_runtime\Event\PipelineSignalPollEvent;
use Drupal\flowdrop_runtime\Service\Runtime\InterruptCancellationDispatcher;
use Psr\Log\LoggerInterface;
use Symfony\Component\EventDispatcher\EventSubscriberInterface;

/**
 * Answers the runtime's inward-signal poll with any pending Cancel/Pause.
 *
 * Subscribes to {@see PipelineSignalPollEvent} (dispatched by the
 * orchestrator between job iterations) and runs the indexed
 * `(target_pipeline_id, direction, status)` lookup. If a Cancel signal is
 * found it's acknowledged here (and pending outward interrupts on the
 * pipeline are cancelled as a side effect). Pause signals stay pending
 * — the signal entity IS the resume key.
 *
 * @internal
 */
final class PipelineSignalPollSubscriber implements EventSubscriberInterface {

  /**
   * Minimum interval between DB lookups for the same pipeline (microseconds).
   *
   * The orchestrator dispatches PipelineSignalPollEvent on every job
   * iteration. For pipelines made up of many short jobs this would mean a
   * per-iteration indexed entity query — measurable overhead. We memoise
   * "no signal" answers for this window so the lookup runs at most once per
   * 100 ms per pipeline. Cancel/Pause therefore has up to 100 ms latency
   * before it's noticed by the orchestrator, which is well below human
   * tolerance and far smaller than the typical sub-second job in practice.
   *
   * Subscribers MUST be cheap and idempotent — see PipelineSignalPollEvent.
   */
  private const POLL_MIN_INTERVAL_US = 100_000;

  /**
   * Per-pipeline "last DB lookup" timestamp in microseconds.
   *
   * @var array<string, float>
   */
  private array $lastLookupAt = [];

  public function __construct(
    private readonly InterruptManagerInterface $interruptManager,
    private readonly LoggerInterface $logger,
    private readonly ?InterruptCancellationDispatcher $cancellationDispatcher = NULL,
  ) {}

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

  /**
   * Looks up any pending inward signal and attaches it to the event.
   *
   * Errors are logged — the runtime expects polling to be best-effort.
   * If the lookup fails the pipeline keeps running, which is the safe
   * default (operator can re-issue the signal).
   *
   * The DB lookup is throttled per pipeline via
   * {@see self::POLL_MIN_INTERVAL_US}
   * so a pipeline with many short jobs doesn't pay the full lookup cost on
   * every iteration. Once a signal IS found (or the throttle window elapses)
   * we re-query.
   */
  public function onPoll(PipelineSignalPollEvent $event): void {
    $now = microtime(TRUE) * 1_000_000;
    $lastAt = $this->lastLookupAt[$event->pipelineId] ?? 0.0;
    if ($lastAt > 0.0 && ($now - $lastAt) < self::POLL_MIN_INTERVAL_US) {
      // Within the throttle window — treat as no-signal.
      return;
    }

    try {
      $signal = $this->interruptManager->findPendingInwardSignal($event->pipelineId);
      $this->lastLookupAt[$event->pipelineId] = $now;
      if ($signal === NULL) {
        return;
      }

      $type = $signal->getType();
      $signalId = (string) ($signal->uuid() ?? '');
      $reason = $signal->getReason();
      $initiatorUid = $signal->getInitiatorUid();

      switch ($type) {
        case InterruptType::Cancel:
          // Cancel: invoke per-node cancel() hooks (Phase 4) BEFORE marking
          // the entities cancelled, then acknowledge the signal. The
          // orchestrator marks the pipeline cancelled and emits
          // ExecutionCancelledEvent.
          $cancelledCount = $this->cancelOutwardInterruptsWithHooks($event->pipelineId);
          if ($cancelledCount > 0) {
            $this->logger->info(
              'Cancel signal @id: cancelled @n outward interrupt(s) on pipeline @pipeline.',
              [
                '@id' => $signalId,
                '@n' => $cancelledCount,
                '@pipeline' => $event->pipelineId,
              ]);
          }
          $signal->acknowledge();
          $signal->save();
          $event->attachCancel($signalId, $reason, $initiatorUid);
          return;

        case InterruptType::Pause:
          // Pause: leave the signal pending — resolving it via the inbox
          // or `/resume` API is what unpauses the pipeline.
          $event->attachPause($signalId, $reason, $initiatorUid);
          return;

        default:
          $this->logger->warning(
            'PipelineSignalPollSubscriber: unexpected inward signal type @type for interrupt @id; ignoring.',
            ['@type' => $type->value, '@id' => $signalId],
          );
      }
    }
    catch (\Throwable $e) {
      $this->logger->error(
        'PipelineSignalPollSubscriber failed for pipeline @pipeline: @msg',
        ['@pipeline' => $event->pipelineId, '@msg' => $e->getMessage()],
      );
    }
  }

  /**
   * Cancels pending outward interrupts, invoking node cancel() hooks first.
   *
   * Phase 4 iterates manually (rather than calling the manager's bulk
   * cancelOutwardInterruptsForPipeline) so each interrupt's owning node
   * can release resources before the entity status flips to cancelled.
   *
   * @return int
   *   Number of interrupts cancelled.
   */
  private function cancelOutwardInterruptsWithHooks(string $pipelineId): int {
    $pending = $this->interruptManager->getPendingInterruptsForPipeline($pipelineId);
    $count = 0;
    foreach ($pending as $interrupt) {
      if ($interrupt->getDirection() !== InterruptDirection::Outward) {
        continue;
      }
      if ($interrupt->getStatus() !== InterruptStatus::Pending) {
        continue;
      }
      // Invoke node cancel() hook (best-effort) before flipping status.
      if ($this->cancellationDispatcher !== NULL) {
        $this->cancellationDispatcher->invokeCancelHook(
          $interrupt->getJobId(),
          $interrupt,
        );
      }
      $interrupt->cancel();
      $interrupt->save();
      $count++;
    }
    return $count;
  }

}
