<?php

declare(strict_types=1);

namespace Drupal\Tests\flowdrop_session\Kernel;

use Drupal\Tests\flowdrop\Kernel\WorkflowExecutionTestBase;
use Drupal\flowdrop_pipeline\Constants\PipelineStatus;
use Drupal\flowdrop_runtime\Event\RealTimeBroadcastEvent;
use Drupal\flowdrop_session\Constants\MessageRole;
use Drupal\flowdrop_session\Constants\MessageStatus;
use Drupal\flowdrop_session\Constants\SessionStatus;
use Drupal\flowdrop_session\DTO\TurnOptions;
use Drupal\flowdrop_session\DTO\TurnResult;
use Drupal\flowdrop_session\Entity\FlowDropSessionInterface;
use Drupal\flowdrop_session\Entity\FlowDropSessionMessageInterface;
use Drupal\flowdrop_session\Event\SessionTurnCompletedEvent;
use Drupal\flowdrop_session\Event\SessionTurnStartedEvent;
use Drupal\flowdrop_session\Exception\ConcurrentTurnException;
use Drupal\flowdrop_stategraph\Service\StateGraphOrchestrator;
use Drupal\flowdrop_workflow\WorkflowDefinitionInterface;

/**
 * Pins the chat-turn verb: executeTurn() on a session-bound workflow.
 *
 * The turn composes postMessage() with the existing session execution
 * bookkeeping — user message persisted, pipeline ran, assistant message
 * written back — and ends at any pipeline stop with the status saying
 * why. A second turn while the session is RUNNING is refused with a
 * typed exception.
 *
 * @group flowdrop_session
 */
class SessionTurnTest extends WorkflowExecutionTestBase {

  /**
   * Collected SessionTurnStartedEvent instances.
   *
   * @var array<int, \Drupal\flowdrop_session\Event\SessionTurnStartedEvent>
   */
  protected array $startedEvents = [];

  /**
   * Collected SessionTurnCompletedEvent instances.
   *
   * @var array<int, \Drupal\flowdrop_session\Event\SessionTurnCompletedEvent>
   */
  protected array $completedEvents = [];

  /**
   * Collected session-typed RealTimeBroadcastEvent instances.
   *
   * @var array<int, \Drupal\flowdrop_runtime\Event\RealTimeBroadcastEvent>
   */
  protected array $sessionBroadcasts = [];

  /**
   * Registers listeners collecting the turn events and their broadcasts.
   */
  protected function collectTurnEvents(): void {
    $dispatcher = $this->container->get('event_dispatcher');
    $dispatcher->addListener(SessionTurnStartedEvent::NAME, function (SessionTurnStartedEvent $event): void {
      $this->startedEvents[] = $event;
    });
    $dispatcher->addListener(SessionTurnCompletedEvent::NAME, function (SessionTurnCompletedEvent $event): void {
      $this->completedEvents[] = $event;
    });
    $dispatcher->addListener(RealTimeBroadcastEvent::NAME, function (RealTimeBroadcastEvent $event): void {
      if ($event->type === 'session') {
        $this->sessionBroadcasts[] = $event;
      }
    });
  }

  /**
   * Creates a chat-shaped StateGraph workflow: chat input → chat output.
   *
   * @param string $id
   *   The workflow entity id.
   *
   * @return \Drupal\flowdrop_workflow\WorkflowDefinitionInterface
   *   The saved workflow.
   */
  protected function createChatWorkflow(string $id): WorkflowDefinitionInterface {
    return $this->createStateGraphWorkflow($id, [
      $this->node('chat_input.1', 'chat_input', [], [
        'name' => 'Chat Input',
        'category' => 'input',
      ]),
      $this->node('chat_output.1', 'chat_output', [], [
        'name' => 'Chat Output',
        'category' => 'output',
      ]),
    ], [
      $this->edge('chat_input.1', 'message', 'chat_output.1', 'message'),
    ]);
  }

  /**
   * Creates a session bound to the given workflow.
   *
   * @param \Drupal\flowdrop_workflow\WorkflowDefinitionInterface $workflow
   *   The workflow.
   *
   * @return \Drupal\flowdrop_session\Entity\FlowDropSessionInterface
   *   The session.
   */
  protected function createSession(WorkflowDefinitionInterface $workflow): FlowDropSessionInterface {
    return $this->container->get('flowdrop_session.service')->createSession($workflow);
  }

  /**
   * Executes a turn via the turn service.
   *
   * @param \Drupal\flowdrop_session\Entity\FlowDropSessionInterface $session
   *   The session.
   * @param string $content
   *   The user message content.
   * @param \Drupal\flowdrop_session\DTO\TurnOptions|null $options
   *   The turn options.
   *
   * @return \Drupal\flowdrop_session\DTO\TurnResult
   *   The turn result.
   */
  protected function executeTurn(FlowDropSessionInterface $session, string $content, ?TurnOptions $options = NULL): TurnResult {
    return $this->container->get('flowdrop_session.turn_service')
      ->executeTurn((string) $session->id(), $content, $options ?? new TurnOptions());
  }

  /**
   * A waited turn runs the workflow and writes the assistant reply back.
   */
  public function testWaitedTurnCompletesWithAssistantWriteBack(): void {
    $workflow = $this->createChatWorkflow('turn_chat');
    $session = $this->createSession($workflow);
    $this->collectTurnEvents();

    $result = $this->executeTurn($session, 'Hello turn', new TurnOptions(wait: TRUE));

    $this->assertSame((string) $session->id(), $result->sessionId);
    $this->assertNotNull($result->pipelineId);

    $pipeline = $this->loadPipeline($result->pipelineId);
    $this->assertSame(TurnResult::STATUS_COMPLETED, $result->status, $this->jobTrail($pipeline));
    $this->assertSame(PipelineStatus::COMPLETED, $pipeline->getStatus());

    // User message persisted, terminal, and stamped with the execution.
    $messageStorage = $this->container->get('entity_type.manager')
      ->getStorage('flowdrop_session_message');
    $userMessage = $messageStorage->load($result->userMessageId);
    $this->assertInstanceOf(FlowDropSessionMessageInterface::class, $userMessage);
    $this->assertSame(MessageRole::USER, $userMessage->getRole());
    $this->assertSame('Hello turn', $userMessage->getContent());
    $this->assertSame(MessageStatus::COMPLETED, $userMessage->getStatus());
    $this->assertSame($result->pipelineId, $userMessage->getExecutionId());

    // Assistant write-back from the chat_output node, echoing the input.
    $this->assertNotEmpty($result->assistantMessageIds);
    $assistant = $messageStorage->load($result->assistantMessageIds[0]);
    $this->assertInstanceOf(FlowDropSessionMessageInterface::class, $assistant);
    $this->assertSame(MessageRole::ASSISTANT, $assistant->getRole());
    $this->assertStringContainsString('Hello turn', $assistant->getContent());

    // The turn released the session.
    $session = $this->reloadSession($session);
    $this->assertSame(SessionStatus::IDLE, $session->getStatus());

    // One started + one completed event, correlated by session/turn ids.
    $this->assertCount(1, $this->startedEvents);
    $started = $this->startedEvents[0];
    $this->assertSame($result->sessionId, $started->sessionId);
    $this->assertSame($result->userMessageId, $started->turnId);
    $this->assertSame($result->pipelineId, $started->pipelineId);
    $this->assertSame((string) $workflow->id(), $started->workflowId);

    $this->assertCount(1, $this->completedEvents);
    $completed = $this->completedEvents[0];
    $this->assertSame($result->sessionId, $completed->sessionId);
    $this->assertSame($result->userMessageId, $completed->turnId);
    $this->assertSame($result->pipelineId, $completed->pipelineId);
    $this->assertSame(SessionTurnCompletedEvent::STATUS_COMPLETED, $completed->status);
    $this->assertSame($result->assistantMessageIds, $completed->assistantMessageIds);
    $this->assertNull($completed->pausedReason);

    // The RealTime bridge re-keys both onto the broadcast channel.
    $this->assertCount(2, $this->sessionBroadcasts);
    [$startBroadcast, $completeBroadcast] = $this->sessionBroadcasts;
    $this->assertSame('turn_started', $startBroadcast->event);
    $this->assertSame($result->sessionId, $startBroadcast->executionId);
    $this->assertSame($result->userMessageId, $startBroadcast->data['turn_id']);
    $this->assertSame('turn_completed', $completeBroadcast->event);
    $this->assertSame($result->sessionId, $completeBroadcast->executionId);
    $this->assertSame(SessionTurnCompletedEvent::STATUS_COMPLETED, $completeBroadcast->data['status']);
    $this->assertSame($result->assistantMessageIds, $completeBroadcast->data['assistant_message_ids']);
  }

  /**
   * A second turn while the session is RUNNING is refused with 409 shape.
   */
  public function testConcurrentTurnIsRejected(): void {
    $workflow = $this->createChatWorkflow('turn_conflict');
    $session = $this->createSession($workflow);
    $session->setStatus(SessionStatus::RUNNING)->save();

    $this->expectException(ConcurrentTurnException::class);
    $this->executeTurn($session, 'Second turn');
  }

  /**
   * An unwaited turn schedules deferred processing and returns running.
   *
   * The pipelineId is NULL by design: the pipeline entity is created after
   * the HTTP response; the session turn events carry it once started.
   */
  public function testUnwaitedTurnSchedulesDeferredProcessing(): void {
    $workflow = $this->createChatWorkflow('turn_deferred');
    $session = $this->createSession($workflow);

    $result = $this->executeTurn($session, 'Deferred turn');

    $this->assertSame(TurnResult::STATUS_RUNNING, $result->status);
    $this->assertNull($result->pipelineId);
    $this->assertSame([], $result->assistantMessageIds);

    $session = $this->reloadSession($session);
    $this->assertSame(SessionStatus::RUNNING, $session->getStatus());
    $this->assertTrue(
      $this->container->get('flowdrop_session.deferred_message_processor')->hasPendingMessages(),
      'The turn was handed to the deferred scheduler.',
    );
  }

  /**
   * A turn ends at an input pause; the input request is its reply.
   *
   * AWAITING_INPUT does not gate the next turn — only RUNNING does
   * (pinned by the absence of ConcurrentTurnException in the guard).
   */
  public function testTurnEndsAtInputPause(): void {
    $workflow = $this->createStateGraphWorkflow('turn_hitl', [
      $this->node('chat_input.1', 'chat_input', [], [
        'name' => 'Chat Input',
        'category' => 'input',
      ]),
      $this->confirmationNode('confirm'),
      $this->node('chat_output.1', 'chat_output', [], [
        'name' => 'Chat Output',
        'category' => 'output',
      ]),
    ], [
      ['id' => 'e1', 'source' => 'chat_input.1', 'target' => 'confirm'],
      ['id' => 'e2', 'source' => 'confirm', 'target' => 'chat_output.1'],
    ]);
    $session = $this->createSession($workflow);
    $this->collectTurnEvents();

    $result = $this->executeTurn($session, 'Needs approval', new TurnOptions(wait: TRUE));

    $this->assertSame(TurnResult::STATUS_AWAITING_INPUT, $result->status);
    $session = $this->reloadSession($session);
    $this->assertSame(SessionStatus::AWAITING_INPUT, $session->getStatus());

    // The input pause ends the turn: completed event with awaiting_input.
    $this->assertCount(1, $this->completedEvents);
    $completed = $this->completedEvents[0];
    $this->assertSame($result->userMessageId, $completed->turnId);
    $this->assertSame(SessionTurnCompletedEvent::STATUS_AWAITING_INPUT, $completed->status);
    $this->assertNull($completed->pausedReason);
  }

  /**
   * A HITL turn runs on the entity checkpointer so the resume keeps state.
   *
   * Sessions resume HITL interrupts in a later request, where
   * memory-checkpointer state no longer exists — the resumed graph would
   * restart with an empty state channel. The session execution path must
   * therefore default to the persistent (entity) checkpointer, even when
   * the workflow metadata carries the StateGraph form's persisted
   * `memory` default (which createStateGraphWorkflow writes, mirroring
   * every workflow saved through the plugin form).
   */
  public function testHitlTurnPersistsEntityCheckpointsForResume(): void {
    $workflow = $this->createStateGraphWorkflow('turn_hitl_checkpoint', [
      $this->node('chat_input.1', 'chat_input', [], [
        'name' => 'Chat Input',
        'category' => 'input',
      ]),
      $this->confirmationNode('confirm'),
      $this->node('chat_output.1', 'chat_output', [], [
        'name' => 'Chat Output',
        'category' => 'output',
      ]),
    ], [
      ['id' => 'e1', 'source' => 'chat_input.1', 'target' => 'confirm'],
      ['id' => 'e2', 'source' => 'confirm', 'target' => 'chat_output.1'],
    ]);
    $session = $this->createSession($workflow);

    $result = $this->executeTurn($session, 'Needs approval', new TurnOptions(wait: TRUE));

    $this->assertSame(TurnResult::STATUS_AWAITING_INPUT, $result->status);
    $this->assertNotNull($result->pipelineId);
    $pipeline = $this->loadPipeline($result->pipelineId);

    // The stamped config is what an interrupt resume rebuilds in the later
    // request — it must say entity, the workflow's nested `memory` form
    // default notwithstanding.
    $config = $pipeline->getExecutionContext()[StateGraphOrchestrator::CONTEXT_STATEGRAPH_CONFIG] ?? NULL;
    $this->assertIsArray($config);
    $this->assertSame('entity', $config['checkpointerType']);
    $thread = "playground_session_{$session->id()}";
    $this->assertSame($thread, $config['threadId']);

    // The pre-pause run wrote persistent checkpoints on the session thread.
    $checkpointsBeforeResume = count($this->checkpointsForThread($thread));
    $this->assertGreaterThan(0, $checkpointsBeforeResume,
      'The pre-pause run persisted entity checkpoints for the session thread.');

    // Resolving the HITL interrupt resumes the pipeline through the
    // standard InterruptResolvedSubscriber chain.
    $interrupts = $this->container->get('entity_type.manager')
      ->getStorage('flowdrop_interrupt')
      ->loadByProperties(['pipeline_id' => (string) $pipeline->id()]);
    $this->assertCount(1, $interrupts);
    $this->container->get('flowdrop_interrupt.manager')
      ->resolveInterrupt((string) reset($interrupts)->uuid(), TRUE);

    $pipeline = $this->loadPipeline($result->pipelineId);
    $this->assertSame(PipelineStatus::COMPLETED, $pipeline->getStatus(), $this->jobTrail($pipeline));

    // The resume advanced the same thread's persistent checkpoint trail —
    // it restored from and kept writing entity checkpoints.
    $this->assertGreaterThan(
      $checkpointsBeforeResume,
      count($this->checkpointsForThread($thread)),
      'The resume wrote further checkpoints on the session thread.',
    );
  }

  /**
   * Loads the persistent checkpoint entities for a thread.
   *
   * @param string $threadId
   *   The thread id.
   *
   * @return array<int|string, \Drupal\Core\Entity\EntityInterface>
   *   The checkpoint entities.
   */
  protected function checkpointsForThread(string $threadId): array {
    $storage = $this->container->get('entity_type.manager')->getStorage('state_checkpoint');
    $storage->resetCache();
    return $storage->loadByProperties(['thread_id' => $threadId]);
  }

  /**
   * Reloads a session fresh from storage.
   *
   * @param \Drupal\flowdrop_session\Entity\FlowDropSessionInterface $session
   *   The session.
   *
   * @return \Drupal\flowdrop_session\Entity\FlowDropSessionInterface
   *   The reloaded session.
   */
  protected function reloadSession(FlowDropSessionInterface $session): FlowDropSessionInterface {
    $storage = $this->container->get('entity_type.manager')->getStorage('flowdrop_session');
    $storage->resetCache([$session->id()]);
    $reloaded = $storage->load($session->id());
    $this->assertInstanceOf(FlowDropSessionInterface::class, $reloaded);
    return $reloaded;
  }

}
