<?php

declare(strict_types=1);

namespace Drupal\flowdrop_session\Service;

use Drupal\Core\Entity\EntityTypeManagerInterface;
use Drupal\Core\Queue\QueueFactory;
use Drupal\flowdrop_pipeline\Constants\PipelineStatus;
use Drupal\flowdrop_pipeline\Entity\FlowDropPipelineInterface;
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\Exception\ConcurrentTurnException;
use Psr\Log\LoggerInterface;

/**
 * Default implementation of the chat-turn verb.
 *
 * Composition only: postMessage() appends the user message, then the turn
 * is scheduled — deferred-sync after the HTTP response, or the session
 * message queue, per the session's execution mode.
 * The execution half is typed as MessageProcessorInterface so it stays
 * swappable when the self-sufficient chat_input (Concept 3) shrinks the
 * turn to post + trigger.
 */
class SessionTurnService implements SessionTurnServiceInterface {

  /**
   * Constructs a new SessionTurnService.
   *
   * @param \Drupal\Core\Entity\EntityTypeManagerInterface $entityTypeManager
   *   The entity type manager.
   * @param \Drupal\flowdrop_session\Service\SessionService $sessionService
   *   The session service (message/session CRUD).
   * @param \Drupal\flowdrop_session\Service\MessageProcessorInterface $messageProcessor
   *   The execution half — processes a posted message (wait path only;
   *   async paths reach it via the scheduler / queue worker).
   * @param \Drupal\flowdrop_session\Service\DeferredSchedulerInterface $deferredScheduler
   *   The deferred (post-response) scheduler.
   * @param \Drupal\Core\Queue\QueueFactory $queueFactory
   *   The queue factory, for sessions in async execution mode.
   * @param \Psr\Log\LoggerInterface $logger
   *   The flowdrop_session logger channel.
   */
  public function __construct(
    protected readonly EntityTypeManagerInterface $entityTypeManager,
    protected readonly SessionService $sessionService,
    protected readonly MessageProcessorInterface $messageProcessor,
    protected readonly DeferredSchedulerInterface $deferredScheduler,
    protected readonly QueueFactory $queueFactory,
    protected readonly LoggerInterface $logger,
  ) {}

  /**
   * {@inheritdoc}
   */
  public function executeTurn(string $sessionId, string $content, TurnOptions $options = new TurnOptions()): TurnResult {
    $session = $this->entityTypeManager
      ->getStorage('flowdrop_session')
      ->load($sessionId);
    if (!$session instanceof FlowDropSessionInterface) {
      throw new \InvalidArgumentException(sprintf('Session %s does not exist.', $sessionId));
    }
    if ($session->getWorkflow() === NULL) {
      throw new \RuntimeException(sprintf('Session %s has no associated workflow.', $sessionId));
    }

    // A turn is a serialized unit: refuse a second one while the session
    // is RUNNING. AWAITING_INPUT deliberately passes — the next user
    // message is the resume and starts a new turn.
    if ($session->getStatus() === SessionStatus::RUNNING) {
      throw new ConcurrentTurnException((string) $session->id());
    }

    $metadata = $options->inputs === [] ? [] : ['inputs' => $options->inputs];
    $message = $this->sessionService->postMessage($session, $content, MessageRole::USER, $metadata);

    if ($options->wait) {
      // Inline processing, same entry the deferred/queue paths use.
      // Failures propagate to the caller; they are not folded into the
      // result (the session already carries the error bubble).
      $this->messageProcessor->processMessage($message);
      return $this->buildWaitResult($session, $message);
    }

    if ($session->getExecutionMode() === 'async') {
      // Queue path: PENDING marks the message as not yet processed for the
      // queue worker. Item shape mirrors the flowdrop_session_message
      // worker's expectations (see SessionService::queueMessageProcessing).
      $message->setStatus(MessageStatus::PENDING)->save();
      $this->queueFactory->get('flowdrop_session_message')->createItem([
        'message_id' => $message->id(),
        'session_id' => $message->getSessionId(),
        'sequence_number' => $message->getSequenceNumber(),
      ]);
      $this->logger->info('Queued turn @message for session @session', [
        '@message' => $message->id(),
        '@session' => $session->id(),
      ]);
      return new TurnResult((string) $session->id(), (string) $message->id(), NULL, TurnResult::STATUS_QUEUED);
    }

    // Deferred-sync path: flip the session RUNNING now (triggers frontend
    // polling / gates the next turn) and process after the HTTP response.
    $session->setStatus(SessionStatus::RUNNING)->save();
    $this->deferredScheduler->schedule($message);
    return new TurnResult((string) $session->id(), (string) $message->id(), NULL, TurnResult::STATUS_RUNNING);
  }

  /**
   * Builds the final result for a waited turn.
   *
   * @param \Drupal\flowdrop_session\Entity\FlowDropSessionInterface $session
   *   The session (mutated in place by the execution service).
   * @param \Drupal\flowdrop_session\Entity\FlowDropSessionMessageInterface $message
   *   The user message (execution id stamped during processing).
   *
   * @return \Drupal\flowdrop_session\DTO\TurnResult
   *   The result with the stop status and assistant write-backs.
   */
  protected function buildWaitResult(
    FlowDropSessionInterface $session,
    FlowDropSessionMessageInterface $message,
  ): TurnResult {
    // The execution service transitions the session on its own loaded
    // instance; re-read persisted state rather than this method's copy.
    $fresh = $this->entityTypeManager
      ->getStorage('flowdrop_session')
      ->loadUnchanged($session->id());
    if ($fresh instanceof FlowDropSessionInterface) {
      $session = $fresh;
    }
    $pipelineId = $message->getExecutionId();

    // The turn ends at any pipeline stop; the status says why.
    $status = match (TRUE) {
      $session->getStatus() === SessionStatus::AWAITING_INPUT => TurnResult::STATUS_AWAITING_INPUT,
      $this->pipelineIsPaused($pipelineId) => TurnResult::STATUS_PAUSED,
      $message->getStatus() === MessageStatus::FAILED => TurnResult::STATUS_FAILED,
      default => TurnResult::STATUS_COMPLETED,
    };

    return new TurnResult(
      (string) $session->id(),
      (string) $message->id(),
      $pipelineId,
      $status,
      $pipelineId !== NULL
        ? $this->sessionService->getAssistantMessageIdsForExecution((string) $session->id(), $pipelineId)
        : [],
    );
  }

  /**
   * Checks whether the turn's pipeline stopped in a paused state.
   *
   * @param string|null $pipelineId
   *   The pipeline id, if one was created.
   *
   * @return bool
   *   TRUE when the pipeline exists and is paused (e.g. budget pause).
   */
  protected function pipelineIsPaused(?string $pipelineId): bool {
    if ($pipelineId === NULL || $pipelineId === '') {
      return FALSE;
    }
    $pipeline = $this->entityTypeManager
      ->getStorage('flowdrop_pipeline')
      ->load($pipelineId);
    return $pipeline instanceof FlowDropPipelineInterface
      && $pipeline->getStatus() === PipelineStatus::PAUSED;
  }

}
