<?php

declare(strict_types=1);

namespace Drupal\Tests\flowdrop_interrupt\Unit\Plugin\FlowDropNodeProcessor;

use Drupal\Core\Flood\FloodInterface;
use Drupal\flowdrop\DTO\ExecutionContextDTO;
use Drupal\flowdrop\DTO\ParameterBagInterface;
use Drupal\flowdrop_interrupt\Exception\InterruptRequiredException;
use Drupal\flowdrop_interrupt\Message\TimerInterruptMessage;
use Drupal\flowdrop_interrupt\Plugin\FlowDropNodeProcessor\RateLimitNode;
use Drupal\flowdrop_orchestration\DTO\ResolvedInterruptInterface;
use Drupal\Tests\UnitTestCase;
use Prophecy\PhpUnit\ProphecyTrait;
use Symfony\Component\DependencyInjection\ContainerBuilder;

/**
 * Tests the RateLimitNode flood gate and re-arming resume.
 *
 * @coversDefaultClass \Drupal\flowdrop_interrupt\Plugin\FlowDropNodeProcessor\RateLimitNode
 * @group flowdrop_interrupt
 */
class RateLimitNodeTest extends UnitTestCase {

  use ProphecyTrait;

  /**
   * The flood event name the node registers against.
   */
  protected const FLOOD_NAME = 'flowdrop_interrupt.rate_limit';

  /**
   * Builds a RateLimitNode with the given flood double.
   */
  protected function createNode(FloodInterface $flood): RateLimitNode {
    $container = new ContainerBuilder();
    $container->set('config.factory', $this->getConfigFactoryStub([
      'flowdrop_interrupt.settings' => ['timer_max_delay' => 31536000],
    ]));
    $container->set('flood', $flood);
    return RateLimitNode::create($container, [], 'rate_limit', []);
  }

  /**
   * Builds a parameter bag prophecy for limit/window/identifier.
   */
  protected function params(int $limit, int $window, string $identifier = 'bucket-a'): ParameterBagInterface {
    $params = $this->prophesize(ParameterBagInterface::class);
    $params->getInt('limit', 1)->willReturn($limit);
    $params->getInt('window', 60)->willReturn($window);
    $params->getString('identifier')->willReturn($identifier);
    return $params->reveal();
  }

  /**
   * An open bucket passes through, registers the event, and is not throttled.
   */
  public function testProcessPassesThroughWhenAllowed(): void {
    $flood = $this->prophesize(FloodInterface::class);
    $flood->isAllowed(self::FLOOD_NAME, 10, 3600, 'bucket-a')->willReturn(TRUE);
    $flood->register(self::FLOOD_NAME, 3600, 'bucket-a')->shouldBeCalledOnce();

    $output = $this->createNode($flood->reveal())->process($this->params(10, 3600));

    $this->assertFalse($output['throttled']);
    $this->assertTrue($output['resolved_by_timer']);
    $this->assertSame(0, $output['drift_seconds']);
  }

  /**
   * A full bucket throws a timer paced at window/limit, without registering.
   */
  public function testProcessThrowsPacedTimerWhenBlocked(): void {
    $flood = $this->prophesize(FloodInterface::class);
    $flood->isAllowed(self::FLOOD_NAME, 10, 3600, 'bucket-a')->willReturn(FALSE);
    $flood->register(self::FLOOD_NAME, 3600, 'bucket-a')->shouldNotBeCalled();

    try {
      $this->createNode($flood->reveal())->process($this->params(10, 3600));
      $this->fail('Expected InterruptRequiredException');
    }
    catch (InterruptRequiredException $e) {
      $message = $e->getInterruptMessage();
      $this->assertInstanceOf(TimerInterruptMessage::class, $message);
      // ceil(3600 / 10) = 360 seconds of average pacing.
      $this->assertEqualsWithDelta(time() + 360, $message->fireAt, 2);
      $this->assertSame('bucket-a', $message->getContext()['rate_limit_key']);
    }
  }

  /**
   * Resume() re-checks the bucket: a freed slot is acquired and throttled.
   */
  public function testResumeAcquiresFreedSlot(): void {
    $flood = $this->prophesize(FloodInterface::class);
    $flood->isAllowed(self::FLOOD_NAME, 10, 3600, 'bucket-a')->willReturn(TRUE);
    $flood->register(self::FLOOD_NAME, 3600, 'bucket-a')->shouldBeCalledOnce();

    $interrupt = $this->prophesize(ResolvedInterruptInterface::class);
    $interrupt->getResponseData()->willReturn([
      'resolved_by_timer' => TRUE,
      'scheduled_at' => 1700000000,
    ]);
    $interrupt->getResolvedAt()->willReturn(1700000005);
    $interrupt->getResolvedBy()->willReturn(NULL);

    $output = $this->createNode($flood->reveal())
      ->resume($interrupt->reveal(), $this->params(10, 3600));

    $this->assertTrue($output['throttled']);
    $this->assertTrue($output['resolved_by_timer']);
    $this->assertSame(5, $output['drift_seconds']);
  }

  /**
   * Resume() re-arms a fresh timer when the bucket is still full.
   *
   * Other runs may have refilled the shared bucket while this one waited —
   * including a human resolving the timer early, which therefore cannot
   * bypass the limit.
   */
  public function testResumeReArmsWhenStillBlocked(): void {
    $flood = $this->prophesize(FloodInterface::class);
    $flood->isAllowed(self::FLOOD_NAME, 10, 3600, 'bucket-a')->willReturn(FALSE);
    $flood->register(self::FLOOD_NAME, 3600, 'bucket-a')->shouldNotBeCalled();

    $interrupt = $this->prophesize(ResolvedInterruptInterface::class);

    $this->expectException(InterruptRequiredException::class);
    $this->createNode($flood->reveal())
      ->resume($interrupt->reveal(), $this->params(10, 3600));
  }

  /**
   * The default bucket key is workflow:node from the execution context.
   */
  public function testDefaultBucketKeyFromExecutionContext(): void {
    $flood = $this->prophesize(FloodInterface::class);
    $flood->isAllowed(self::FLOOD_NAME, 5, 60, 'wf_1:node_a')->willReturn(TRUE);
    $flood->register(self::FLOOD_NAME, 60, 'wf_1:node_a')->shouldBeCalledOnce();

    $node = $this->createNode($flood->reveal());
    $node->setExecutionContext(new ExecutionContextDTO([], 'wf_1', 'pipe-1', 'exec-1', 'node_a'));

    $output = $node->process($this->params(5, 60, identifier: ''));
    $this->assertFalse($output['throttled']);
  }

  /**
   * Long and non-ASCII identifiers are hashed to fit flood's ASCII column.
   *
   * Core's flood table stores identifiers as varchar_ascii(128); the
   * connectable identifier param can carry URLs, emails, or multibyte
   * strings, which must never reach flood raw.
   *
   * @dataProvider unsafeIdentifierProvider
   */
  public function testUnsafeIdentifiersAreHashed(string $identifier): void {
    $expected = 'sha:' . hash('sha256', $identifier);

    $flood = $this->prophesize(FloodInterface::class);
    $flood->isAllowed(self::FLOOD_NAME, 5, 60, $expected)->willReturn(TRUE);
    $flood->register(self::FLOOD_NAME, 60, $expected)->shouldBeCalledOnce();

    $output = $this->createNode($flood->reveal())->process($this->params(5, 60, $identifier));
    $this->assertFalse($output['throttled']);
  }

  /**
   * Data provider for identifiers that must be hashed.
   */
  public static function unsafeIdentifierProvider(): array {
    return [
      'over 64 chars' => ['https://example.com/very/long/path/that/exceeds/the/sixty-four-character/raw/key/limit'],
      'non-ASCII' => ['kunde-müller@example.com'],
    ];
  }

  /**
   * Short ASCII identifiers pass through unhashed (readable in flood table).
   */
  public function testShortAsciiIdentifierStaysRaw(): void {
    $flood = $this->prophesize(FloodInterface::class);
    $flood->isAllowed(self::FLOOD_NAME, 5, 60, 'vendor-api')->willReturn(TRUE);
    $flood->register(self::FLOOD_NAME, 60, 'vendor-api')->shouldBeCalledOnce();

    $output = $this->createNode($flood->reveal())->process($this->params(5, 60, 'vendor-api'));
    $this->assertFalse($output['throttled']);
  }

  /**
   * Validation rejects missing, non-numeric, and sub-1 limit/window.
   */
  public function testValidateParams(): void {
    $flood = $this->prophesize(FloodInterface::class);
    $node = $this->createNode($flood->reveal());

    $this->assertTrue($node->validateParams(['limit' => 10, 'window' => 3600])->isValid());
    $this->assertFalse($node->validateParams(['window' => 3600])->isValid());
    $this->assertFalse($node->validateParams(['limit' => 0, 'window' => 3600])->isValid());
    $this->assertFalse($node->validateParams(['limit' => 'many', 'window' => 3600])->isValid());
    $this->assertFalse($node->validateParams(['limit' => 10, 'window' => -1])->isValid());
  }

}
