<?php

declare(strict_types=1);

namespace Drupal\flowdrop_interrupt\Plugin\FlowDropNodeProcessor;

use Drupal\Core\Flood\FloodInterface;
use Drupal\Core\StringTranslation\TranslatableMarkup;
use Drupal\flowdrop\Attribute\FlowDropNodeProcessor;
use Drupal\flowdrop\DTO\ParameterBagInterface;
use Drupal\flowdrop\DTO\ValidationResult;
use Drupal\flowdrop_interrupt\Exception\InterruptRequiredException;
use Drupal\flowdrop_interrupt\Message\TimerInterruptMessage;
use Drupal\flowdrop_orchestration\DTO\ResolvedInterruptInterface;
use Symfony\Component\DependencyInjection\ContainerInterface;

/**
 * Rate limit node processor — a flood-backed gate on workflow throughput.
 *
 * Two pieces working together:
 * - **Cross-run coordination** comes from core's flood service: a shared
 *   events-per-window counter keyed by the bucket identifier, so all
 *   pipelines passing through the same bucket collectively stay under the
 *   limit.
 * - **The durable wait** comes from the Timer interrupt machinery: when
 *   the bucket is full, the node pauses with a timer paced at
 *   `window / limit` seconds; the cron sweep resumes it, and `resume()`
 *   RE-CHECKS the bucket — other runs may have refilled it while this one
 *   waited — re-arming a fresh timer when still over the limit.
 *
 * Resolving the timer early through the inbox or resume API cannot bypass
 * the limit: it only triggers an early re-check.
 *
 * The limit is best-effort, not a hard guarantee: the flood check and
 * register are not atomic, so N runs racing the same bucket in the same
 * instant can each pass the check before any registers. Right for being
 * polite to vendors and pacing bulk work; wrong for billing enforcement.
 *
 * Precision is cron-bound, so this suits coarse limits ("100 calls/hour
 * to this vendor"), not per-second throttling — throttle inside the
 * calling node processor for sub-minute rates.
 *
 * Config:
 * - limit: Maximum passes per window.
 * - window: The window length in seconds.
 * - identifier: Bucket key (connectable; defaults to workflow + node ID,
 *   giving each node instance its own bucket).
 *
 * Outputs: the timer audit trail (scheduled_at / resumed_at /
 * drift_seconds / resolved_by_timer / user_id) plus `throttled`.
 *
 * @internal This class is not part of the public API and may change without notice.
 */
#[FlowDropNodeProcessor(
  id: "rate_limit",
  label: new TranslatableMarkup("Rate Limit"),
  description: "Limits how often execution passes through, pausing excess runs until the window frees up",
  version: "1.0.0"
)]
class RateLimitNode extends AbstractTimerNode {

  /**
   * The flood event name shared by all rate-limit buckets.
   *
   * Buckets are separated by the flood identifier, not the event name.
   */
  protected const FLOOD_NAME = 'flowdrop_interrupt.rate_limit';

  /**
   * The flood service.
   */
  protected FloodInterface $flood;

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

  /**
   * {@inheritdoc}
   */
  public function validateParams(array $params): ValidationResult {
    foreach (['limit', 'window'] as $name) {
      $value = $params[$name] ?? NULL;
      if ($value === NULL || $value === '' || !is_numeric($value) || (int) $value < 1) {
        return ValidationResult::error($name, ucfirst($name) . " must be a positive integer.", "INVALID_VALUE");
      }
    }
    return ValidationResult::success();
  }

  /**
   * {@inheritdoc}
   *
   * Average pacing: with the bucket full, the earliest a slot can
   * plausibly free is one inter-arrival interval away.
   */
  protected function computeFireAt(ParameterBagInterface $params): int {
    $limit = max(1, $params->getInt('limit', 1));
    $window = max(1, $params->getInt('window', 60));
    return time() + max(1, (int) ceil($window / $limit));
  }

  /**
   * {@inheritdoc}
   */
  public function process(ParameterBagInterface $params): array {
    if ($this->tryAcquire($params)) {
      $now = time();
      return $this->buildOutputs($now, $now, TRUE, NULL) + ['throttled' => FALSE];
    }
    throw $this->buildWaitException($params);
  }

  /**
   * {@inheritdoc}
   *
   * The bucket is shared: other runs may have refilled it while this one
   * waited, so the slot is re-checked rather than assumed. Still over the
   * limit → re-arm a fresh timer (the runtime catches interrupt
   * exceptions from resume() exactly like process()).
   */
  public function resume(
    ResolvedInterruptInterface $interrupt,
    ParameterBagInterface $params,
  ): array {
    if (!$this->tryAcquire($params)) {
      throw $this->buildWaitException($params);
    }
    return parent::resume($interrupt, $params) + ['throttled' => TRUE];
  }

  /**
   * Registers a pass through the bucket if the limit allows it.
   *
   * @param \Drupal\flowdrop\DTO\ParameterBagInterface $params
   *   The node parameters.
   *
   * @return bool
   *   TRUE when a slot was acquired (and counted), FALSE when the bucket
   *   is full.
   */
  protected function tryAcquire(ParameterBagInterface $params): bool {
    $limit = max(1, $params->getInt('limit', 1));
    $window = max(1, $params->getInt('window', 60));
    $key = $this->bucketKey($params);

    if (!$this->flood->isAllowed(self::FLOOD_NAME, $limit, $window, $key)) {
      return FALSE;
    }
    $this->flood->register(self::FLOOD_NAME, $window, $key);
    return TRUE;
  }

  /**
   * Builds the pause exception for a full bucket.
   */
  protected function buildWaitException(ParameterBagInterface $params): InterruptRequiredException {
    $fireAt = $this->computeFireAt($params);
    $key = $this->bucketKey($params);
    return new InterruptRequiredException(new TimerInterruptMessage(
      message: sprintf('Rate limit reached for "%s"; retrying at %s', $key, date('c', $fireAt)),
      fireAt: $fireAt,
      context: ['scheduled_at' => $fireAt, 'rate_limit_key' => $key],
    ), [], 'Workflow paused by rate limit');
  }

  /**
   * Maximum raw key length passed to flood unhashed.
   *
   * Core's flood table stores identifiers as varchar_ascii(128); the
   * prefixed event name shares nothing with the identifier column, but
   * keys are normalized well under the limit to stay safe and readable.
   */
  protected const MAX_RAW_KEY_LENGTH = 64;

  /**
   * Resolves the bucket identifier.
   *
   * Defaults to `{workflowId}:{nodeId}` so each node instance gets its own
   * bucket; connect or configure `identifier` to share a bucket across
   * nodes/workflows or to partition per user, domain, etc.
   *
   * The identifier is connectable, so arbitrary runtime values (URLs,
   * emails, multibyte strings) can land here — but core's flood table
   * stores identifiers as `varchar_ascii(128)`. Anything long or
   * non-ASCII is hashed; same input, same bucket, no DB exception.
   *
   * @param \Drupal\flowdrop\DTO\ParameterBagInterface $params
   *   The node parameters.
   *
   * @return string
   *   The flood identifier, guaranteed short ASCII.
   */
  protected function bucketKey(ParameterBagInterface $params): string {
    $identifier = trim($params->getString('identifier'));
    if ($identifier === '') {
      $identifier = $this->executionContext !== NULL
        ? $this->executionContext->getWorkflowId() . ':' . $this->executionContext->getNodeId()
        : 'default';
    }
    if (strlen($identifier) > self::MAX_RAW_KEY_LENGTH || !mb_check_encoding($identifier, 'ASCII')) {
      return 'sha:' . hash('sha256', $identifier);
    }
    return $identifier;
  }

  /**
   * {@inheritdoc}
   */
  public function getParameterSchema(): array {
    return [
      "type" => "object",
      "properties" => [
        "limit" => [
          "type" => "integer",
          "title" => "Limit",
          "description" => "Maximum number of passes per window",
          "minimum" => 1,
          "default" => 10,
        ],
        "window" => [
          "type" => "integer",
          "title" => "Window (seconds)",
          "description" => "The window length in seconds (e.g. 3600 for per-hour limits)",
          "minimum" => 1,
          "default" => 3600,
        ],
        "identifier" => [
          "type" => "string",
          "title" => "Bucket identifier",
          "description" => "Runs sharing this key share the limit. Leave empty for a per-node bucket; connect a value to partition per user, domain, etc.",
        ],
        "__interrupt_id__" => $this->interruptIdParameterSchema(),
      ],
      "required" => ["limit", "window"],
    ];
  }

  /**
   * {@inheritdoc}
   */
  public function getOutputSchema(): array {
    $schema = parent::getOutputSchema();
    $schema["properties"]["throttled"] = [
      "type" => "boolean",
      "description" => "TRUE when the run had to wait for a slot, FALSE when it passed straight through",
    ];
    return $schema;
  }

}
