<?php

declare(strict_types=1);

namespace Drupal\flowdrop_interrupt\Plugin\FlowDropNodeProcessor;

use Drupal\Core\Config\ConfigFactoryInterface;
use Drupal\flowdrop\DTO\ExecutionContextDTO;
use Drupal\flowdrop\DTO\ParameterBagInterface;
use Drupal\flowdrop\Plugin\FlowDropNodeProcessor\AbstractFlowDropNodeProcessor;
use Drupal\flowdrop\Plugin\FlowDropNodeProcessor\ExecutionContextAwareInterface;
use Drupal\flowdrop\Plugin\FlowDropNodeProcessor\ResumableNodeProcessorInterface;
use Drupal\flowdrop_interrupt\Exception\InterruptRequiredException;
use Drupal\flowdrop_interrupt\Message\TimerInterruptMessage;
use Drupal\flowdrop_orchestration\DTO\ResolvedInterruptInterface;
use Symfony\Component\DependencyInjection\ContainerInterface;

/**
 * Shared base for time-based pause nodes (Delay, Schedule).
 *
 * Subclasses supply {@see self::computeFireAt()}; this base owns the
 * pause/resume machine:
 *
 * - `process()` computes the fire-at time. Already past → returns outputs
 *   immediately (no pause, no cron round-trip). Otherwise it throws an
 *   {@see InterruptRequiredException} with a {@see TimerInterruptMessage};
 *   the cron-driven ScheduledInterruptSweeper resolves the persisted Timer
 *   interrupt once due, resuming the pipeline.
 * - `resume()` builds outputs from the resolved interrupt — whether the
 *   clock fired (`resolved_by_timer = true`) or a human skipped the wait
 *   through the inbox or /resume API (`false`, with `user_id` set).
 *
 * Requires a pipeline-backed orchestrator: under the stateless synchronous
 * orchestrator there is no pipeline to resume, so the interrupt would
 * pause the run permanently.
 *
 * @internal This class is not part of the public API and may change without notice.
 */
abstract class AbstractTimerNode extends AbstractFlowDropNodeProcessor implements ExecutionContextAwareInterface, ResumableNodeProcessorInterface {

  /**
   * Fallback for `timer_max_delay` when the config key is absent.
   *
   * One year, matching config/install. Sites updated from a pre-timer
   * release won't have the key until they re-import config.
   */
  protected const MAX_DELAY_FALLBACK = 31536000;

  /**
   * The execution context.
   */
  protected ?ExecutionContextDTO $executionContext = NULL;

  /**
   * The config factory.
   */
  protected ConfigFactoryInterface $configFactory;

  /**
   * {@inheritdoc}
   */
  public static function create(
    ContainerInterface $container,
    array $configuration,
    $plugin_id,
    $plugin_definition,
  ): static {
    $instance = new static($configuration, $plugin_id, $plugin_definition);
    $instance->configFactory = $container->get('config.factory');
    return $instance;
  }

  /**
   * {@inheritdoc}
   */
  public function setExecutionContext(ExecutionContextDTO $context): void {
    $this->executionContext = $context;
  }

  /**
   * Computes the absolute fire-at timestamp from the node's parameters.
   *
   * @param \Drupal\flowdrop\DTO\ParameterBagInterface $params
   *   The node parameters.
   *
   * @return int
   *   Unix timestamp at which the workflow should resume.
   *
   * @throws \InvalidArgumentException
   *   When the parameters can't produce a timestamp (e.g. unparseable
   *   datetime fed through a connection at runtime).
   */
  abstract protected function computeFireAt(ParameterBagInterface $params): int;

  /**
   * {@inheritdoc}
   *
   * First-time entry only — the orchestrator routes resume invocations to
   * {@see self::resume()} via the ResumableNodeProcessorInterface contract.
   */
  public function process(ParameterBagInterface $params): array {
    $fireAt = $this->computeFireAt($params);
    $now = time();

    // Already due: pass straight through instead of a pause + cron
    // round-trip that would only add latency.
    if ($fireAt <= $now) {
      return $this->buildOutputs($fireAt, $now, TRUE, NULL);
    }

    throw new InterruptRequiredException(new TimerInterruptMessage(
      message: sprintf('Waiting until %s', date('c', $fireAt)),
      fireAt: $fireAt,
      context: ['scheduled_at' => $fireAt],
    ), [], 'Workflow paused on a timer');
  }

  /**
   * {@inheritdoc}
   *
   * Reads the timer-vs-human signal from the response payload written by
   * TimerInterruptHandler (sweep: `resolved_by_timer = true`; inbox or
   * /resume API skip: `false`).
   */
  public function resume(
    ResolvedInterruptInterface $interrupt,
    ParameterBagInterface $params,
  ): array {
    $responseData = $interrupt->getResponseData();
    $responseData = is_array($responseData) ? $responseData : [];

    $resumedAt = $interrupt->getResolvedAt() ?? time();
    $scheduledAt = (int) ($responseData['scheduled_at'] ?? $resumedAt);
    $byTimer = (bool) ($responseData['resolved_by_timer'] ?? FALSE);

    return $this->buildOutputs($scheduledAt, $resumedAt, $byTimer, $interrupt->getResolvedBy());
  }

  /**
   * {@inheritdoc}
   */
  public function getOutputSchema(): array {
    return [
      "type" => "object",
      "properties" => [
        "scheduled_at" => [
          "type" => "integer",
          "description" => "Unix timestamp the workflow was scheduled to resume at",
        ],
        "resumed_at" => [
          "type" => "integer",
          "description" => "Unix timestamp the workflow actually resumed at",
        ],
        "drift_seconds" => [
          "type" => "integer",
          "description" => "Seconds between scheduled and actual resume (negative when resumed early)",
        ],
        "resolved_by_timer" => [
          "type" => "boolean",
          "description" => "TRUE when the timer fired; FALSE when a user skipped the wait",
        ],
        "user_id" => [
          "type" => "integer",
          "description" => "ID of the user who skipped the wait (null when the timer fired)",
        ],
      ],
    ];
  }

  /**
   * Builds the node outputs.
   *
   * @param int $scheduledAt
   *   The scheduled fire-at timestamp.
   * @param int $resumedAt
   *   The actual resume timestamp.
   * @param bool $byTimer
   *   TRUE when the timer fired (or the delay was already due), FALSE when
   *   a user skipped the wait.
   * @param int|null $resolvedBy
   *   The skipping user's ID, if any.
   *
   * @return array<string, mixed>
   *   The outputs.
   */
  protected function buildOutputs(int $scheduledAt, int $resumedAt, bool $byTimer, ?int $resolvedBy): array {
    return [
      "scheduled_at" => $scheduledAt,
      "resumed_at" => $resumedAt,
      "drift_seconds" => $resumedAt - $scheduledAt,
      "resolved_by_timer" => $byTimer,
      "user_id" => $byTimer ? NULL : $resolvedBy,
    ];
  }

  /**
   * Returns the configured maximum delay in seconds (0 = unlimited).
   */
  protected function getMaxDelay(): int {
    $value = $this->configFactory
      ->get('flowdrop_interrupt.settings')
      ->get('timer_max_delay');
    return (int) ($value ?? self::MAX_DELAY_FALLBACK);
  }

  /**
   * Shared schema fragment for the internal resume parameter.
   *
   * @return array<string, mixed>
   *   The `__interrupt_id__` property definition.
   */
  protected function interruptIdParameterSchema(): array {
    return [
      "type" => "string",
      "description" => "Internal: Interrupt ID for resume operations",
      "connectable" => FALSE,
      "configurable" => FALSE,
    ];
  }

}
